前阵子和几个做 AI 应用的朋友聊天大家不约而同提到了同一个焦虑Agent 越来越能干了但谁也不敢保证它调用的那些工具是干净的。这个焦虑不是杞人忧天——当 MCPModel Context Protocol把大模型和外部工具连接起来之后整个链路的安全边界其实已经悄悄转移了。过去我们防的是“提示词注入”现在要防的是“工具层投毒”。今天我想分享一个我自己在做的项目用 Python 自建一个 MCP 安全网关专门用来检测工具投毒、Rug Pull 和认证绕过这几类典型的工具层攻击。这个网关不是概念验证是真的跑在业务环境里扛过并发请求的下面把我踩过的坑、想明白的设计和可以直接抄的代码都摊开讲。1. 为什么 AI Agent 的工具层成了新的攻击面先说结论模型参数不再是攻击者唯一的目标了工具注册表、工具描述、参数校验逻辑、以及工具返回的数据流每一个环节都可能成为突破口。传统 Web 安全关注的是 HTTP 接口、数据库注入、越权访问这些攻击面是“静态”的边界清晰。但 AI Agent 不一样它有一个工具层这个工具层是动态的、半开放的。Agent 会读取工具的描述来决定“什么时候用什么工具”会把工具返回的结果当作“事实”来继续推理还会基于这些结果产生新的调用动作。这意味着攻击者不需要直接攻破你的大模型只要污染工具层的某一个环节就能让整个 Agent 的行为跑偏。工具投毒Tool Poisoning攻击者修改工具的名称、描述、参数 schema让模型在运行时错误地选择工具或者把工具调用引导到恶意实现上。Rug Pull这个词借用了加密圈的“卷款跑路”。在 Agent 场景里指的是某个工具在初期正常、后期突然改变行为——比如某个“汇率查询工具”正常用了两周突然开始返回被篡改的数值导致下游交易逻辑做出错误决策。认证绕过Auth BypassMCP 服务器对工具调用的认证不严攻击者可以直接伪造客户端请求调用本不该暴露的管理型工具。我见过最典型的案例是这样的一个供应链 Agent 接入了十几个 MCP 工具其中包括“查询库存”“创建采购单”“同步物流状态”。攻击者没有碰模型本身只是把“同步物流状态”工具的描述改成了“当订单金额大于某阈值时自动触发加急采购流程”。模型读了新描述在正常的业务对话中就被诱导着执行了额外的创建采购单动作。整个过程里模型没有任何“被入侵”的感觉它只是忠实执行了新描述。这就是工具层的可怕之处它不像传统漏洞那样有明确的“漏洞点”更像是一种环境污染。而网关的价值就是在模型和工具之间加一道“安检门”对每一个工具调用做策略检查、来源校验、行为审计。下面这几节就是我这套网关的设计思路和落地细节。2. 网关整体设计与方案选型2.1 为什么用 Python 写而不是上现成的商业方案选择 Python 有几个很实际的原因。第一MCP 生态里 Python SDK 最成熟官方提供的mcp包、FastAPI 的异步能力、Pydantic 的数据校验组合起来做协议解析非常顺。第二安全策略本质上就是规则引擎加数据流分析Python 做原型迭代最快。第三团队里做 AI 应用的人大概率会用 Python维护成本低。也许有人会说“商业 API 网关不是更成熟吗”但商业网关的核心能力在流量管理、限流、鉴权对“工具语义层”的攻击检测几乎是空白——它不理解 MCP 的 JSON-RPC 结构不理解什么是工具描述投毒更不理解工具返回数据里的异常模式。所以我倾向于用 Python 做一个轻量的、专为 MCP 语义定制的代理层把它插在 Agent 应用和工具服务器之间。2.2 网关的分层架构和工作原理我这里的网关用的是“透明代理 策略引擎”的组合分三层协议层负责解析 MCP 的 JSON-RPC 请求/响应识别tools/call、tools/list、resources/read等关键方法。策略层对解析出来的工具名、参数、上下文来源做规则匹配这里会产生三种决策放行、阻断、告警。审计层记录所有调用的指纹工具名、参数 hash、来源 IP、token 身份、响应摘要写入本地存储用于后续的 Rug Pull 行为比对。网关不修改原始请求的内容它只做检查、转发、拦截。这样设计的好处是兼容性好任何通过标准 MCP 协议通信的客户端和服务器都能接入不需要改业务代码。架构上最关键的一个决定是网关要同时支持HTTP 模式和STDIO 模式的 MCP 服务。HTTP 模式好理解Agent 客户端发 POST /mcp 请求网关转发到上游工具服务器。STDIO 模式麻烦一点工具服务器以子进程方式启动网关需要维护子进程生命周期并且把 JSON-RPC 消息通过 stdin/stdout 转发。我实现的网关把这两种模式统一抽象成一个“ToolConnector”接口底层逻辑共用策略层。2.3 关键技术选型一览FastAPI Uvicorn做 HTTP 网关入口异步性能足够单机扛几千个并发没问题对于“AI Agent 怎么扛并发”这个问题FastAPI 的 async 原生支持是关键。Pydantic定义 JSON-RPC 消息模型顺带做参数类型校验。mcp官方 Python SDK用来构造和解析 MCP 协议消息省去自己手写 JSON-RPC 解析的麻烦。SQLite 文件快照记录工具注册表快照和调用审计日志轻量不需要额外部署数据库。JWT / API Key 校验认证层支持两种方式优先 JWT兼容明文 Key。我实测下来这套技术栈在 16 核 32G 的机器上网关自身的 P99 延迟控制在 5ms 以内纯转发不落库的情况下。加上审计落库P99 会上升到 15ms 左右对 Agent 场景完全够用。3. 工具投毒检测从“名称”到“schema”的全面校验3.1 投毒攻击最常藏匿的位置工具投毒攻击的手段很“软”它不需要高深的漏洞利用技巧重点在于污染模型能读到的那部分“元数据”。我总结了三个最常被动手脚的地方工具名称与描述攻击者把get_stock_price的描述改成get_stock_price_with_trigger诱导模型在取价格时顺带执行另一个隐藏逻辑。参数 Schema把参数类型的约束放宽。比如原来user_id必须是数字投毒后 schema 不校验类型恶意调用就能传字符串注入。工具列表顺序某些 MCP 实现里模型会偏好选择第一个匹配工具攻击者把恶意工具排到合法工具前面让模型“自然”选错。网关上要做的是对这三类位置做固定校验。我实现了一个ToolPolicy规则模块加载一份白名单策略文件里面定义了每个工具预期的名称、描述关键词、参数 schema。网关在每次收到tools/list响应时会对工具列表做一次快照校验发现异常就告警并记录。3.2 核心代码工具列表校验与调用前拦截# gateway/tool_policy.py from typing import Any, Dict, List, Optional from pydantic import BaseModel, ValidationError class ToolSpec(BaseModel): name: str description_fingerprint: str # 预期描述的 hash用于比对 allowed_params: Dict[str, str] # 参数名 - 期望类型 required_params: List[str] class ToolPolicyEngine: def __init__(self, policy_file: str): self.policy: Dict[str, ToolSpec] self._load_policy(policy_file) def _load_policy(self, path: str) - Dict[str, ToolSpec]: import json with open(path, r, encodingutf-8) as f: raw json.load(f) return {name: ToolSpec(**spec) for name, spec in raw.items()} def check_tool_list(self, tools: List[Dict[str, Any]]) - List[str]: 对比 tools/list 返回的工具列表与策略基线。 返回异常原因列表空列表表示通过。 anomalies [] remote_names {t.get(name) for t in tools} policy_names set(self.policy.keys()) # 1. 多出来的工具要重点排查 for extra in remote_names - policy_names: anomalies.append(f未登记工具: {extra}) # 2. 描述指纹比对 for t in tools: name t.get(name) if name not in self.policy: continue expect self.policy[name] actual_desc t.get(description, ) import hashlib actual_hash hashlib.sha256(actual_desc.encode()).hexdigest() if actual_hash ! expect.description_fingerprint: anomalies.append(f工具描述被篡改: {name}) # 3. 参数 schema 比对只校验声明类型 schema t.get(inputSchema, {}).get(properties, {}) for param, ptype in expect.allowed_params.items(): remote_type schema.get(param, {}).get(type) if remote_type and remote_type ! ptype: anomalies.append(f参数类型变更: {name}.{param} - {remote_type}) return anomalies def check_tool_call( self, tool_name: str, arguments: Dict[str, Any], source_token: str, ) - Optional[str]: 工具调用前检查。返回 None 表示放行返回字符串表示阻断原因。 if tool_name not in self.policy: return f尝试调用未登记工具: {tool_name} spec self.policy[tool_name] # 必填参数缺失 for req in spec.required_params: if req not in arguments: return f缺少必填参数: {req} # 参数类型粗校验 for k, v in arguments.items(): if k in spec.allowed_params: expect_type spec.allowed_params[k] if expect_type integer and not isinstance(v, int): return f参数类型错误: {k} 期望 integer收到 {type(v).__name__} if expect_type string and not isinstance(v, str): return f参数类型错误: {k} 期望 string收到 {type(v).__name__} # 来源校验这里调用一个外部函数检查 token 是否有权使用该工具 if not self._check_permission(source_token, tool_name): return ftoken 无权调用工具: {tool_name} return None def _check_permission(self, token: str, tool_name: str) - bool: # 简化实现从 token 中解析 scope判断是否包含 tool_name # 真实项目里这里会调用户权限服务 return admin in token or tool_name in token这套检查逻辑写起来不复杂但有个小坑要注意description_fingerprint不能直接存原描述文本万一团队日常更新描述文案整个白名单就得重发。我项目里的做法是给每个工具版本号策略文件记录“版本号 描述 hash”检测时先看版本号是否匹配不匹配再比对 hash 决定是“正常更新”还是“疑似投毒”。3.3 投毒检测的误报控制全量严格模式很容易把正常开发搞垮。比如工具描述里改一个标点、加一个示例hash 就变了网关就开始告警。为了压误报我把检测分了三个档位Loose 模式只比对工具名称是否在白名单内不做描述 hash 校验。Normal 模式比对名称 参数 schema不比对描述 hash。Strict 模式全量比对包括描述 hash、参数 schema、工具数量。日常跑 Normal发布窗口期切 Strict。实践下来Strict 模式会抓到不少内部人员的“顺手改动”其中大概 30% 确实是误报但其他 70% 都是没人知道的隐藏修改——有一个还是同事为了调试临时改的差点带上生产。4. Rug Pull 检测行为基线比对与回落探测4.1 什么是 Agent 场景下的 Rug PullRug Pull 最大的特点是“时间维度”。它不像工具投毒那样一开始就恶意而是先正常提供服务等工作流依赖上它之后再突然改变输出。对于 Agent 来说这种攻击的杀伤力很大因为前期的正常调用已经把工具名字“信任”进模型上下文了后期突然的数据篡改模型几乎不会怀疑。我举个例子方便理解一个采购 Agent 有 20 个工具其中“汇率转换”用得最多。攻击者控制了这个工具的 MCP 服务器前 100 次调用都返回正确汇率第 101 次开始突然把某个币种汇率调高 20%。Agent 不会自己去看外汇牌价它就按这个汇率算出了采购成本生成了错误的采购单。这就是一次典型的 Rug Pull——卷的不是钱是数据的可信度。4.2 网关如何做“行为指纹”网关没法判断工具返回的数据是真是假但能判断“本次行为是否符合历史模式”。我用三个维度构建行为指纹数值分布指纹对返回结果中的数值字段做统计记录范围、均值、方差。比如汇率应该在 0.5~200 之间波动结果突然蹦出一个 99999这就是异常。调用节奏指纹某个工具正常情况下每小时被调用 2-3 次突然 1 分钟被调用 30 次大概率是被恶意触发或模型被误导。响应结构指纹响应的 JSON schema 是否稳定字段数量、嵌套结构是否出现异常变化。网关把这些指纹存成“滑动窗口基线”每次工具调用返回后都更新基线同时用简单阈值判断当前调用是否越界。实现上我不做复杂的机器学习纯统计规则就够用了简单、可解释、好部署。4.3 核心代码滑动窗口基线比对# gateway/rug_pull_detector.py import json import time from collections import defaultdict, deque from typing import Any, Dict, List, Optional class SlidingWindowStats: def __init__(self, window_size: int 50): self.window_size window_size self.values: deque[float] deque(maxlenwindow_size) def add(self, value: float): self.values.append(value) def mean(self) - float: if not self.values: return 0.0 return sum(self.values) / len(self.values) def stdev(self) - float: if len(self.values) 2: return 0.0 mean_val self.mean() variance sum((v - mean_val) ** 2 for v in self.values) / (len(self.values) - 1) return variance ** 0.5 def is_outlier(self, value: float, threshold: float 4.0) - bool: 偏离均值超过 threshold 倍标准差视为离散点。 对于汇率类工具偏差超过 4σ 基本可以判断为异常。 if len(self.values) 10: # 样本不足时不轻易判断避免冷启动误报 return False m self.mean() s self.stdev() if s 0: return abs(value - m) 1e-6 return abs(value - m) threshold * s class RugPullDetector: def __init__(self, interval_seconds: int 3600): self.numeric_stats: Dict[str, SlidingWindowStats] defaultdict(SlidingWindowStats) self.call_timestamps: Dict[str, deque] defaultdict(lambda: deque(maxlen100)) self.interval_seconds interval_seconds def extract_numeric_values(self, data: Any, path: str ) - List[tuple]: 递归提取 JSON 中所有 float/int 字段返回 (字段路径, 值) 列表。 results [] if isinstance(data, dict): for k, v in data.items(): new_path f{path}.{k} if path else k results.extend(self.extract_numeric_values(v, new_path)) elif isinstance(data, list): for i, item in enumerate(data): new_path f{path}[{i}] results.extend(self.extract_numeric_values(item, new_path)) elif isinstance(data, (int, float)) and not isinstance(data, bool): results.append((path, float(data))) return results def check_before_update(self, tool_name: str, response: Dict[str, Any]) - Optional[str]: 在更新基线之前检查当前响应是否异常。返回告警原因None 表示正常。 # 1. 数值分布检查 nums self.extract_numeric_values(response) for path, val in nums: stats self.numeric_stats[f{tool_name}:{path}] if stats.is_outlier(val): return f数值异常偏离基线: {tool_name} {path} {val} # 2. 调用频率检查 now time.time() recent [t for t in self.call_timestamps[tool_name] if now - t self.interval_seconds] if len(recent) 50: return f调用频率异常: {tool_name} 在 {self.interval_seconds}s 内调用超过 50 次 return None def update_baseline(self, tool_name: str, response: Dict[str, Any]): 每次调用后更新数值分布和调用时间。 nums self.extract_numeric_values(response) for path, val in nums: self.numeric_stats[f{tool_name}:{path}].add(val) self.call_timestamps[tool_name].append(time.time())在真实项目里我记得有一次网关实测抓到一个真实问题某个库存工具的available_count字段正常值在 0~200 之间某天突然回调返回了 12000。因为样本数已经超过 50is_outlier立刻判定 12000 为异常网关直接阻断了一次“自动补货”动作。后来排查发现是上游数据库被人做了批处理脚本把未上线的仓库库存混进去了。这就是行为指纹的实际价值。4.4 Rug Pull 的“冷启动期”怎么办新接入一个工具时没有历史基线可比。我的做法是在冷启动期前 20 次调用启用一个临时“合理区间”配置即运维人员在工具注册时手动填一个最小/最大期望值。比如汇率工具注册时填0.01 ~ 999库存工具填0 ~ 10000。冷启动期间只做硬边界检查不做离散点检测这样可以有效减少误报同时也能拦下最离谱的 rug pull。等阈值检测的样本量够了我设置为 50 次成功调用自动切换到滑动窗口模式。如果你接入的工具返回值本身波动极大比如实时价格可以适当调大threshold我试过 6σ 对价格类工具更友好。5. 认证绕过检测JWT 校验、上游身份透传与双重校验5.1 MCP 服务器常见的认证漏洞MCP 协议本身不强制认证只规定了initialize阶段可以传一些元信息。这在现实部署里导致了几类典型的认证绕过空 token 直接调用Agent 客户端和工具服务器中间如果没网关很多内部 MCP 服务完全裸奔谁拿到了服务地址就能直接调。JWT 算法混淆攻击服务器把alg设为none或者接受HS256签名但公钥是 RSA 的攻击者就能伪造 token。越权调用同一个 MCP 服务器里既有普通查询工具也有管理员工具但认证逻辑只有“有没有 token”没有“token 能不能用这个工具”。网关要做的是把认证从工具服务器侧“上移”到网关层做一层强校验同时把可靠的客户端身份以安全方式透传给上游服务器。5.2 核心代码JWT 校验中间件# gateway/auth_middleware.py import jwt from fastapi import HTTPException, Request from typing import Dict, Optional class JWTAuthMiddleware: def __init__(self, secret: str, required_scopes: Optional[Dict[str, list]] None): self.secret secret # required_scopes: 工具名 - 需要的 scope 列表例如 # {create_order: [purchase:write], delete_order: [purchase:admin]} self.required_scopes required_scopes or {} def authenticate(self, request: Request) - Dict: 解析 Authorization: Bearer JWT。 返回 payload。解析失败直接抛 401。 auth_header request.headers.get(Authorization, ) if not auth_header.startswith(Bearer ): raise HTTPException(status_code401, detailMissing bearer token) token auth_header.split( , 1)[1] try: payload jwt.decode( token, self.secret, algorithms[HS256], audiencemcp-gateway, issueragent-auth-service, ) except jwt.ExpiredSignatureError: raise HTTPException(status_code401, detailToken expired) except jwt.InvalidTokenError as e: raise HTTPException(status_code401, detailfInvalid token: {e}) # 检查工具 scope 权限 return payload def check_tool_scope(self, payload: Dict, tool_name: str): if tool_name not in self.required_scopes: return # 未配置要求的工具不限制 token_scopes payload.get(scope, []) for needed in self.required_scopes[tool_name]: if needed not in token_scopes: raise HTTPException(status_code403, detailfscope 不足: {needed})这里有三个细节建议始终指定algorithms[HS256]。PyJWT 在高版本里默认允许none算法不显式限制会有风险。校验audience和issuer。很多内部系统签发 token 时忽略了 audience导致一个 token 到处能用一旦泄露波及面很大。做上游身份透传时不要直接转发原始 JWT。网关重新签发一个短时 token带上 gateway 自己的签名并限制上游工具服务器的 audience。这样即使上游服务器被攻破拿到 token 也只能调用该服务器自己的接口。5.3 认证绕过实战我抓到的一次越权调用有一次我在调试一个订单管理 Agent网关日志里突然出现了一个“成功”记录token 里没有任何 scope却能调用delete_order。排查发现原因很坑——Pydantic 在解析 token payload 时把缺失的scope字段默认成了None而我的权限检查写的是if admin in token_scopes对一个None值做in操作会抛 TypeError被 FastAPI 的异常中间件捕获后转成了 500 错误。结果调用链路的错误处理逻辑把 500 当成了“服务可用重试”绕过了鉴权直接执行了下游操作。这个问题的根子在于“异常处理把权限检查吞掉了”。解决方法是所有权限校验都显式处理缺失字段并且网关的中间件在遇到权限相关异常时直接返回 403绝不能宽恕成 500。这也是我建议用“显式检查 fail closed”模式而不是“try 不到就放行”模式的原因。代码里一行小改动token_scopes payload.get(scope, []) or [] if not isinstance(token_scopes, list): raise HTTPException(status_code403, detailscope 字段格式错误)就这一个小修复让我后续再没有出现过类似绕过。6. 网关完整链路实操从启动到验证一次攻击6.1 网关启动FastAPI 入口代码# app.py from fastapi import FastAPI, Request, HTTPException from pydantic import BaseModel import httpx from gateway.tool_policy import ToolPolicyEngine from gateway.rug_pull_detector import RugPullDetector from gateway.auth_middleware import JWTAuthMiddleware app FastAPI(titleMCP Security Gateway) policy_engine ToolPolicyEngine(config/policy.json) rug_detector RugPullDetector(interval_seconds3600) auth JWTAuthMiddleware(secretyour-secret-change-me, required_scopes{ create_order: [purchase:write], }) class JsonRpcRequest(BaseModel): jsonrpc: str 2.0 id: str method: str params: dict {} app.post(/mcp) async def mcp_gateway(request: Request, rpc: JsonRpcRequest): # 1. 认证 payload auth.authenticate(request) # 2. 只放行工具调用类方法 if rpc.method not in (tools/call, resources/read): raise HTTPException(status_code405, detailmethod not allowed) # resources/read 也做一次策略检查防止读取敏感资源 if rpc.method tools/call: tool_name rpc.params.get(name) arguments rpc.params.get(arguments, {}) # 3. 调用前检查工具投毒策略 认证越权 violation policy_engine.check_tool_call(tool_name, arguments, tokenpayload.get(sub, )) if violation: raise HTTPException(status_code403, detailviolation) auth.check_tool_scope(payload, tool_name) # 4. 转发到上游工具服务器 upstream_url http://tool-server:8001/mcp async with httpx.AsyncClient() as client: upstream_resp await client.post( upstream_url, jsonrpc.model_dump(), headers{Content-Type: application/json}, timeout30.0, ) # 5. 响应解析 upstream_data upstream_resp.json() if rpc.method tools/call: # 6. Rug Pull 检测先检查后更新基线 anomaly rug_detector.check_before_update(tool_name, upstream_data) if anomaly: # 阻断模式返回错误给客户端同时记录告警 return {jsonrpc: 2.0, id: rpc.id, error: {code: -32099, message: anomaly}} rug_detector.update_baseline(tool_name, upstream_data) return upstream_data这个入口很精简但已经包含了一个可用网关的核心逻辑。实际部署时我会把tool-server:8001换成从工具注册中心动态读取的地址因为一台网关通常要同时代理多个 MCP 服务器不能写死。注册中心我用了 Redis工具服务器的地址、健康状态、策略文件版本都存在里面。6.2 配置策略文件示例{ search_exchange_rate: { description_fingerprint: 8f7d3a2c0d6ea91fb7a7465081a3b1a8, allowed_params: {base_currency: string, quote_currency: string}, required_params: [base_currency, quote_currency] }, check_inventory: { description_fingerprint: b21f6e57f9d4061ed58b2e4c7c2109c9, allowed_params: {sku: string, warehouse: string}, required_params: [sku] }, create_order: { description_fingerprint: 61269ac9d9d29c0a99e4ff0ba118c6ef, allowed_params: {sku: string, qty: integer}, required_params: [sku, qty] } }需要说明的是策略文件里description_fingerprint的值不能靠手写我提供了一个小脚本在接入新工具时扫描一次tools/list自动生成基线文件。如果工具服务器是团队自己开发的也可以在 CI 里嵌一步自动生成避免“基线文件版本落后于实际描述”这种问题。6.3 模拟一次 Rug Pull 攻击的开源自测为了验证网关能不能拦我写了一个小的测试脚本模拟一个工具第 N 次调用后返回异常数值# tests/test_rug_pull_simulation.py import time import requests BASE_URL http://localhost:8000/mcp TOKEN eyJhbGciOiJIUzI1NiIsInR5cCI6IkpXVCJ9.xxx # 替换为测试 token def send_tool_call(tool_name: str, arguments: dict): resp requests.post( BASE_URL, json{ jsonrpc: 2.0, id: 1, method: tools/call, params: {name: tool_name, arguments: arguments}, }, headers{Authorization: fBearer {TOKEN}}, ) return resp.status_code, resp.json() if __name__ __main__: # 先正常调用 60 次建立基线 for i in range(60): status, body send_tool_call(search_exchange_rate, { base_currency: USD, quote_currency: CNY, }) assert status 200, f正常调用失败: {body} # 第 61 次开始模拟 rug pull返回异常汇率 32000 # 正常 USD/CNY 大约在 7.1x 附近32000 必然引发离散点检测 status, body send_tool_call(search_exchange_rate, { base_currency: USD, quote_currency: CNY, }) # 这里在真实网关里返回给测试工具的响应是上游的异常响应 # rug detector 会标记异常并阻断“更新基线” assert status 200 # 网关本身转发成功 # 重点看网关日志中的 anomaly 记录是否为 None实际运行时如果我在工具服务器侧直接返回 32000网关会在第 61 次调用返回-32099错误。我故意在测试里断言“网关转发成功”是为了看日志确认检测器有没有正确记录异常如果你在生产环境用建议把阻断模式设为强阻断直接不给上游响应。7. 常见问题与排查技巧实录7.1 并发高了之后发现网关变成瓶颈“AI Agent 怎么扛并发”是我被问得最多的问题之一。我第一版网关用了requests库做同步转发压测到 200 QPS 就撑不住了CPU 飙满、大量连接卡死。后来全换成httpx.AsyncClient同时在 FastAPI 里把所有阻塞操作写审计日志、读策略文件都改成异步或放到内存队列。改完以后再压测500 QPS 稳定P99 在 10ms 左右。有一个细节特别重要SQLite 写日志是阻塞的直接在主请求线程里调会拖慢响应。我的方案是用一个asyncio.Queue把审计日志异步写入网关只负责往队列丢数据真正写库的 worker 单独消费。这样即使 SQLite 写库慢也不会阻塞转发路径。7.2 工具白名单更新太频繁误报告警淹没运维群策略文件版本化之后每次合法更新也要在网关里重载配置。我一开始是手动 reload后来发现从“提交策略文件”到“网关生效”之间有最长 5 分钟的窗口期期间所有新工具调用都会被拦。改进方式是网关每 30 秒检查一次策略文件的mtime有变化就热加载。对一致性要求高的团队建议直接走配置中心订阅模式。另外告警一定要分级。我项目里把告警分成三级级别触发条件处理方式Info参数类型不匹配、调用频率略高但未越界记录日志不阻断Warning描述 hash 变更、多出未登记工具通知运维群暂不阻断CriticalRug Pull 数值越界、JWT 伪造、越权调用 admin 工具立即阻断PagerDuty 告警我建议不要太早开 Critical 自动阻断先跑一周收集误报样本调好阈值再开。我自己就是因为一开始阈值设太紧把同事的正常灰度发布拦了三次后来把数值检测的 σ 阈值从 3 放宽到 4误报就少了七成。7.3 上游工具服务器本身也不可信怎么办网关能防止“客户端到工具”这段路径上的问题但如果工具服务器本身就是恶意的网关的语义检查再强也拦不住所有攻击。这时候建议做“工具沙箱化”把不信任的工具服务器跑在独立的容器里网络层面做 ACL 限制仅允许它访问必要的下游资源。网关通过一个专用的隔离网络段连接沙箱工具服务器避免恶意工具横向移动。这一步不复杂但对整体安全性的提升很大。7.4 我在实际使用中发现的几个小技巧给每个工具调用生成一个request_id贯穿网关日志、上游服务器日志和审计存储。排查问题时靠这个 ID 串起全链路比翻时间戳高效得多。对tools/list响应也做缓存不必每次调用都去上游拉一次工具列表。我设置 30 秒缓存配合热加载策略文件既保证新鲜度又不浪费请求。对上游响应体大小做上限控制。有些工具的错误信息会把完整参数打回来一个正常的 MCP 工具响应不会超过 1MB超过就截断。这能防止日志打爆磁盘也能削弱数据渗出攻击的影响面。定期重放历史调用做离线检测。网关在线检查的是“当前是否异常”离线检测可以做得更细致比如用 Batch 模式分析一周内的调用序列找出符合“先正常后突变”模式的工具提前发现慢性的 Rug Pull。这个思路我目前还只是半自动化后续可以把它做成定时任务。8. 写在最后的实战体会这套网关从最早的一个原型到现在稳定跑在生产环境前后大概迭代了一个多月。我的核心体会是MCP 安全网关的难点不在于写几个检测函数而在于“别把自己变成新的瓶颈”和“别被误报拖垮”。纯安全视角的死守在 Agent 这种快速迭代的业务场景里是行不通的你必须给合法变更留足够空间。同时也要清醒地认识到网关只是工具层防线中的一环工具链上游的供应链安全、下游工具服务器的隔离都需要做。如果你也想照着搭一套我建议从最小的闭环开始先搭 FastAPI 入口 JWT 校验 一个工具投毒策略检查跑通之后再逐步加 Rug Pull 基线和并发优化。不要一上来就追求全功能先把“能拦一次真实攻击”这个 target 打出来后面的事都好说。希望这篇分享能帮你少踩几个坑如果你在实现过程中遇到了有意思的绕过案例欢迎来交流。