MCP Gateway 团队权限与审计配置实战（含 Claude Code 接入完整代码）
·
团队多人共用 MCP Server 时，最先出现的问题不是功能，而是权限边界和成本归因。本文直接从配置层面讲 MCP Gateway 的治理实现，涵盖权限路由、审计日志结构、用量限额策略和 Claude Code 接入的标准配置，代码可直接参考。
架构概览
┌─────────────────────────────┐
│ Claude Code Clients (多人) │
│ Client A / B / C │
└────────────┬────────────────┘
│ HTTP/SSE (MCP Protocol)
▼
┌─────────────────────────────┐
│ MCP Gateway │
│ ┌─────────────────────┐ │
│ │ 认证 & 权限路由 │ │
│ │ 审计日志记录 │ │
│ │ 用量限额检查 │ │
│ └─────────────────────┘ │
└────────────┬────────────────┘
│
┌─────────┼─────────┐
▼ ▼ ▼
GitHub-MCP DB-MCP Docs-MCP
环境准备
前置条件：
- Python 3.9+，已安装
anthropicSDK（pip install anthropic>=0.40.0） - 已有 API Key（兼容 Anthropic SDK 格式，
base_url指向https://gw.claudeapi.com） - 已部署或本地运行 MCP Gateway 实例（监听
http://localhost:8080） - Claude Code CLI 已安装（具体版本要求以 Anthropic 官方文档为准）
# 安装依赖
pip install anthropic>=0.40.0
# 配置环境变量（写入 .env 或 team .env.template）
export ANTHROPIC_BASE_URL="https://gw.claudeapi.com"
export ANTHROPIC_API_KEY="your-claudeapi-key"
export MCP_GATEWAY_URL="http://localhost:8080"
export MCP_GATEWAY_TOKEN="your-member-token"
第一部分：权限路由配置
Gateway 路由规则（YAML）
# mcp-gateway-config.yaml
# 说明：此为说明性配置示例，具体字段以所用 Gateway 实现的文档为准
server:
listen: ":8080"
auth:
type: "bearer_token" # 通过 Authorization: Bearer <token> 识别成员身份
members:
- id: "admin"
token: "${ADMIN_TOKEN}"
role: "admin"
- id: "engineer_alice"
token: "${ALICE_TOKEN}"
role: "developer"
- id: "intern_bob"
token: "${BOB_TOKEN}"
role: "restricted"
routes:
- role: "admin"
allow_servers: ["*"] # 管理员可访问所有 Server
rate_limit:
requests_per_minute: 200
- role: "developer"
allow_servers:
- "github-mcp"
- "docs-search-mcp"
- "ci-runner-mcp"
deny_servers:
- "production-db-mcp"
rate_limit:
requests_per_minute: 60
tokens_per_day: 500000 # 成员级每日 Token 上限
- role: "restricted"
allow_servers:
- "docs-search-mcp" # 只读工具
deny_servers:
- "*"
rate_limit:
requests_per_minute: 20
tokens_per_day: 100000
audit:
enabled: true
output: "file"
path: "/var/log/mcp-gateway/audit.jsonl"
redact_fields: # 脱敏字段，不记录完整 payload
- "input.query"
- "input.sql"
- "input.content"
retain_days: 90
验证权限规则是否生效
# 用开发者 Token 尝试访问被拒绝的 production-db-mcp，预期返回 403
curl -X POST http://localhost:8080/mcp/production-db-mcp/call \
-H "Authorization: Bearer ${ALICE_TOKEN}" \
-H "Content-Type: application/json" \
-d '{"tool": "query", "input": {"sql": "SELECT 1"}}' \
-w "\nHTTP Status: %{http_code}\n"
# 预期输出：
# {"error": "forbidden", "message": "server 'production-db-mcp' not allowed for role 'developer'"}
# HTTP Status: 403
第二部分：审计日志结构与 Python 解析
标准审计日志格式（JSONL）
每行一条记录，字段说明如下：
{
"event_type": "mcp_tool_call",
"timestamp": "2026-07-02T09:12:34.123Z",
"trace_id": "req_abc123xyz",
"member_id": "engineer_alice",
"member_role": "developer",
"mcp_server": "github-mcp",
"tool_name": "create_pull_request",
"input_summary": "repo=acme/api branch=feat/auth-refactor", // 脱敏后摘要
"output_status": "success",
"tokens_input": 890,
"tokens_output": 350,
"tokens_total": 1240,
"model": "claude-sonnet-4-6",
"latency_ms": 890,
"gateway_action": "forwarded" // forwarded | blocked | rate_limited
}
Python：解析审计日志 + 用量统计
# audit_analyzer.py
import json
from pathlib import Path
from collections import defaultdict
from datetime import datetime, timezone
def parse_audit_log(log_path: str) -> list[dict]:
"""解析 JSONL 格式的 MCP Gateway 审计日志"""
records = []
with open(log_path, "r", encoding="utf-8") as f:
for line_num, line in enumerate(f, 1):
line = line.strip()
if not line:
continue
try:
records.append(json.loads(line))
except json.JSONDecodeError as e:
print(f"[WARN] 第 {line_num} 行解析失败: {e}")
return records
def summarize_usage(records: list[dict]) -> dict:
"""
按成员汇总 Token 用量
返回: {member_id: {tokens_total, call_count, blocked_count}}
"""
summary = defaultdict(lambda: {
"tokens_total": 0,
"call_count": 0,
"blocked_count": 0,
"tools_used": set()
})
for r in records:
member = r.get("member_id", "unknown")
action = r.get("gateway_action", "forwarded")
if action == "blocked" or action == "rate_limited":
summary[member]["blocked_count"] += 1
continue
summary[member]["tokens_total"] += r.get("tokens_total", 0)
summary[member]["call_count"] += 1
tool = f"{r.get('mcp_server')}/{r.get('tool_name')}"
summary[member]["tools_used"].add(tool)
# set 转 list，便于序列化
for member in summary:
summary[member]["tools_used"] = sorted(summary[member]["tools_used"])
return dict(summary)
def detect_anomalies(records: list[dict], threshold_per_5min: int = 100) -> list[dict]:
"""
检测短时间内高频调用异常
threshold_per_5min: 同一成员 5 分钟内调用次数阈值
"""
from collections import deque
anomalies = []
# 按成员分组，时间窗口检测
member_timestamps: dict[str, deque] = defaultdict(deque)
for r in sorted(records, key=lambda x: x.get("timestamp", "")):
if r.get("gateway_action") == "blocked":
continue
member = r.get("member_id", "unknown")
ts_str = r.get("timestamp", "")
try:
ts = datetime.fromisoformat(ts_str.replace("Z", "+00:00"))
except ValueError:
continue
dq = member_timestamps[member]
dq.append(ts)
# 滑动窗口：移除 5 分钟前的记录
cutoff = ts.timestamp() - 300
while dq and dq[0].timestamp() < cutoff:
dq.popleft()
if len(dq) >= threshold_per_5min:
anomalies.append({
"member_id": member,
"window_end": ts_str,
"calls_in_5min": len(dq),
"threshold": threshold_per_5min
})
return anomalies
if __name__ == "__main__":
log_file = "/var/log/mcp-gateway/audit.jsonl"
if not Path(log_file).exists():
print(f"日志文件不存在: {log_file}")
exit(1)
records = parse_audit_log(log_file)
print(f"共解析 {len(records)} 条审计记录\n")
# 用量汇总
usage = summarize_usage(records)
print("=== 成员用量汇总 ===")
for member, stats in sorted(usage.items(), key=lambda x: -x[1]["tokens_total"]):
print(f" {member}: {stats['tokens_total']:,} tokens | "
f"{stats['call_count']} calls | "
f"{stats['blocked_count']} blocked")
# 异常检测
anomalies = detect_anomalies(records)
if anomalies:
print(f"\n=== 检测到 {len(anomalies)} 处高频调用异常 ===")
for a in anomalies:
print(f" [{a['window_end']}] {a['member_id']}: "
f"{a['calls_in_5min']} calls in 5 min (threshold: {a['threshold']})")
else:
print("\n未检测到高频调用异常。")
第三部分：Claude Code 接入配置
Step 1：Claude Code 环境变量配置
# 写入项目根目录 .env（加入 .gitignore，不要提交）
ANTHROPIC_BASE_URL=https://gw.claudeapi.com
ANTHROPIC_API_KEY=your-claudeapi-key
# MCP Gateway 认证（与 Claude API Key 独立，不要混用）
MCP_GATEWAY_URL=http://your-gateway-host:8080
MCP_GATEWAY_TOKEN=your-member-token
Step 2：Claude Code MCP Server 配置
截至 2026-07-03，Claude Code 官方文档推荐使用 claude mcp add 添加 MCP Server。团队共享配置建议使用 project scope，使配置进入项目根目录 .mcp.json，密钥通过环境变量展开。
claude mcp add --transport http gateway --scope project \
"http://localhost:8080/mcp" \
--header "Authorization: Bearer ${MCP_GATEWAY_TOKEN}"
.mcp.json 示例：
{
"mcpServers": {
"gateway": {
"type": "http",
"url": "${MCP_GATEWAY_URL:-http://localhost:8080}/mcp",
"headers": {
"Authorization": "Bearer ${MCP_GATEWAY_TOKEN}"
}
}
}
}
如果使用本地 stdio 方式包装 Gateway client，也可以继续使用
command/args/env字段；但团队治理场景更推荐统一 HTTP endpoint，便于集中鉴权和审计。
Step 3：验证 MCP 配置是否加载
# 查看已注册 MCP Server
claude mcp list
# 查看 gateway 详情
claude mcp get gateway
如果是 project scope，首次进入项目时 Claude Code 可能要求确认是否信任 .mcp.json 中的 Server。团队可以在 README 或 onboarding 文档中说明这个审批动作，避免成员误以为 Gateway 未生效。
Step 4：Python SDK 验证工具调用链路
# verify_claude_mcp_chain.py
import os
import anthropic
def verify_mcp_chain():
"""
验证 Claude Code → MCP Gateway → MCP Server 的完整链路
通过 Tool Use API 模拟一次工具调用
"""
client = anthropic.Anthropic(
api_key=os.environ["ANTHROPIC_API_KEY"],
base_url=os.environ.get("ANTHROPIC_BASE_URL", "https://gw.claudeapi.com")
)
# 使用轻量模型做验证，降低验证成本
model = "claude-haiku-4-5-20251001"
tools = [
{
"name": "check_gateway_health",
"description": "检查 MCP Gateway 的连通性和当前成员的可用工具列表",
"input_schema": {
"type": "object",
"properties": {
"member_id": {
"type": "string",
"description": "当前成员 ID"
}
},
"required": ["member_id"]
}
}
]
try:
response = client.messages.create(
model=model,
max_tokens=512,
tools=tools,
messages=[
{
"role": "user",
"content": "请调用 check_gateway_health 工具，member_id 填 'test-verify'，确认 Gateway 链路正常。"
}
]
)
print(f"[OK] 模型响应正常，stop_reason: {response.stop_reason}")
print(f"[OK] 使用 Token: input={response.usage.input_tokens}, "
f"output={response.usage.output_tokens}")
for block in response.content:
if block.type == "tool_use":
print(f"[OK] 工具调用触发: {block.name}, input={block.input}")
elif block.type == "text":
print(f"[INFO] 模型文本: {block.text[:200]}")
except anthropic.AuthenticationError as e:
print(f"[ERROR] 认证失败，检查 ANTHROPIC_API_KEY 是否正确: {e}")
except anthropic.APIConnectionError as e:
print(f"[ERROR] 连接失败，检查 ANTHROPIC_BASE_URL 和网络: {e}")
except anthropic.RateLimitError as e:
print(f"[ERROR] 用量限额触发 (429)，检查成员级或团队级限额设置: {e}")
except Exception as e:
print(f"[ERROR] 未预期错误: {type(e).__name__}: {e}")
if __name__ == "__main__":
verify_mcp_chain()
第四部分：用量限额中间件示例（Python）
以下是一个轻量级限额检查中间件的参考实现，适合在自建 Gateway 中集成：
# quota_middleware.py
import time
import threading
from collections import defaultdict
from dataclasses import dataclass, field
from typing import Optional
@dataclass
class QuotaConfig:
"""三级限额配置"""
member_daily_tokens: int = 500_000 # 成员每日 Token 上限
project_monthly_tokens: int = 2_000_000 # 项目每月 Token 上限
team_monthly_tokens: int = 10_000_000 # 团队每月 Token 总上限
alert_threshold: float = 0.8 # 预警阈值（80%）
@dataclass
class UsageCounter:
tokens_used: int = 0
reset_at: float = field(default_factory=time.time)
class QuotaMiddleware:
"""
三级 Token 用量限额检查
说明：此为简化示例，生产环境建议使用 Redis 等分布式存储
"""
def __init__(self, config: QuotaConfig):
self.config = config
self._lock = threading.Lock()
# member_id -> UsageCounter (每日重置)
self._member_usage: dict[str, UsageCounter] = defaultdict(UsageCounter)
# project_id -> UsageCounter (每月重置)
self._project_usage: dict[str, UsageCounter] = defaultdict(UsageCounter)
# 团队总量 (每月重置)
self._team_usage = UsageCounter()
def _is_new_day(self, counter: UsageCounter) -> bool:
now = time.time()
return (now - counter.reset_at) >= 86400 # 86400s = 1 day
def _is_new_month(self, counter: UsageCounter) -> bool:
now = time.time()
return (now - counter.reset_at) >= 2592000 # 30 days
def check_and_record(
self,
member_id: str,
project_id: str,
tokens_to_use: int
) -> tuple[bool, Optional[str]]:
"""
检查是否超出限额，并在允许时记录用量
返回: (allowed: bool, reason: Optional[str])
"""
with self._lock:
member_c = self._member_usage[member_id]
project_c = self._project_usage[project_id]
team_c = self._team_usage
# 重置过期计数器
if self._is_new_day(member_c):
member_c.tokens_used = 0
member_c.reset_at = time.time()
if self._is_new_month(project_c):
project_c.tokens_used = 0
project_c.reset_at = time.time()
if self._is_new_month(team_c):
team_c.tokens_used = 0
team_c.reset_at = time.time()
# 检查各级限额
if member_c.tokens_used + tokens_to_use > self.config.member_daily_tokens:
return False, f"member '{member_id}' daily quota exceeded"
if project_c.tokens_used + tokens_to_use > self.config.project_monthly_tokens:
return False, f"project '{project_id}' monthly quota exceeded"
if team_c.tokens_used + tokens_to_use > self.config.team_monthly_tokens:
return False, "team monthly quota exceeded"
# 通过检查，记录用量
member_c.tokens_used += tokens_to_use
project_c.tokens_used += tokens_to_use
team_c.tokens_used += tokens_to_use
# 预警检查
alerts = []
if member_c.tokens_used / self.config.member_daily_tokens >= self.config.alert_threshold:
alerts.append(
f"[ALERT] member '{member_id}' daily usage at "
f"{member_c.tokens_used/self.config.member_daily_tokens:.0%}"
)
if team_c.tokens_used / self.config.team_monthly_tokens >= self.config.alert_threshold:
alerts.append(
f"[ALERT] team monthly usage at "
f"{team_c.tokens_used/self.config.team_monthly_tokens:.0%}"
)
for alert in alerts:
print(alert) # 生产环境替换为 Slack/邮件通知
return True, None
# 使用示例
if __name__ == "__main__":
config = QuotaConfig(
member_daily_tokens=500_000,
project_monthly_tokens=2_000_000,
team_monthly_tokens=10_000_000,
alert_threshold=0.8
)
quota = QuotaMiddleware(config)
# 模拟一次请求
allowed, reason = quota.check_and_record(
member_id="engineer_alice",
project_id="proj_backend_v2",
tokens_to_use=1240
)
if allowed:
print("[OK] 请求通过限额检查，转发到 MCP Server")
else:
print(f"[BLOCKED] 请求被限额拦截: {reason}")
常见报错排查
| 错误码 / 现象 | 原因 | 排查步骤 |
|---|---|---|
403 Forbidden |
Gateway 权限规则拒绝当前 Token | 检查成员 Token 对应的 role，确认 allow_servers 包含目标 Server |
429 Too Many Requests |
成员/项目/团队限额触发 | 查看 Gateway 日志确认是哪级限额，调整配置或等待重置 |
AuthenticationError |
ANTHROPIC_API_KEY 未设置或无效 |
确认环境变量已 export，Key 格式正确 |
APIConnectionError |
ANTHROPIC_BASE_URL 配置错误或网络问题 |
确认 base_url 为 https://gw.claudeapi.com，无尾部斜杠 |
| 工具调用静默失败，返回空 content | MCP Server 名称与 Gateway 配置不匹配 | 检查 .mcp.json / ~/.claude.json 中 Server 名称与路由规则完全一致（区分大小写） |
| Gateway Token 与 API Key 混用报错 | 两套认证体系字段填错位置 | ANTHROPIC_API_KEY 填 ClaudeAPI Key，MCP_GATEWAY_TOKEN 填 Gateway 成员凭据，分开存储 |
小结
MCP Gateway 治理的工程实现核心是三个配置：
- 路由规则 YAML：按成员 role 声明 allow/deny Server 列表
- 审计日志 JSONL：每条记录包含身份、工具、摘要、Token、状态，入参脱敏
- 三级限额中间件：成员日限 → 项目月限 → 团队月限，80% 触发预警
Claude Code 接入侧通常只需两个模型调用环境变量（ANTHROPIC_BASE_URL + ANTHROPIC_API_KEY）和一份 MCP 配置（推荐项目级 .mcp.json）。Gateway Token 与 ClaudeAPI Key 分开存储，治理逻辑对客户端透明。
参考资料
- Anthropic MCP 协议文档：https://docs.anthropic.com/（截至 2026-07-02）
- Anthropic Tool Use API 文档：https://docs.anthropic.com/en/docs/build-with-claude/tool-use
- Claude Code 配置说明：以当前安装版本的官方文档为准
ClaudeAPI 为独立第三方技术服务商。Claude、Claude Code、Anthropic 等名称归其各自权利方所有，本文仅基于公开信息进行技术解读，不代表 ClaudeAPI 与相关权利方存在官方授权、代理、经销、合作或背书关系。
本文的持续更新版本可查看：MCP Gateway 治理指南：团队权限、审计日志、用量限额与 Claude Code 接入规范
数据与事实声明
- MCP 协议发布时间（2024-11-25）：来源为 Anthropic 官方博客
https://www.anthropic.com/news/model-context-protocol - 模型 ID 及价格（claude-haiku-4-5-20251001 ¥1/¥5、claude-sonnet-4-6 ¥4/¥20、claude-opus-4-7 ¥20/¥100，每M Token 输入/输出）：来源为 ClaudeAPI 官方定价页，以控制台实际显示为准
- Gateway 路由规则 YAML、审计日志 JSONL 格式、限额中间件代码：均为说明性示例，非特定开源项目的完整实现，生产环境请结合实际 Gateway 文档调整
- Claude Code MCP 配置 scope、
.mcp.json、~/.claude.json、环境变量展开：来源为 Claude Code MCP 官方文档https://code.claude.com/docs/en/mcp，具体以当前安装版本文档为准。
欢迎加入 MCP 技术社区！与志同道合者携手前行，一同解锁 MCP 技术的无限可能！
更多推荐
2
0- 0
已为社区贡献3条内容
所有评论(0)