做 AI Agent 做得越久我对“工具全开”这件事越警惕。给 pi agent 接 MCP 那天我一股脑把浏览器控制、文件读写、命令行执行全都挂上去结果第一次真实跑任务就翻车了——agent 在没经过我同意的情况下把生产环境一个临时目录清了个干净。从那之后我开始认真琢磨能不能做一套按需开启的 MCP 机制让工具默认关闭需要时由人显式介入、授权、再执行这套方案最后就是 pi agent 上这套“人工介入网关”。这篇文章我会从 MCP 协议的理解讲起把带开关的 server 实现、审批流设计、客户端接入和真实环境里踩过的坑全部盘一遍。如果你手上正好有一个基于 LLM 的 agent 项目想让它调用外部工具又不想把控制权完全交出去这篇应该正是你需要的。1. 为什么人工介入是刚需先想清楚工具授权这件事1.1 一次“工具全开”的翻车现场先说那次事故的细节。我的 pi agent 当时跑了大概一周接了两个 MCP server一个 filesystem 工具集一个 shell 工具集。某天我想让它“把临时目录下的缓存文件清理一下”结果 agent 的规划器理解成“清理临时目录”直接调用了delete_tree把整个目录递归删掉了。这个目录里正好有正在跑的构建缓存等于我在没有任何确认的情况下让 AI 替我做了一个破坏性操作。事后复盘问题完全不在模型本身而在于我当时把所有工具都放进了它的可见列表里。这个案例特别典型的地方在于LLM 对用户意图的理解是概率性的它不是确定性程序。同一个“清理一下”在不同上下文里可能被翻译成完全不同的工具调用。工具一旦在 agent 的可见范围内它就会倾向于使用而且它不会主动判断“这个操作是不是太危险了我先问问人”。所以指望模型自觉不如从工程上堵住这个口子。MCP 在这里是一个能力放大器。协议本身让 agent 能够接入的工具数量和复杂度都上了一个台阶但这意味着破坏半径也跟着放大。浏览器自动化能帮你填表也能帮你点错按钮数据库工具能帮你查数据也能帮你删表。能力越大越需要把“谁能调用、什么条件下调用”这件事变成显式的工程约束而不是模型临场发挥的一部分。1.2 一句话拆清楚 MCP 的工作模型MCPModel Context Protocol说白了就是一套 AI 工具调用的标准协议解决的是“模型怎么发现工具、怎么调用工具、工具结果怎么返回给模型”这类问题。你可以把它类比成 USB-C 接口AI 像是电脑MCP server 像是外设协议规定了接口形状、数据格式、握手流程大家按这个标准插上就能用。MCP 的架构核心是客户端-服务器模型分几个关键角色MCP Client运行在 agent 或 AI 应用内部负责跟模型对话、向 server 发起工具发现和调用请求。MCP Server暴露工具Tools、资源Resources、提示词模板Prompts的独立服务可以跑在本地进程stdio 传输也可以跑在远程HTTP/SSE、WebSocket 传输。协议层基于 JSON-RPC 2.0定义了initialize、tools/list、tools/call、resources/read等标准方法。最常用的还是 Tools 原语。agent 启动时会从 server 拉取工具清单模型根据任务需要挑选合适的工具然后通过tools/call发出调用请求server 执行完把结果返回模型。整个调用链就是用户请求 → 模型规划 → 请求工具 → 工具执行 → 结果回传 → 模型整理输出。理解这个模型有什么用因为它决定了“按需开启”可以插在哪几个环节。你既可以在 client 端控制“不把某些工具发给模型”也可以在 server 端拦截“即使模型请求了这个工具也不让它执行”。我的做法是两层都做让 pi agent 的开关机制足够严密。1.3 按需开启要解决的三件事设计按需开启机制之前我给自己定了三个目标后面所有代码都是围绕这三个目标展开的第一默认关闭、显式开启。所有工具默认不出现在 agent 的工具列表里。只有当任务上下文明确需要某个工具时才把它临时挂载上去。这样做除了安全考虑还有个很实际的好处——减少上下文膨胀。每多一个工具模型要处理的 token 就多一堆工具列表越精简模型做决策的速度和准确率都会好一些。第二分级授权、人工介入。不是所有工具都需要人工审批。只读工具可以自动放行写操作工具必须弹确认破坏性工具默认拒绝。人工介入不是“不管三七二十一都问一遍”那样 agent 根本没法用而是像红绿灯一样该放行的放行该减速的减速该停的停。这套分级体系在后面会展开。第三全程审计、可追溯。每次工具调用都要留下记录谁调用的、什么时候、传了什么参数、结果如何、审批人是谁。这一条在出了问题之后价值最大。没有审计翻车了都不知道是哪一步的锅。这三个目标听起来简单落地的时候牵扯到的细节非常多。接下来我会把整套架构和代码一步一步拆开讲。2. 架构与选型给 pi agent 设计可控的 MCP 层2.1 双层开关Client 侧裁剪 Server 侧拦截我最终采用的架构是双层开关。为什么需要双层因为光在 server 端拦截agent 的模型还是会看到所有工具列表它规划的路径里可能包含禁用工具虽然最后会被拒但推理路径已经偏离了光在 client 端裁剪也不行因为一旦 agent 拿到了一个带危险工具的 server 地址它可以绕过 client 直接调用。所以两头都要管。具体到 pi agent 上就是这样的链路pi agent 本体作为 MCP Client启动时读取一份开关配置只加载当前任务需要的那几个 server 和工具每个 server 内部又做了一层守卫工具被调用时会经过一个审批网关网关根据配置决定是直接放行、询问人工还是直接拒绝。双保险的好处是即使 client 端配置出错把不该开的工具暴露了出去server 端也能拦得住。这套架构没有引入额外的编排框架核心就是两个部分一个配置驱动的工具加载器一个审批网关。pi agent 原本的 model 调用逻辑完全不用改只是在工具调用前加了一道检查。这也是我做这个项目比较满意的一点——侵入性很小。2.2 工具分域与权限分级给工具分级是整个方案里最需要想清楚的部分。我按两个维度给工具分类操作是否只读以及影响范围是否可恢复。只读、可恢复的工具放行写操作、可恢复的工具询问写操作、影响大或者不可恢复的工具直接拒绝或要求二次确认。实际操作中我把工具分成了三档对应不同的处理策略安全级别典型场景代表工具默认策略低危查询、读取Fetch 网页抓取、文件只读、数据库 SELECT自动放行中危局部写入、状态变更文件写入、代码执行、浏览器点击询问人工高危删除、批量修改、外部副作用文件删除、DROP/TRUNCATE、邮件发送拒绝或二次确认这个分级不是死的得结合业务上下文调。比如同样是文件写入写到 /tmp 缓存和写到生产配置目录风险天差地别。所以我在分级之外还加了一层路径/域名白名单规则命中了高危路径的工具即使级别是中危也会被自动升级拦截。规则引擎不用写得多复杂一组正则加一个前缀匹配表就够用。分级的作用是让“人工介入”不再是全有全无的选择而是可以根据风险动态调整介入力度。这也呼应了标题里说的“艺术”——介入的时机和程度要拿捏而不是一刀切。2.3 技术选型为什么用 FastMCPMCP server 的实现方式我对比过三条路线直接用官方 TS SDK 手搓、用 Python 的mcp官方 SDK 手写 JSON-RPC 处理、用 FastMCP 这类高层封装。最后选了 FastMCP原因很直接它是纯 Python 的装饰器注册工具的方式最贴合我这种以业务逻辑为主的场景而且它同时支持 stdio、SSE、streamable HTTP 三种传输方式方便我后期把 server 部署到远程。用 FastMCP 写一个工具只需要这样from fastmcp import FastMCP mcp FastMCP(pi-gateway) mcp.tool() def read_config(path: str) - str: 读取配置文件内容只读操作 with open(path, r, encodingutf-8) as f: return f.read()不用自己处理 JSON-RPC 的消息格式框架把协议的细节都包掉了。选 streamable HTTP 作为默认传输方式是因为它比 stdio 更适合跨进程部署pi agent 和 MCP server 可以不在同一台机器上agent 跑在服务器、工具服务跑在开发机上这种拓扑都能支持。选型的时候也考虑过用一个 MCP server 汇聚所有工具还是每种工具一个 server。最后选了后者按工具域拆成多个 server一个 filesystem server、一个 browser server、一个 database server。这样每个 server 的权限边界更清晰只暴露最小功能集出问题时隔离性也好。工具多了以后运维上要管的进程数会多一点但换来的是清晰的安全边界这笔账是划算的。3. 代码落地实现一个带按需开关的 MCP Server3.1 初始化项目与配置文件确定架构后先把项目骨架搭起来。我习惯用这样的目录结构pi-agent-gateway/ ├── pyproject.toml ├── config/ │ └── tools_config.yaml ├── src/ │ ├── server.py # FastMCP 入口 │ ├── gateway.py # 审批网关 │ └── tools/ │ ├── file_tools.py │ └── web_tools.py ├── logs/ │ └── audit.logpyproject.toml里依赖很简单核心就是fastmcp和pyyaml。如果你的环境还没装直接pip install fastmcp pyyaml就行。FastMCP 的版本演进比较快我写这篇文章时用的 2.x 版本接口上装饰器注册和mcp.run()这套是稳定的。配置文件是整条链路的“总开关”。我把所有工具的状态都收敛到一个 YAML 文件里这样改配置不用动代码审核变更也方便。核心的内容是每个工具的启用状态、安全级别和审批模式。# config/tools_config.yaml server: name: pi-agent-gateway transport: streamable-http tools: read_file: enabled: true safety: low mode: auto fetch_url: enabled: true safety: low mode: auto write_file: enabled: false safety: medium mode: ask delete_file: enabled: false safety: high mode: deny approval: timeout_seconds: 30 default_mode: ask这个配置文件承载了整个按需开启的核心逻辑enabled控制工具是否注册到 server 上未注册的工具即使被调用也会返回 not foundmode分成auto、ask、deny三档分别对应自动放行、询问人工和直接拒绝。后面 server 启动时会逐条读这个配置来动态注册工具。3.2 配置驱动的动态工具注册有了配置之后关键就是怎么让 server 按配置去注册工具而不是把所有工具写死在代码里。这里我用了一个很朴素的“注册器”模式定义一组工具实现函数启动时遍历配置只把enabled: true的工具挂到 FastMCP 实例上。拿文件工具举例核心代码长这样# src/server.py import yaml from fastmcp import FastMCP from tools import file_tools, web_tools mcp FastMCP(pi-agent-gateway) # 工具注册表名称 - (实现函数, 安全级别) TOOL_REGISTRY { read_file: (file_tools.read_file, low), write_file: (file_tools.write_file, medium), delete_file: (file_tools.delete_file, high), fetch_url: (web_tools.fetch_url, low), } def load_tools_from_config(config_path: str): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) for name, tool_config in config[tools].items(): if not tool_config.get(enabled, False): continue func, _ TOOL_REGISTRY[name] # 把工具函数注册到 mcp 实例上 mcp.tool()(func) print(f[gateway] tool registered: {name})这段代码解决了“按需开启”的第一层问题配置里没启用的工具压根不会出现在tools/list返回结果里模型看不到、也用不了。想临时开一个工具改一行 YAML 重启进程就行不用改任何业务代码。线上紧急恢复现场的时候这个能力非常救命。这里有一个容易踩的点动态注册工具后FastMCP 的 schema 生成是在注册时确定的如果你的工具函数有复杂的 Pydantic 模型入参注册前一定要保证模型定义完整不然生成的 JSON Schema 会不完整客户端拉取的时候可能直接报错。我后来统一规范了工具入参能不用复杂嵌套模型就不用全部用基本类型加Field(description...)这一个改动让工具发现的稳定性好了很多。3.3 人工审批网关三种模式与超时策略工具注册解决了“能不能看到”的问题审批网关解决“能不能执行”的问题。网关的核心是一个ApprovalGateway类每次tools/call进来的时候先走一遍检查逻辑。实现思路是这样的# src/gateway.py import enum import time import logging from typing import Any logger logging.getLogger(gateway) class ApprovalMode(enum.Enum): AUTO auto ASK ask DENY deny class ApprovalGateway: def __init__(self, mode_map: dict[str, dict], timeout_seconds: int 30): self.mode_map mode_map self.timeout_seconds timeout_seconds def check(self, tool_name: str, params: dict[str, Any]) - tuple[bool, str]: 返回 (是否允许执行, 原因/审批信息) if tool_name not in self.mode_map: return False, ftool {tool_name} is not registered mode ApprovalMode(self.mode_map[tool_name][mode]) if mode ApprovalMode.DENY: return False, tool is denied by policy if mode ApprovalMode.AUTO: return True, auto approved # ASK 模式需要人工介入 return self._ask_human(tool_name, params) def _ask_human(self, tool_name: str, params: dict[str, Any]) - tuple[bool, str]: # 把审批请求发送到外部人工端例如 Slack 机器人、Webhook 或终端提示 # 这里用一个阻塞等待的实现支持超时 decision self._request_approval(tool_name, params) if decision is False: return False, rejected by human return True, approved by human这个_request_approval的实现可以根据你的环境选择本地开发可以直接用input()在终端弹提示线上可以用 Webhook 推到企业微信或者 Slack让值班的人手机确认。我没有把这部分写死保留了一个可插拔的接口。超时策略是这里最容易被忽略的细节。我一开始没设超时审批请求发出去之后人一直没回应agent 的整个会话就挂在那里了。后来加了两层保护第一层是网关层的超时超过 30 秒默认拒绝执行保证调用链不会无限期阻塞第二层是客户端层的超时这个后面会讲到。安全侧的策略是“超时等同于拒绝”宁可让任务失败也不能在没人确认的情况下继续执行。这个原则我建议所有做人工介入机制的团队都采纳。4. 接入客户端与实战演示让按需开启真正跑起来4.1 客户端连接与工具发现server 端写完之后接下来就是让 pi agent 作为 MCP Client 去连它。用 FastMCP 跑起的 server 默认会暴露一个 HTTP endpoint客户端通过 streamable HTTP 传输去连接。客户端的连接代码大致如下# client_example.py import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def main(): endpoint http://localhost:8000/mcp async with streamablehttp_client(endpoint) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(Available tools:, [t.name for t in tools.tools]) result await session.call_tool( read_file, arguments{path: /tmp/example.txt} ) print(Tool result:, result) asyncio.run(main())这里有一个很容易搞混的点list_tools拿到的是当前 server 端已经加载的工具列表。因为 server 是按配置注册的所以你在这个列表里看到的就是真正可用的。如果某个工具没启用它根本不会出现在这个列表里。这也是我实测下来验证双层开关最直接的方式——先 list 一下看工具在不在。pi agent 本身的模型调用逻辑不需要大改只需要在原来的“模型选工具”和“执行工具”之间插入这个 MCP Client。模型输出一个工具调用意图pi agent 把它翻译成一次call_tool拿到结果再拼回去给模型。整个介入逻辑对模型完全透明模型不需要知道背后有没有人审批过。4.2 一个完整场景的完整时间线用实际的场景把整个流程串一遍。假设用户给 pi agent 发了一条指令“把 /tmp/pi-agent-cache 下面的 .tmp 文件清理掉保留 .log 文件。”这条指令会触发一连串动作。先看客户端这边pi agent 从工具列表里发现可用工具只有read_file和fetch_url因为write_file、delete_file默认关闭它可能觉得工具不够用会向用户提示“需要文件清理类工具”。这时候就是按需开启发挥作用的地方。你可以让用户直接在对话里确认“开启删除工具”或者由 agent 根据任务自动发起工具开启请求。开启之后server 端重新加载配置delete_file出现在工具列表里。agent 规划调用delete_file网关检查到它的 mode 是ask于是发起人工审批。审批通过后命令才真正执行。完整时间线是这样的用户: 清理 /tmp/pi-agent-cache 下的 .tmp 文件 → pi agent: 检查工具列表发现 delete_file 未启用 → 用户: 确认开启删除工具 → server: 重载配置delete_file 注册成功 → pi agent: 规划调用 delete_file(/tmp/pi-agent-cache, *.tmp) → 网关: modeask发起人工审批请求 → 人工端: 确认允许执行 → server: 执行删除返回结果给 agent → pi agent: 汇总清理结果给用户这条时间线里人工介入出现了两个地方一次是用户层面的“开启工具”一次是执行层面的“审批确认”。这就是我说的“按需开启”的完整形态——不是启动时全开而是在任务的每个关键节点由人决定要不要放行。你可以看到这套机制对用户来说并不繁琐只读操作完全不打扰破坏性操作也只多一步确认。4.3 运营侧的三档开关策略网关机制落地之后我经验里最值钱的部分其实是运营策略——什么任务跑什么模式。单任务临时执行我一般全程 auto 加 ask 混合批量任务或者定时任务我倾向于把 ask 模式关掉改成 deny 加白名单。这不是偷懒而是批量任务中如果每个工具都弹审批审批人很容易疲劳疲劳之后就会随手点同意那审批机制就形同虚设了。我目前的运营策略是三档场景策略说明探索性任务auto ask只读自动放行写操作人工确认正式执行任务ask 白名单高危路径白名单自动放行其他全部人工确认定时批处理deny 最小工具集只开白名单工具其余全部拒绝三档策略的核心是同一个配置模板只是参数不同。这正好是配置驱动模式带来的好处改 YAML 就能切换策略不用重新部署代码。我在实际运营中切换过很多次策略稳定性还是不错的。5. 常见问题与排查实录5.1 工具调用超时agent 干等第一次上线人工审批网关很快就发现问题agent 调用一个 ask 模式的工具审批请求发出去了人迟迟没有处理然后 agent 那边就报Tool call timed out。排查后发现是客户端侧的响应超时设置得太短默认只有 10 秒而人工审批从看到消息到做出决定正常都要 20 秒以上。解决方案是分层设置超时客户端连接层设 60 秒审批网关层设 30 秒两层超时独立工作。这样即使审批耗时较长agent 会话也不会被轻易打断反过来如果审批真的超时网关会先返回拒绝客户端收到明确结果不会出现一边在等、一边已经断开的情况。另一个连接相关的问题streamable HTTP 的 endpoint 如果长时间空闲连接会被中间层的代理或者负载均衡断开。我之前遇到过一个反复出现的Connection reset报错排查下来就是空闲超时问题。解决办法是客户端加心跳请求并且把错误处理改成重试逻辑工具调用失败时自动重新建立连接再试一次。5.2 审批阻塞导致会话卡死审批阻塞比超时更隐蔽。有一次线上任务跑着跑着整个 agent 会话不动了日志里没有任何报错就是静默挂起。查了半天发现是审批网关的_ask_human实现里有问题——审批请求发出去了但回调处理线程和主事件循环之间没有做好同步导致主循环一直阻塞在等待队列上既没有超时兜底也没有报错。这个坑的教训是要用异步化设计。现在的实现里审批请求发出后立刻返回一个 pending 状态主循环不会被阻塞审批结果通过回调或者轮询写入结果队列调用方拿到结果后再继续执行。这样即使人工端迟迟不回应agent 会话也只会停在等待状态不会整个崩溃。排查这类问题我有个习惯就是看日志里的 trace-id。每个工具调用我都带上一个唯一标识从客户端发起、网关审批、到执行完成全过程打同一个 trace-id 的日志。出问题的时候按 trace-id 一搜整个调用链一目了然定位快了不止一倍。5.3 token 与认证的保管问题最后说一个安全上的细节。MCP server 如果走远程传输endpoint 和 token 是像钥匙一样的东西。我的项目里把 endpoint、token 这类敏感信息全部放到环境变量里配置文件只放相对安全的开关参数。比如连接地址写成MCP_SERVER_URLtoken 写成MCP_SERVER_TOKEN通过系统环境变量注入而不是硬编码在代码里。另外日志和审批消息里不要打印完整 token。我犯过一次低级错误把带 token 的 endpoint 直接打到了调试日志里后来不得不换了 token。这个教训让我养成了一个习惯凡是打印 server 信息的地方统一做脱敏处理只保留协议头和域名query 参数里的 token 一律用***替代。按需开启的机制本质上是在给 agent 加一道保险。做这个项目的过程中我最深的体会是人工介入不是对模型能力的不信任而是对复杂系统稳定性的敬畏。模型再强也只是在给定的上下文里做概率推理它看不到整个系统全貌更不知道哪些操作会引发连锁反应。给关键步骤加一个“人踩刹车”的环节很多事故就能在发生之前停下来。最后再分享一个小技巧我给每个工具调用都加了一个 trace-id从 agent 规划、网关审批到执行返回全程串联。这个习惯帮我解决过很多疑难问题比如定位“为什么 gating 层没有拦截到这个调用”“这条记录是哪个任务产生的”。按需开启只是一句口号真正落地全靠这些细枝末节的工程细节撑起来。
MCP按需开启的人工介入网关：AI Agent工具安全实战
做 AI Agent 做得越久我对“工具全开”这件事越警惕。给 pi agent 接 MCP 那天我一股脑把浏览器控制、文件读写、命令行执行全都挂上去结果第一次真实跑任务就翻车了——agent 在没经过我同意的情况下把生产环境一个临时目录清了个干净。从那之后我开始认真琢磨能不能做一套按需开启的 MCP 机制让工具默认关闭需要时由人显式介入、授权、再执行这套方案最后就是 pi agent 上这套“人工介入网关”。这篇文章我会从 MCP 协议的理解讲起把带开关的 server 实现、审批流设计、客户端接入和真实环境里踩过的坑全部盘一遍。如果你手上正好有一个基于 LLM 的 agent 项目想让它调用外部工具又不想把控制权完全交出去这篇应该正是你需要的。1. 为什么人工介入是刚需先想清楚工具授权这件事1.1 一次“工具全开”的翻车现场先说那次事故的细节。我的 pi agent 当时跑了大概一周接了两个 MCP server一个 filesystem 工具集一个 shell 工具集。某天我想让它“把临时目录下的缓存文件清理一下”结果 agent 的规划器理解成“清理临时目录”直接调用了delete_tree把整个目录递归删掉了。这个目录里正好有正在跑的构建缓存等于我在没有任何确认的情况下让 AI 替我做了一个破坏性操作。事后复盘问题完全不在模型本身而在于我当时把所有工具都放进了它的可见列表里。这个案例特别典型的地方在于LLM 对用户意图的理解是概率性的它不是确定性程序。同一个“清理一下”在不同上下文里可能被翻译成完全不同的工具调用。工具一旦在 agent 的可见范围内它就会倾向于使用而且它不会主动判断“这个操作是不是太危险了我先问问人”。所以指望模型自觉不如从工程上堵住这个口子。MCP 在这里是一个能力放大器。协议本身让 agent 能够接入的工具数量和复杂度都上了一个台阶但这意味着破坏半径也跟着放大。浏览器自动化能帮你填表也能帮你点错按钮数据库工具能帮你查数据也能帮你删表。能力越大越需要把“谁能调用、什么条件下调用”这件事变成显式的工程约束而不是模型临场发挥的一部分。1.2 一句话拆清楚 MCP 的工作模型MCPModel Context Protocol说白了就是一套 AI 工具调用的标准协议解决的是“模型怎么发现工具、怎么调用工具、工具结果怎么返回给模型”这类问题。你可以把它类比成 USB-C 接口AI 像是电脑MCP server 像是外设协议规定了接口形状、数据格式、握手流程大家按这个标准插上就能用。MCP 的架构核心是客户端-服务器模型分几个关键角色MCP Client运行在 agent 或 AI 应用内部负责跟模型对话、向 server 发起工具发现和调用请求。MCP Server暴露工具Tools、资源Resources、提示词模板Prompts的独立服务可以跑在本地进程stdio 传输也可以跑在远程HTTP/SSE、WebSocket 传输。协议层基于 JSON-RPC 2.0定义了initialize、tools/list、tools/call、resources/read等标准方法。最常用的还是 Tools 原语。agent 启动时会从 server 拉取工具清单模型根据任务需要挑选合适的工具然后通过tools/call发出调用请求server 执行完把结果返回模型。整个调用链就是用户请求 → 模型规划 → 请求工具 → 工具执行 → 结果回传 → 模型整理输出。理解这个模型有什么用因为它决定了“按需开启”可以插在哪几个环节。你既可以在 client 端控制“不把某些工具发给模型”也可以在 server 端拦截“即使模型请求了这个工具也不让它执行”。我的做法是两层都做让 pi agent 的开关机制足够严密。1.3 按需开启要解决的三件事设计按需开启机制之前我给自己定了三个目标后面所有代码都是围绕这三个目标展开的第一默认关闭、显式开启。所有工具默认不出现在 agent 的工具列表里。只有当任务上下文明确需要某个工具时才把它临时挂载上去。这样做除了安全考虑还有个很实际的好处——减少上下文膨胀。每多一个工具模型要处理的 token 就多一堆工具列表越精简模型做决策的速度和准确率都会好一些。第二分级授权、人工介入。不是所有工具都需要人工审批。只读工具可以自动放行写操作工具必须弹确认破坏性工具默认拒绝。人工介入不是“不管三七二十一都问一遍”那样 agent 根本没法用而是像红绿灯一样该放行的放行该减速的减速该停的停。这套分级体系在后面会展开。第三全程审计、可追溯。每次工具调用都要留下记录谁调用的、什么时候、传了什么参数、结果如何、审批人是谁。这一条在出了问题之后价值最大。没有审计翻车了都不知道是哪一步的锅。这三个目标听起来简单落地的时候牵扯到的细节非常多。接下来我会把整套架构和代码一步一步拆开讲。2. 架构与选型给 pi agent 设计可控的 MCP 层2.1 双层开关Client 侧裁剪 Server 侧拦截我最终采用的架构是双层开关。为什么需要双层因为光在 server 端拦截agent 的模型还是会看到所有工具列表它规划的路径里可能包含禁用工具虽然最后会被拒但推理路径已经偏离了光在 client 端裁剪也不行因为一旦 agent 拿到了一个带危险工具的 server 地址它可以绕过 client 直接调用。所以两头都要管。具体到 pi agent 上就是这样的链路pi agent 本体作为 MCP Client启动时读取一份开关配置只加载当前任务需要的那几个 server 和工具每个 server 内部又做了一层守卫工具被调用时会经过一个审批网关网关根据配置决定是直接放行、询问人工还是直接拒绝。双保险的好处是即使 client 端配置出错把不该开的工具暴露了出去server 端也能拦得住。这套架构没有引入额外的编排框架核心就是两个部分一个配置驱动的工具加载器一个审批网关。pi agent 原本的 model 调用逻辑完全不用改只是在工具调用前加了一道检查。这也是我做这个项目比较满意的一点——侵入性很小。2.2 工具分域与权限分级给工具分级是整个方案里最需要想清楚的部分。我按两个维度给工具分类操作是否只读以及影响范围是否可恢复。只读、可恢复的工具放行写操作、可恢复的工具询问写操作、影响大或者不可恢复的工具直接拒绝或要求二次确认。实际操作中我把工具分成了三档对应不同的处理策略安全级别典型场景代表工具默认策略低危查询、读取Fetch 网页抓取、文件只读、数据库 SELECT自动放行中危局部写入、状态变更文件写入、代码执行、浏览器点击询问人工高危删除、批量修改、外部副作用文件删除、DROP/TRUNCATE、邮件发送拒绝或二次确认这个分级不是死的得结合业务上下文调。比如同样是文件写入写到 /tmp 缓存和写到生产配置目录风险天差地别。所以我在分级之外还加了一层路径/域名白名单规则命中了高危路径的工具即使级别是中危也会被自动升级拦截。规则引擎不用写得多复杂一组正则加一个前缀匹配表就够用。分级的作用是让“人工介入”不再是全有全无的选择而是可以根据风险动态调整介入力度。这也呼应了标题里说的“艺术”——介入的时机和程度要拿捏而不是一刀切。2.3 技术选型为什么用 FastMCPMCP server 的实现方式我对比过三条路线直接用官方 TS SDK 手搓、用 Python 的mcp官方 SDK 手写 JSON-RPC 处理、用 FastMCP 这类高层封装。最后选了 FastMCP原因很直接它是纯 Python 的装饰器注册工具的方式最贴合我这种以业务逻辑为主的场景而且它同时支持 stdio、SSE、streamable HTTP 三种传输方式方便我后期把 server 部署到远程。用 FastMCP 写一个工具只需要这样from fastmcp import FastMCP mcp FastMCP(pi-gateway) mcp.tool() def read_config(path: str) - str: 读取配置文件内容只读操作 with open(path, r, encodingutf-8) as f: return f.read()不用自己处理 JSON-RPC 的消息格式框架把协议的细节都包掉了。选 streamable HTTP 作为默认传输方式是因为它比 stdio 更适合跨进程部署pi agent 和 MCP server 可以不在同一台机器上agent 跑在服务器、工具服务跑在开发机上这种拓扑都能支持。选型的时候也考虑过用一个 MCP server 汇聚所有工具还是每种工具一个 server。最后选了后者按工具域拆成多个 server一个 filesystem server、一个 browser server、一个 database server。这样每个 server 的权限边界更清晰只暴露最小功能集出问题时隔离性也好。工具多了以后运维上要管的进程数会多一点但换来的是清晰的安全边界这笔账是划算的。3. 代码落地实现一个带按需开关的 MCP Server3.1 初始化项目与配置文件确定架构后先把项目骨架搭起来。我习惯用这样的目录结构pi-agent-gateway/ ├── pyproject.toml ├── config/ │ └── tools_config.yaml ├── src/ │ ├── server.py # FastMCP 入口 │ ├── gateway.py # 审批网关 │ └── tools/ │ ├── file_tools.py │ └── web_tools.py ├── logs/ │ └── audit.logpyproject.toml里依赖很简单核心就是fastmcp和pyyaml。如果你的环境还没装直接pip install fastmcp pyyaml就行。FastMCP 的版本演进比较快我写这篇文章时用的 2.x 版本接口上装饰器注册和mcp.run()这套是稳定的。配置文件是整条链路的“总开关”。我把所有工具的状态都收敛到一个 YAML 文件里这样改配置不用动代码审核变更也方便。核心的内容是每个工具的启用状态、安全级别和审批模式。# config/tools_config.yaml server: name: pi-agent-gateway transport: streamable-http tools: read_file: enabled: true safety: low mode: auto fetch_url: enabled: true safety: low mode: auto write_file: enabled: false safety: medium mode: ask delete_file: enabled: false safety: high mode: deny approval: timeout_seconds: 30 default_mode: ask这个配置文件承载了整个按需开启的核心逻辑enabled控制工具是否注册到 server 上未注册的工具即使被调用也会返回 not foundmode分成auto、ask、deny三档分别对应自动放行、询问人工和直接拒绝。后面 server 启动时会逐条读这个配置来动态注册工具。3.2 配置驱动的动态工具注册有了配置之后关键就是怎么让 server 按配置去注册工具而不是把所有工具写死在代码里。这里我用了一个很朴素的“注册器”模式定义一组工具实现函数启动时遍历配置只把enabled: true的工具挂到 FastMCP 实例上。拿文件工具举例核心代码长这样# src/server.py import yaml from fastmcp import FastMCP from tools import file_tools, web_tools mcp FastMCP(pi-agent-gateway) # 工具注册表名称 - (实现函数, 安全级别) TOOL_REGISTRY { read_file: (file_tools.read_file, low), write_file: (file_tools.write_file, medium), delete_file: (file_tools.delete_file, high), fetch_url: (web_tools.fetch_url, low), } def load_tools_from_config(config_path: str): with open(config_path, r, encodingutf-8) as f: config yaml.safe_load(f) for name, tool_config in config[tools].items(): if not tool_config.get(enabled, False): continue func, _ TOOL_REGISTRY[name] # 把工具函数注册到 mcp 实例上 mcp.tool()(func) print(f[gateway] tool registered: {name})这段代码解决了“按需开启”的第一层问题配置里没启用的工具压根不会出现在tools/list返回结果里模型看不到、也用不了。想临时开一个工具改一行 YAML 重启进程就行不用改任何业务代码。线上紧急恢复现场的时候这个能力非常救命。这里有一个容易踩的点动态注册工具后FastMCP 的 schema 生成是在注册时确定的如果你的工具函数有复杂的 Pydantic 模型入参注册前一定要保证模型定义完整不然生成的 JSON Schema 会不完整客户端拉取的时候可能直接报错。我后来统一规范了工具入参能不用复杂嵌套模型就不用全部用基本类型加Field(description...)这一个改动让工具发现的稳定性好了很多。3.3 人工审批网关三种模式与超时策略工具注册解决了“能不能看到”的问题审批网关解决“能不能执行”的问题。网关的核心是一个ApprovalGateway类每次tools/call进来的时候先走一遍检查逻辑。实现思路是这样的# src/gateway.py import enum import time import logging from typing import Any logger logging.getLogger(gateway) class ApprovalMode(enum.Enum): AUTO auto ASK ask DENY deny class ApprovalGateway: def __init__(self, mode_map: dict[str, dict], timeout_seconds: int 30): self.mode_map mode_map self.timeout_seconds timeout_seconds def check(self, tool_name: str, params: dict[str, Any]) - tuple[bool, str]: 返回 (是否允许执行, 原因/审批信息) if tool_name not in self.mode_map: return False, ftool {tool_name} is not registered mode ApprovalMode(self.mode_map[tool_name][mode]) if mode ApprovalMode.DENY: return False, tool is denied by policy if mode ApprovalMode.AUTO: return True, auto approved # ASK 模式需要人工介入 return self._ask_human(tool_name, params) def _ask_human(self, tool_name: str, params: dict[str, Any]) - tuple[bool, str]: # 把审批请求发送到外部人工端例如 Slack 机器人、Webhook 或终端提示 # 这里用一个阻塞等待的实现支持超时 decision self._request_approval(tool_name, params) if decision is False: return False, rejected by human return True, approved by human这个_request_approval的实现可以根据你的环境选择本地开发可以直接用input()在终端弹提示线上可以用 Webhook 推到企业微信或者 Slack让值班的人手机确认。我没有把这部分写死保留了一个可插拔的接口。超时策略是这里最容易被忽略的细节。我一开始没设超时审批请求发出去之后人一直没回应agent 的整个会话就挂在那里了。后来加了两层保护第一层是网关层的超时超过 30 秒默认拒绝执行保证调用链不会无限期阻塞第二层是客户端层的超时这个后面会讲到。安全侧的策略是“超时等同于拒绝”宁可让任务失败也不能在没人确认的情况下继续执行。这个原则我建议所有做人工介入机制的团队都采纳。4. 接入客户端与实战演示让按需开启真正跑起来4.1 客户端连接与工具发现server 端写完之后接下来就是让 pi agent 作为 MCP Client 去连它。用 FastMCP 跑起的 server 默认会暴露一个 HTTP endpoint客户端通过 streamable HTTP 传输去连接。客户端的连接代码大致如下# client_example.py import asyncio from mcp import ClientSession from mcp.client.streamable_http import streamablehttp_client async def main(): endpoint http://localhost:8000/mcp async with streamablehttp_client(endpoint) as (read, write): async with ClientSession(read, write) as session: await session.initialize() tools await session.list_tools() print(Available tools:, [t.name for t in tools.tools]) result await session.call_tool( read_file, arguments{path: /tmp/example.txt} ) print(Tool result:, result) asyncio.run(main())这里有一个很容易搞混的点list_tools拿到的是当前 server 端已经加载的工具列表。因为 server 是按配置注册的所以你在这个列表里看到的就是真正可用的。如果某个工具没启用它根本不会出现在这个列表里。这也是我实测下来验证双层开关最直接的方式——先 list 一下看工具在不在。pi agent 本身的模型调用逻辑不需要大改只需要在原来的“模型选工具”和“执行工具”之间插入这个 MCP Client。模型输出一个工具调用意图pi agent 把它翻译成一次call_tool拿到结果再拼回去给模型。整个介入逻辑对模型完全透明模型不需要知道背后有没有人审批过。4.2 一个完整场景的完整时间线用实际的场景把整个流程串一遍。假设用户给 pi agent 发了一条指令“把 /tmp/pi-agent-cache 下面的 .tmp 文件清理掉保留 .log 文件。”这条指令会触发一连串动作。先看客户端这边pi agent 从工具列表里发现可用工具只有read_file和fetch_url因为write_file、delete_file默认关闭它可能觉得工具不够用会向用户提示“需要文件清理类工具”。这时候就是按需开启发挥作用的地方。你可以让用户直接在对话里确认“开启删除工具”或者由 agent 根据任务自动发起工具开启请求。开启之后server 端重新加载配置delete_file出现在工具列表里。agent 规划调用delete_file网关检查到它的 mode 是ask于是发起人工审批。审批通过后命令才真正执行。完整时间线是这样的用户: 清理 /tmp/pi-agent-cache 下的 .tmp 文件 → pi agent: 检查工具列表发现 delete_file 未启用 → 用户: 确认开启删除工具 → server: 重载配置delete_file 注册成功 → pi agent: 规划调用 delete_file(/tmp/pi-agent-cache, *.tmp) → 网关: modeask发起人工审批请求 → 人工端: 确认允许执行 → server: 执行删除返回结果给 agent → pi agent: 汇总清理结果给用户这条时间线里人工介入出现了两个地方一次是用户层面的“开启工具”一次是执行层面的“审批确认”。这就是我说的“按需开启”的完整形态——不是启动时全开而是在任务的每个关键节点由人决定要不要放行。你可以看到这套机制对用户来说并不繁琐只读操作完全不打扰破坏性操作也只多一步确认。4.3 运营侧的三档开关策略网关机制落地之后我经验里最值钱的部分其实是运营策略——什么任务跑什么模式。单任务临时执行我一般全程 auto 加 ask 混合批量任务或者定时任务我倾向于把 ask 模式关掉改成 deny 加白名单。这不是偷懒而是批量任务中如果每个工具都弹审批审批人很容易疲劳疲劳之后就会随手点同意那审批机制就形同虚设了。我目前的运营策略是三档场景策略说明探索性任务auto ask只读自动放行写操作人工确认正式执行任务ask 白名单高危路径白名单自动放行其他全部人工确认定时批处理deny 最小工具集只开白名单工具其余全部拒绝三档策略的核心是同一个配置模板只是参数不同。这正好是配置驱动模式带来的好处改 YAML 就能切换策略不用重新部署代码。我在实际运营中切换过很多次策略稳定性还是不错的。5. 常见问题与排查实录5.1 工具调用超时agent 干等第一次上线人工审批网关很快就发现问题agent 调用一个 ask 模式的工具审批请求发出去了人迟迟没有处理然后 agent 那边就报Tool call timed out。排查后发现是客户端侧的响应超时设置得太短默认只有 10 秒而人工审批从看到消息到做出决定正常都要 20 秒以上。解决方案是分层设置超时客户端连接层设 60 秒审批网关层设 30 秒两层超时独立工作。这样即使审批耗时较长agent 会话也不会被轻易打断反过来如果审批真的超时网关会先返回拒绝客户端收到明确结果不会出现一边在等、一边已经断开的情况。另一个连接相关的问题streamable HTTP 的 endpoint 如果长时间空闲连接会被中间层的代理或者负载均衡断开。我之前遇到过一个反复出现的Connection reset报错排查下来就是空闲超时问题。解决办法是客户端加心跳请求并且把错误处理改成重试逻辑工具调用失败时自动重新建立连接再试一次。5.2 审批阻塞导致会话卡死审批阻塞比超时更隐蔽。有一次线上任务跑着跑着整个 agent 会话不动了日志里没有任何报错就是静默挂起。查了半天发现是审批网关的_ask_human实现里有问题——审批请求发出去了但回调处理线程和主事件循环之间没有做好同步导致主循环一直阻塞在等待队列上既没有超时兜底也没有报错。这个坑的教训是要用异步化设计。现在的实现里审批请求发出后立刻返回一个 pending 状态主循环不会被阻塞审批结果通过回调或者轮询写入结果队列调用方拿到结果后再继续执行。这样即使人工端迟迟不回应agent 会话也只会停在等待状态不会整个崩溃。排查这类问题我有个习惯就是看日志里的 trace-id。每个工具调用我都带上一个唯一标识从客户端发起、网关审批、到执行完成全过程打同一个 trace-id 的日志。出问题的时候按 trace-id 一搜整个调用链一目了然定位快了不止一倍。5.3 token 与认证的保管问题最后说一个安全上的细节。MCP server 如果走远程传输endpoint 和 token 是像钥匙一样的东西。我的项目里把 endpoint、token 这类敏感信息全部放到环境变量里配置文件只放相对安全的开关参数。比如连接地址写成MCP_SERVER_URLtoken 写成MCP_SERVER_TOKEN通过系统环境变量注入而不是硬编码在代码里。另外日志和审批消息里不要打印完整 token。我犯过一次低级错误把带 token 的 endpoint 直接打到了调试日志里后来不得不换了 token。这个教训让我养成了一个习惯凡是打印 server 信息的地方统一做脱敏处理只保留协议头和域名query 参数里的 token 一律用***替代。按需开启的机制本质上是在给 agent 加一道保险。做这个项目的过程中我最深的体会是人工介入不是对模型能力的不信任而是对复杂系统稳定性的敬畏。模型再强也只是在给定的上下文里做概率推理它看不到整个系统全貌更不知道哪些操作会引发连锁反应。给关键步骤加一个“人踩刹车”的环节很多事故就能在发生之前停下来。最后再分享一个小技巧我给每个工具调用都加了一个 trace-id从 agent 规划、网关审批到执行返回全程串联。这个习惯帮我解决过很多疑难问题比如定位“为什么 gating 层没有拦截到这个调用”“这条记录是哪个任务产生的”。按需开启只是一句口号真正落地全靠这些细枝末节的工程细节撑起来。
