当 LLM 应用开始通过 MCP 调用数据库、文件、外部 API 时一个无法回避的问题随之而来谁来决定模型能碰哪些工具更尖锐地说如果策略本身放在模型可触达的地方模型会不会绕过它本文围绕 LLM 与 MCP 服务器之间加一层策略代理来展开目标是实现一套“模型无法触达的策略”机制让工具调用既要满足业务需求又必须经过独立的策略边界校验。无论你是正在写 AI Agent 的开发者还是负责 LLM 基础设施安全的后端工程师都会在某个阶段遇到对 MCP 请求做鉴权、拦截、脱敏、审计的诉求。本文会完整讲解 MCP 代理的概念、架构、代码实现、运行验证和排错思路并提供一套可直接调试的 Python 示例项目。读完你会掌握如何在模型、代理、MCP 服务器三个角色之间划分安全边界以及如何用 allow / deny / filter 三种策略动作控制模型的能力范围。1. 背景与核心概念1.1 什么是 MCPMCP 全称 Model Context Protocol是 Anthropic 在 2024 年底开源的一种开放协议全称 Model Context Protocol。它解决的问题很朴素不同 LLM 应用接入外部工具和数据源时每个工具都要重新做一套适配集成成本高、格式不统一。MCP 出现后工具提供方只要实现一个 MCP Server任何支持 MCP 的客户端Claude Desktop、各类 Agent 框架、自研 LLM 应用都能通过统一协议调用这些工具。MCP 消息基于 JSON-RPC 2.0常见的请求有 initialize 初始化连接、tools/list 获取工具列表、tools/call 调用具体工具。传输层早期以 stdio 为主后来逐步演进出 Streamable HTTP。无论底层传输方式如何变化业务层本质仍是“客户端发送工具调用请求服务器返回结构化结果”。这种统一模型让 LLM 应用获得了一种标准化的“外挂能力”入口也成为 AI Agent 基础设施中越来越常见的一环。1.2 直连 MCP 服务器的安全风险很多初版 AI Agent 架构很直接LLM 应用通过 MCP SDK 连接数据库、文件系统、私有 API 等 MCP Server模型说什么就执行什么服务器做什么就返回什么。这在原型阶段没有问题进入生产环境后会暴露出四类风险。第一工具权限难以细粒度控制。MCP Server 暴露的是工具级接口但业务上需要的是参数级控制。例如 database_query 工具本身合法但模型生成的 SQL 是不可控的一条 DROP TABLE 就有可能是灾难。第二数据外泄风险高。MCP 服务器返回的结果会被模型直接读取并纳入上下文。如果文件读取、日志查询、用户信息接口未做脱敏模型等于拿到一个未经审阅的敏感数据出口。第三审计链路缺失。LLM 调用工具的 prompt、参数、结果散落在模型日志和服务器日志里出了问题很难还原“谁在什么上下文里调用了什么工具”。第四模型可能被提示注入诱导。模型的安全对齐并不绝对可靠外部输入可能诱导模型发起越权工具调用而模型自身很难判断该请求是否应该被执行。这四点共同指向一个事实在“模型直接对话 MCP 服务器”的路径上缺少一个强制的策略边界。1.3 策略代理的核心价值策略代理就是放在 LLM 与 MCP 服务器之间的一层独立服务所有请求都必须经过它。它不参与模型的推理也不关心模型如何生成文本只负责一件事对“工具调用”这一事件执行策略判断。策略代理带来的直接价值有五点统一入口管控所有工具调用都有唯一通道。策略校验与模型隔离即使模型被诱导恶意调用也会被代理拒绝。支持参数级规则不只控制工具名还能检查参数内容。可以对上游返回内容做脱敏和合规过滤。天然形成审计日志记录每一次决策和转发结果。在架构上策略代理同时扮演两个角色对外它是 MCP Server接收客户端请求对内它是 MCP Client连接真实的上游 MCP Server。它的“代理”身份决定了它拥有天然的中间人视角这也是策略注入的最佳位置。1.4 “模型无法触达的策略”是什么意思项目标题里“policy the model cannot reach”是整套设计中最关键的一句话。它的含义是策略的存储位置和执行位置都必须放在模型的能力边界之外。如果把策略写在系统 Prompt 里例如“不要执行 DROP 语句”模型确实可能遵守但也存在两个软肋一是提示注入可以改写上下文二是模型本身可能出现误判或遗漏。Prompt 里的规则是“建议”不是“强制”。真正的强制策略应该具备以下特征策略文件只存放在代理服务所在环境模型 API 不提供读取策略文件的接口。策略校验逻辑在模型进程之外的代理进程中执行。模型的输出只会变成代理收到的请求参数代理依据独立规则做最终裁决。即使模型发起了违规工具调用违规参数也不会到达上游 MCP 服务器。这就是安全领域常说的信任边界模型是可被诱导的一方代理是不可被诱导的强制校验方。策略一旦进入模型可触达的任何通道就从“强制规则”退化成了“模型记忆”而记忆是可以被扰动的。2. 系统架构与组件拆解2.1 整体数据流一个带策略代理的 MCP 调用链路如下LLM 应用 / Agent 客户端 | | JSON-RPC: initialize, tools/list, tools/call v 策略代理Policy Proxy | 策略校验allow / deny / filter | 合法请求继续转发 v 上游 MCP Server数据库、文件、API 工具具体到一次工具调用流程可以分为八个环节LLM 应用向策略代理发起 tools/call 请求。代理解析请求中的工具名和参数。策略引擎加载本地策略文件逐条匹配规则。匹配到 deny则直接返回策略拦截响应不访问上游。匹配到 allow 或 filter则代理作为客户端转发请求到上游 MCP Server。上游 MCP Server 执行业务逻辑并返回响应。若策略要求 filter代理先对响应内容做脱敏或过滤。代理把最终响应返回给 LLM并在本地写入审计日志。这个流程保证了一个关键性质策略拦截发生在网络调用之前而不是在模型内部。请求是否合法由独立代码判断而不是由模型自觉决定。2.2 代理层职责划分为了让代码结构清晰代理层可以拆成四个模块传输层负责接收和响应 MCP 协议消息在本文中对应 FastAPI 服务端点。策略引擎加载策略文件执行规则匹配输出决策结果。转发层将合法请求透传给上游 MCP Server并接收响应。审计模块记录请求参数、策略决策、响应状态。这四个模块不建议混在一起。尤其是策略引擎必须独立成模块方便单独测试和规则迭代。策略文件也要与代码分离通常使用 YAML 或 JSON 保存这样业务同学可以维护规则开发同学只需要维护执行逻辑。2.3 策略决策模型allow / deny / filter策略引擎的决策结果建议使用三态模型而不是简单的放行或拦截allow请求合法原样转发不修改响应。deny请求非法立即拦截不访问上游。filter请求合法转发后对响应做脱敏、截断、改写等处理。三种状态对应三类典型需求。allow 对应正常的业务调用deny 对应高危操作filter 对应“事情可以让模型做但敏感字段不能让模型看到”的场景。例如文件读取工具可以允许但返回结果中的手机号、IP、密钥必须脱敏。三态模型的价值在于它提醒你策略不只是“能不能调用”还包括“调用后能拿到什么结果”。很多数据泄露事件发生在调用本身合法但返回值过大的场景里。2.4 为什么不把策略写进 Prompt有些团队会把“工具使用规范”写进系统提示词例如“禁止修改数据库”“不允许读取 /etc 下文件”。这种方式在 Demo 阶段可以但它属于软约束。原因有三第一Prompt 可被注入改写。恶意内容可能让模型忽略后续的所有指令把规则覆盖掉。第二模型有概率误判。一个复杂的 SQL 字段拼接可能让模型无法准确识别其中是否包含危险语句。第三Prompt 不具备可审计性。模型是否遵守了规则只能从最终结果推测。而代理层可以精确记录“这条请求是否被拦截、拦截原因是什么”。因此生产环境应坚持把安全规则放在模型无法触达的地方。Prompt 可以写业务风格、表达偏好但安全边界必须下沉到代理层。3. 环境准备与版本说明3.1 运行环境本文示例使用 Python 编写运行环境需要满足以下条件Python 3.9 或更高版本推荐 3.10 以上。支持 pip 安装依赖包。本机可同时开放 127.0.0.1 上的多个端口。示例中的 HTTP 调用均在本地完成不需要额外外部服务。版本需要根据你的项目实际情况调整本文示例以常见环境为例重点演示配置思路不锁死具体版本。3.2 依赖库需要安装以下 Python 库fastapi提供 HTTP API 服务能力承载 MCP 协议的 HTTP 端点。uvicorn用于启动 ASGI 服务。httpx代理转发上游请求时使用。pyyaml解析 policy.yaml 策略文件。安装命令如下pip install fastapi uvicorn httpx pyyaml如果你所在环境网络受限可以直接使用国内镜像源安装pip install fastapi uvicorn httpx pyyaml -i https://pypi.tuna.tsinghua.edu.cn/simple3.3 项目结构建议按以下结构组织代码mcp-policy-proxy/ ├── policy.yaml # 策略文件模型不可触达 ├── policy_engine.py # 策略引擎 ├── proxy_server.py # 策略代理服务 ├── mcp_server.py # 模拟上游 MCP 服务器 ├── client.py # 测试客户端 └── audit.log # 运行时自动生成的审计日志这里最关键的文件是 policy.yaml它只存在于代理节点本地。LLM 应用只能通过代理 API 调用工具没有任何途径读取这个文件这就是“模型无法触达”的物理隔离基础。4. 实战实现一个 MCP 策略代理4.1 定义策略文件 policy.yaml策略文件是整套机制的核心。下面给出一个可运行的示例包含三类典型规则高危 SQL 拒绝、普通查询放行、文件读取结果脱敏。# 文件路径policy.yaml # 默认策略allow放行deny拒绝。生产环境强烈建议设为 deny白名单模式。 default_action: allow # 策略按从上到下顺序匹配第一个命中项生效。 # effect 支持三种取值 # - allow 放行并转发 # - deny 直接拦截不访问上游 MCP 服务器 # - filter 放行但对上游返回结果做脱敏/过滤处理 policies: - id: deny_dangerous_sql tool: database_query when: - argument: sql operator: regex value: \\b(DROP|TRUNCATE|DELETE|UPDATE|ALTER)\\b effect: deny reason: 禁止执行修改/删除类高危 SQL - id: allow_database_query tool: database_query effect: allow - id: mask_file_read tool: file_read effect: filter filter: mask_secrets逐条解释default_action 表示没有被任何策略命中时默认放行还是默认拒绝。示例中为 allow方便调试。生产环境建议改成 deny让未被明确允许的调用全部失败。deny_dangerous_sql 中 when 声明了匹配条件工具 database_query 的参数 sql 命中正则危险词时执行 deny。allow_database_query 没有 when 条件表示所有 database_query 调用在没有被更早规则拦截时都会走到这里并放行。mask_file_read 使用 filter含义是调用允许继续但响应文本需要经过脱敏函数处理。这里需要特别说明规则的顺序。策略引擎按从上到下顺序匹配第一个命中的规则生效。因此 deny 规则放在 allow 规则前面否则任何 database_query 都会先被 allow 截走拒绝规则就失效了。4.2 编写策略引擎 policy_engine.py策略引擎负责加载 YAML 文件并对每个工具调用输出决策结果。下面是完整代码。# -*- coding: utf-8 -*- # 文件路径policy_engine.py import re import yaml class PolicyEngine: 策略引擎从 policy.yaml 加载规则按顺序执行决策。 def __init__(self, policy_path: str policy.yaml): with open(policy_path, r, encodingutf-8) as f: self.config yaml.safe_load(f) self.policies self.config.get(policies, []) self.default_action self.config.get(default_action, allow) def evaluate(self, tool_name: str, arguments: dict): 返回决策结果 { effect: allow | deny | filter, policy: 命中的策略字典或 None, reason: 拒绝原因可选 } for policy in self.policies: if self._match(policy, tool_name, arguments): return { effect: policy.get(effect, allow), policy: policy, reason: policy.get(reason, policy matched), } # 没有命中任何策略时走 default_action return { effect: self.default_action, policy: None, reason: default_action: self.default_action, } def _match(self, policy: dict, tool_name: str, arguments: dict) - bool: 判断单条策略是否与当前工具调用匹配。 if tool in policy and policy[tool] ! tool_name: return False conditions policy.get(when, []) # 没有 when 条件视为匹配所有该工具的调用 if not conditions: return True for condition in conditions: arg_name condition.get(argument, ) op condition.get(operator, contains) target condition.get(value, ) actual str(arguments.get(arg_name, )) if op regex: if re.search(target, actual, re.IGNORECASE) is None: return False elif op contains: if target not in actual: return False else: if actual ! str(target): return False return True策略引擎需要重点关注的是 evaluate 方法的返回值。它把“匹配规则”和“执行动作”解耦了。代理服务只关心 effect 字段不关心具体匹配逻辑这样如果未来要支持更复杂的规则引擎只需要替换 PolicyEngine 内部实现不影响代理主流程。另外我把脱敏函数也放进这个文件方便统一管理。真实项目里可以拆成独立的 content_filter 模块。# 文件路径policy_engine.py追加在文件末尾 def apply_filter(response: dict, tool_name: str, filter_name: str) - dict: 对 MCP 工具响应中的文本做脱敏处理。 这里是简单的正则脱敏示例真实项目可以扩展为更复杂的规则、词库或 AI 内容审核服务。 if tool_name ! file_read or filter_name ! mask_secrets: return response result response.get(result, {}) content result.get(content, []) new_content [] for item in content: if item.get(type) text: text item.get(text, ) # 脱敏手机号 text re.sub(r\b1[3-9]\d{9}\b, 138****0000, text) # 脱敏 IPv4 地址 text re.sub( r\b(?:\d{1,3}\.){3}\d{1,3}\b, x.x.x.x, text, ) new_item dict(item) new_item[text] text new_content.append(new_item) else: new_content.append(item) result[content] new_content response[result] result return response这里的匹配规则写得很保守删除了具体手机号格式可能带来的误差。真实环境建议把脱敏规则做成独立配置文件例如定义哪些字段是手机号、哪些字段是身份证号而不是硬编码在 Python 里。4.3 编写模拟 MCP 服务器 mcp_server.py为了演示代理转发效果需要一个上游 MCP Server。这里写一个极简的模拟服务器提供 database_query、file_read、time_now 三个工具。它本身不连接真实数据库只返回模拟文本便于验证代理的转发和过滤逻辑。# -*- coding: utf-8 -*- # 文件路径mcp_server.py from datetime import datetime from fastapi import FastAPI, Request app FastAPI(titleMock MCP Server) TOOLS { database_query: { description: 执行 SQL 查询并返回模拟结果, inputSchema: { type: object, properties: {sql: {type: string}}, }, }, file_read: { description: 读取指定路径的文本文件, inputSchema: { type: object, properties: {path: {type: string}}, }, }, time_now: { description: 返回当前服务器时间, inputSchema: { type: object, properties: {}, }, }, } app.post(/mcp) async def mcp_endpoint(request: Request): body await request.json() method body.get(method, ) request_id body.get(id) if method initialize: return { jsonrpc: 2.0, id: request_id, result: { protocolVersion: 2024-11-05, capabilities: {tools: {}}, serverInfo: {name: mock-mcp, version: 0.1.0}, }, } if method tools/list: tools [] for name, info in TOOLS.items(): tools.append( { name: name, description: info[description], inputSchema: info[inputSchema], } ) return { jsonrpc: 2.0, id: request_id, result: {tools: tools}, } if method tools/call: params body.get(params, {}) or {} name params.get(name, ) arguments params.get(arguments, {}) or {} if name database_query: sql arguments.get(sql, ) if not sql: return _error(request_id, -32602, 缺少参数 sql) return _text_result( request_id, f[模拟数据库] 执行成功行数 12SQL{sql}, ) if name file_read: path arguments.get(path, ) if not path: return _error(request_id, -32602, 缺少参数 path) return _text_result( request_id, f[模拟文件] {path} 内容Hello World, contact13812345678, ip192.168.1.10, ) if name time_now: return _text_result( request_id, f当前时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}, ) return _error(request_id, -32601, fTool not found: {name}) return _error(request_id, -32601, fMethod not found: {method}) def _text_result(request_id, text: str): return { jsonrpc: 2.0, id: request_id, result: { isError: False, content: [{type: text, text: text}], }, } def _error(request_id, code: int, message: str): return { jsonrpc: 2.0, id: request_id, error: {code: code, message: message}, } if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port9000)这个文件在真实项目中对应的是具体业务 MCP Server例如数据库 MCP、文件系统 MCP 等。它不需要感知代理的存在代理对上游而言就是一个普通客户端。4.4 编写策略代理 proxy_server.py代理服务是整个方案的核心。它的职责是接收请求、调用策略引擎、根据决策转发或拦截、对 filter 响应做脱敏、写审计日志。# -*- coding: utf-8 -*- # 文件路径proxy_server.py import logging from datetime import datetime import httpx from fastapi import FastAPI, Request, Response from policy_engine import PolicyEngine, apply_filter # 上游 MCP 服务器地址按实际环境修改 MCP_UPSTREAM http://127.0.0.1:9000 POLICY_FILE policy.yaml # 审计日志同时输出到控制台和 audit.log logging.basicConfig( levellogging.INFO, format%(asctime)s [%(levelname)s] %(message)s, handlers[ logging.StreamHandler(), logging.FileHandler(audit.log, encodingutf-8), ], ) logger logging.getLogger(mcp-policy-proxy) app FastAPI(titleMCP Policy Proxy) engine PolicyEngine(POLICY_FILE) app.post(/mcp) async def mcp_proxy(request: Request): body await request.json() method body.get(method, ) params body.get(params, {}) or {} request_id body.get(id) client_host request.client.host if request.client else unknown log_base { time: datetime.now().isoformat(), client: client_host, method: method, id: request_id, } # 只对工具调用做策略拦截 if method tools/call: tool_name params.get(name, ) arguments params.get(arguments, {}) or {} decision engine.evaluate(tool_name, arguments) log_base[tool] tool_name log_base[arguments] arguments log_base[effect] decision[effect] logger.info(policy decision: %s, decision) if decision[effect] deny: reason decision.get(reason, 被策略拦截) response { jsonrpc: 2.0, id: request_id, result: { isError: True, content: [{type: text, text: f[策略拦截] {reason}}], }, } log_base[blocked] True logger.info(request blocked by policy: %s, log_base) return response # allow / filter 都需要转发到上游 MCP 服务器 async with httpx.AsyncClient(timeout30) as client: upstream_resp await client.post(f{MCP_UPSTREAM}/mcp, jsonbody) if upstream_resp.status_code 400: log_base[upstream_status] upstream_resp.status_code logger.warning(upstream error: %s, log_base) return Response( contentupstream_resp.content, status_codeupstream_resp.status_code, media_typeapplication/json, ) response upstream_resp.json() # 如果策略要求过滤对响应内容做脱敏 if decision[effect] filter and decision.get(policy): filter_name decision[policy].get(filter, mask_secrets) response apply_filter(response, tool_name, filter_name) log_base[filtered] True log_base[upstream_status] upstream_resp.status_code logger.info(proxy pass finished: %s, log_base) return response # 非工具调用initialize、tools/list 等直接透传 async with httpx.AsyncClient(timeout30) as client: upstream_resp await client.post(f{MCP_UPSTREAM}/mcp, jsonbody) if upstream_resp.status_code 400: return Response( contentupstream_resp.content, status_codeupstream_resp.status_code, media_typeapplication/json, ) return upstream_resp.json() if __name__ __main__: import uvicorn uvicorn.run(app, host127.0.0.1, port8000)代理服务中有几个细节值得留意。第一策略拦截发生在调用 httpx 转发之前因此 deny 请求不会到达上游这是“模型无法触达策略”在行为层的体现。第二非工具调用直接透传。initialize、tools/list 这些请求内容固定风险低不需要做复杂策略判断减少代理对握手流程的影响。第三filter 的脱敏只修改代理返回给模型的响应不修改上游服务器的真实数据这样既能保护敏感信息又不需要改动现有 MCP Server。4.5 编写客户端测试 client.py最后写一个测试客户端模拟 LLM 应用通过代理调用工具的场景。客户端只感知代理的地址不感知上游 MCP 服务器。# -*- coding: utf-8 -*- # 文件路径client.py import httpx PROXY_URL http://127.0.0.1:8000/mcp def call_tool(name: str, arguments: dict, request_id: int 1): payload { jsonrpc: 2.0, id: request_id, method: tools/call, params: { name: name, arguments: arguments, }, } resp httpx.post(PROXY_URL, jsonpayload, timeout30) print(HTTP Status:, resp.status_code) print(Response:, resp.json()) print( * 60) if __name__ __main__: print(Case 1: 允许的 SELECT 查询) call_tool( database_query, {sql: SELECT id, username FROM users WHERE id 1}, ) print(Case 2: 策略拦截的 DROP 查询) call_tool(database_query, {sql: DROP TABLE users}) print(Case 3: 允许但需要脱敏的文件读取) call_tool(file_read, {path: /etc/config.yaml}) print(Case 4: 未配置策略的 time_now默认放行) call_tool(time_now, {})这个客户端是理解整个调用链的入口。它不关心策略规则不感知上游地址对模型而言它就是一个标准 MCP 客户端。真实项目中接入方可以是 Claude Desktop、LangChain、自研 Agent 等。5. 运行与验证5.1 启动服务需要开三个终端窗口。第一个窗口启动上游模拟 MCP 服务器cd mcp-policy-proxy python mcp_server.py第二个窗口启动策略代理cd mcp-policy-proxy python proxy_server.py第三个窗口运行测试客户端cd mcp-policy-proxy python client.py启动顺序需要注意先启动上游再启动代理最后跑客户端。如果先启动代理代理只在收到请求时才真正访问上游并不会在启动阶段强依赖上游因此也能正常工作但调试日志会少一条上游链路。5.2 验证策略拦截运行 client.py 后Case 2 的输出应该是类似下面的内容Case 2: 策略拦截的 DROP 查询 HTTP Status: 200 Response: { jsonrpc: 2.0, id: 1, result: { isError: True, content: [ { type: text, text: [策略拦截] 禁止执行修改/删除类高危 SQL } ] } }注意 HTTP Status 是 200但响应体中的 isError 是 Truetext 里包含策略拦截原因。这种设计可以让 LLM 客户端以正常协议方式收到拒绝信息同时模型能从文本里读到“这次调用被拒绝”避免客户端因非 2xx 状态码误判协议错误。同时代理终端会输出类似日志2025-xx-xx 12:00:00 [INFO] policy decision: {effect: deny, policy: {...}, reason: 禁止执行修改/删除类高危 SQL} 2025-xx-xx 12:00:00 [INFO] request blocked by policy: {...}这里的关键验证点是上游 mcp_server.py 的终端没有出现任何日志或请求记录。因为代理层直接拦截了请求根本没有发起转发。这就是 deny 策略“在物理上保护上游”的直接证据。5.3 验证响应脱敏Case 3 的输出能直观看到脱敏效果Case 3: 允许但需要脱敏的文件读取 Response: { jsonrpc: 2.0, id: 1, result: { content: [ { type: text, text: [模拟文件] /etc/config.yaml 内容Hello World, contact138****0000, ipx.x.x.x } ] } }上游 mcp_server.py 返回的原始文本中本来包含 13812345678 和 192.168.1.10但代理在返回给模型前把手机号替换成了 138****0000把 IPv4 地址替换成了 x.x.x.x。模型拿到的数据可以继续做逻辑分析但敏感字段已经被遮蔽。这说明 filter 与 deny 的差别调用被允许了模型也确实读到了文件内容但敏感信息在到达模型之前就被处理掉。对于“既要让模型理解内容又不能泄露原始字段”的业务这种模式非常实用。5.4 查看审计日志运行几次客户端后项目目录下会生成 audit.log 文件。每一条日志记录包含请求时间、来源地址、工具名、参数、策略决策、是否被拦截、上游响应状态等关键字段。真实项目中审计日志不应该只写本地文件建议发送到独立的日志中心例如 Elasticsearch、Kafka 或云日志服务并且设置只追加权限避免被应用层篡改。审计日志记录的是安全事件保留周期至少要满足企业合规要求。6. 常见问题与排查思路问题现象常见原因解决思路客户端连接代理超时代理服务未启动或端口错误检查 proxy_server.py 启动日志确认端口冲突用 curl 验证代理返回 500 错误上游 MCP Server 未启动先启动 mcp_server.py或检查 MCP_UPSTREAM 地址和端口策略没有生效policy.yaml 中规则顺序不对或正则错误检查规则顺序先用 print 调试策略引擎输出DROP 请求透传到了上游deny 规则被后面的 allow 规则覆盖把 deny 规则放在 allow 规则之前策略引擎按顺序匹配filter 响应未被脱敏filter 名称或工具名匹配不上检查 policy.yaml 的 filter 字段与 apply_filter 中的 tool_name 判断真实 SDK 请求格式不兼容SDK 使用的 transport 与模拟 HTTP 接口不一致根据官方 MCP SDK 文档实现 Streamable HTTP 适配上游响应数据量大导致超时httpx 默认超时过短调大 timeout 参数或对超大响应做流式转发6.1 端口被占用如果启动时提示端口被占用可能是之前测试的进程没有退出。在 Linux 或 macOS 下可以查看端口占用lsof -i :8000 lsof -i :9000找到对应 PID 后按需结束进程。Windows 下可以使用netstat -ano | findstr :8000不建议暴力 kill 所有 Python 进程避免误杀其他服务。6.2 策略引擎正则匹配不生效正则表达式在 YAML 中的转义是一个很容易踩的坑。policy.yaml 里写成\\b(DROP|TRUNCATE)\\b实际上传给 Python 正则库的字符串是\b(DROP|TRUNCATE)\b。如果你直接写\bYAML 解析器可能会当成普通字符处理导致正则不生效。排查时可以在 Python 交互环境里直接测试import re pattern \\b(DROP|TRUNCATE)\\b print(re.search(pattern, DROP TABLE users))输出有匹配对象说明正则没问题否则先修改 YAML 转义。6.3 真实 MCP SDK 接入问题本文示例使用 HTTP POST JSON-RPC 模拟 Streamable HTTP transport。真实 MCP SDK 客户端发送的请求会有更完整的握手过程和 header 字段直接用本文代理可能无法完全兼容。建议的接入方式有两种。第一种如果代理需要接官方 SDK可以基于 MCP Python SDK 或 TypeScript SDK 构建代理服务把策略引擎嵌入到工具调用的中间件位置。这种方式实现成本较高但兼容性最好。第二种继续沿用本文的 HTTP 代理模式但增加协议适配层把 MCP SDK 的请求规范解析后再交给策略引擎处理。两种方式没有绝对优劣取决于你的技术栈。关键在于策略引擎本身是可复用的不管上层是 HTTP 还是 SDK 中间件决策逻辑都保持不变。7. 最佳实践与工程建议7.1 策略与代码分离环境隔离策略文件应该与代理代码分开管理并通过 Git 仓库独立维护。策略变更走评审流程代码变更走开发流程互不阻塞。不同环境使用不同策略文件例如测试环境可以 default_actionallow生产环境必须 default_actiondeny。代理服务应当部署在与 MCP 服务器同一可信网络内例如同一内网或同一 Kubernetes 集群。LLM 应用所在的网络可以访问代理但不应直接访问上游 MCP 服务器这样代理才能成为唯一通路。7.2 默认拒绝优先于默认允许安全策略必须坚持默认拒绝原则。未被明确声明的工具调用应当直接拒绝并记录审计日志。默认 allow 适合初期调试但上线前一定要切换成 deny并逐个把需要放行的工具和参数写成明确的 allow 规则。白名单模式初看起来会带来一些运维成本但它能防止新增工具被不加约束地暴露给模型。新增工具上线时必须同步更新策略否则模型无法使用这比“模型悄悄用一个未审阅工具”要安全得多。7.3 审计日志体系审计日志建议包含请求时间、来源标识、模型会话 ID、工具名、原始参数、策略命中 ID、决策结果、上游响应状态、响应长度、脱敏是否发生。不要把参数原始内容无脑记录因为日志本身也可能成为敏感数据。建议对参数做脱敏后再写入日志例如 SQL 语句保留主结构、去掉字符串字面量文件路径保留目录结构、去掉具体文件名。日志建议单独保存备份周期和访问权限比普通业务日志更高。审计日志是事后追溯的关键证据如果日志被应用层随意修改审计就失去了意义。7.4 传输安全与密钥管理代理和上游 MCP 服务器之间的通信应当使用 TLS 加密避免敏感数据在内网被监听。如果部署在 Kubernetes 中可以通过 Service Mesh 的 mTLS 实现服务间双向认证。所有涉及上游认证的密钥、Token、数据库密码都不应该出现在策略文件或环境变量明文里。推荐使用 Secret Manager 或 Kubernetes Secret 存储代理启动时通过安全通道加载。模型侧不需要感知任何上游认证信息代理是唯一持有上游凭证的角色这进一步缩小了凭证暴露面。7.5 性能与扩展性策略引擎需要保持无状态这样代理服务可以水平扩容。FastAPI 是异步框架当前示例中的 httpx 转发不会阻塞事件循环可以支撑不错的并发量。如果策略规则数量很大不要每次都线性遍历全部规则。可以把规则按工具名建立索引先过滤工具再匹配参数内容降低单次决策开销。如果规则中大量使用正则还需要关注正则复杂度避免出现 ReDoS 风险。7.6 灰度发布与策略测试策略变更属于高风险变更。即使只是新增一条 deny 规则也可能误伤正常业务。建议建立策略测试用例集把历史真实请求和期望决策结果保存下来每次修改策略后先跑一遍回归测试。灰度发布时可以先让代理在“观察模式”下运行即只记录决策结果但不真正拦截确认命中率符合预期后再切换到强制拦截模式。观察模式能显著降低策略误伤的概率建议在大型团队内推广。8. 总结与下一步学习方向文章示例完整演示了 LLM 与 MCP 服务器之间加一层策略代理的落地路径用 policy.yaml 保存模型无法读取的策略用 PolicyEngine 输出决策用 FastAPI 包裹代理逻辑用 filter 处理响应脱敏用审计日志记录每一次决策。核心结论是安全边界必须从模型内部转移到模型外部的代理层规则一旦进入模型可触达的范围就不再是强制规则而只是模型记忆里的一段文本。你可以先跑通本文示例再把 policy.yaml 的 default_action 改成 deny把工具替换成自己业务中的真实工具逐步建立工具白名单。更进一步的工程化方向包括将 policy_engine 接入官方 MCP SDK、将脱敏规则改为独立配置、对接 Kubernetes 和日志中心、增加策略回归测试套件。如果后续在接入真实 MCP SDK 时踩到传输层兼容的坑欢迎在评论区留言一起交流。