企业官网建设流程全解析

做 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 的工作模型

MCP（Model 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 工具分域与权限分级

给工具分级是整个方案里最需要想清楚的部分。我按两个维度给工具分类：操作是否只读，以及影响范围是否可恢复。只读、可恢复的工具放行；写操作、可恢复的工具询问；写操作、影响大或者不可恢复的工具直接拒绝或要求二次确认。

实际操作中我把工具分成了三档，对应不同的处理策略：

这个分级不是死的，得结合业务上下文调。比如同样是文件写入，写到 /tmp 缓存和写到生产配置目录，风险天差地别。所以我在分级之外还加了一层路径/域名白名单规则，命中了高危路径的工具即使级别是中危也会被自动升级拦截。规则引擎不用写得多复杂，一组正则加一个前缀匹配表就够用。

分级的作用是让“人工介入”不再是全有全无的选择，而是可以根据风险动态调整介入力度。这也呼应了标题里说的“艺术”——介入的时机和程度要拿捏，而不是一刀切。

2.3 技术选型：为什么用 FastMCP

MCP 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", encoding="utf-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 Server

3.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.log

pyproject.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 found；mode分成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", encoding="utf-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, f"tool {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") → 网关: mode=ask，发起人工审批请求 → 人工端: 确认允许执行 → server: 执行删除，返回结果给 agent → pi agent: 汇总清理结果给用户

这条时间线里，人工介入出现了两个地方：一次是用户层面的“开启工具”，一次是执行层面的“审批确认”。这就是我说的“按需开启”的完整形态——不是启动时全开，而是在任务的每个关键节点，由人决定要不要放行。你可以看到，这套机制对用户来说并不繁琐，只读操作完全不打扰，破坏性操作也只多一步确认。

4.3 运营侧的三档开关策略

网关机制落地之后，我经验里最值钱的部分其实是运营策略——什么任务跑什么模式。单任务临时执行，我一般全程 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_URL，token 写成MCP_SERVER_TOKEN，通过系统环境变量注入，而不是硬编码在代码里。

另外，日志和审批消息里不要打印完整 token。我犯过一次低级错误，把带 token 的 endpoint 直接打到了调试日志里，后来不得不换了 token。这个教训让我养成了一个习惯：凡是打印 server 信息的地方，统一做脱敏处理，只保留协议头和域名，query 参数里的 token 一律用***替代。

按需开启的机制本质上是在给 agent 加一道保险。做这个项目的过程中，我最深的体会是：人工介入不是对模型能力的不信任，而是对复杂系统稳定性的敬畏。模型再强，也只是在给定的上下文里做概率推理，它看不到整个系统全貌，更不知道哪些操作会引发连锁反应。给关键步骤加一个“人踩刹车”的环节，很多事故就能在发生之前停下来。

最后再分享一个小技巧：我给每个工具调用都加了一个 trace-id，从 agent 规划、网关审批到执行返回全程串联。这个习惯帮我解决过很多疑难问题，比如定位“为什么 gating 层没有拦截到这个调用”“这条记录是哪个任务产生的”。按需开启只是一句口号，真正落地全靠这些细枝末节的工程细节撑起来。