企业官网建设流程全解析

当 LLM 应用开始通过 MCP 调用数据库、文件、外部 API 时，一个无法回避的问题随之而来：谁来决定模型能碰哪些工具？更尖锐地说，如果策略本身放在模型可触达的地方，模型会不会绕过它？本文围绕 LLM 与 MCP 服务器之间加一层策略代理来展开，目标是实现一套“模型无法触达的策略”机制，让工具调用既要满足业务需求，又必须经过独立的策略边界校验。

无论你是正在写 AI Agent 的开发者，还是负责 LLM 基础设施安全的后端工程师，都会在某个阶段遇到对 MCP 请求做鉴权、拦截、脱敏、审计的诉求。本文会完整讲解 MCP 代理的概念、架构、代码实现、运行验证和排错思路，并提供一套可直接调试的 Python 示例项目。读完你会掌握如何在模型、代理、MCP 服务器三个角色之间划分安全边界，以及如何用 allow / deny / filter 三种策略动作控制模型的能力范围。

1. 背景与核心概念

1.1 什么是 MCP

MCP 全称 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 工具）

具体到一次工具调用，流程可以分为八个环节：

1. LLM 应用向策略代理发起 tools/call 请求。
2. 代理解析请求中的工具名和参数。
3. 策略引擎加载本地策略文件，逐条匹配规则。
4. 匹配到 deny，则直接返回策略拦截响应，不访问上游。
5. 匹配到 allow 或 filter，则代理作为客户端转发请求到上游 MCP Server。
6. 上游 MCP Server 执行业务逻辑并返回响应。
7. 若策略要求 filter，代理先对响应内容做脱敏或过滤。
8. 代理把最终响应返回给 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/simple

3.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", encoding="utf-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(title="Mock 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"[模拟数据库] 执行成功，行数 12，SQL={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, contact=13812345678, ip=192.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, f"Tool not found: {name}") return _error(request_id, -32601, f"Method 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, host="127.0.0.1", port=9000)

这个文件在真实项目中对应的是具体业务 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( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.StreamHandler(), logging.FileHandler("audit.log", encoding="utf-8"), ], ) logger = logging.getLogger("mcp-policy-proxy") app = FastAPI(title="MCP 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(timeout=30) as client: upstream_resp = await client.post(f"{MCP_UPSTREAM}/mcp", json=body) if upstream_resp.status_code >= 400: log_base["upstream_status"] = upstream_resp.status_code logger.warning("upstream error: %s", log_base) return Response( content=upstream_resp.content, status_code=upstream_resp.status_code, media_type="application/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(timeout=30) as client: upstream_resp = await client.post(f"{MCP_UPSTREAM}/mcp", json=body) if upstream_resp.status_code >= 400: return Response( content=upstream_resp.content, status_code=upstream_resp.status_code, media_type="application/json", ) return upstream_resp.json() if __name__ == "__main__": import uvicorn uvicorn.run(app, host="127.0.0.1", port=8000)

代理服务中有几个细节值得留意。

第一，策略拦截发生在调用 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, json=payload, timeout=30) 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 是 True，text 里包含策略拦截原因。这种设计可以让 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, contact=138****0000, ip=x.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. 常见问题与排查思路

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。如果你直接写\b，YAML 解析器可能会当成普通字符处理，导致正则不生效。

排查时可以在 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_action=allow，生产环境必须 default_action=deny。

代理服务应当部署在与 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 时踩到传输层兼容的坑，欢迎在评论区留言一起交流。