s16：MCP Plugin 工具插件
s16：MCP Plugin（Model Context Protocol 工具插件）
核心问题
Lead 和队友目前只能调用直接写在 code.py 里的内置工具。要接 Jira、部署平台或知识库，Harness 得为每个外部系统单独写工具定义和调用逻辑——外部系统一改，课程代码也得跟着改。
MCP（Model Context Protocol）解决的就是这个问题：用统一的发现与调用协议，在运行时连接外部服务，把它们提供的工具动态加进工具池。
1. MCP 协议基础：tools/list 和 tools/call
MCP 协议规范定义了两个核心 RPC 方法（基于 JSON-RPC 2.0）：
tools/list — 发现工具
客户端请求服务器列出所有可用工具：
// 请求
{
"jsonrpc": "2.0",
"id": 1,
"method": "tools/list",
"params": {}
}
// 响应
{
"jsonrpc": "2.0",
"id": 1,
"result": {
"tools": [
{
"name": "search",
"description": "Search the documentation.",
"inputSchema": {
"type": "object",
"properties": {
"query": {"type": "string"}
},
"required": ["query"]
}
}
]
}
}
tools/call — 调用工具
客户端请求服务器执行某个工具：
// 请求
{
"jsonrpc": "2.0",
"id": 2,
"method": "tools/call",
"params": {
"name": "search",
"arguments": {"query": "hello"}
}
}
// 响应
{
"jsonrpc": "2.0",
"id": 2,
"result": {
"content": [{"type": "text", "text": "Found 3 results..."}]
}
}
2. 真实 MCP 部署架构
本地部署（stdio）
┌─────────────────────────────────────────────────────┐
│ 你的电脑 (Host) │
│ │
│ ┌──────────────┐ ┌──────────────────┐ │
│ │ Agent Harness │ │ MCP Server │ │
│ │ │ │ (docs) │ │
│ │ MCPClient ──┼────────┤ │ │
│ │ │ stdio/ │ search() │ │
│ │ │ WebSocket │ │
│ └──────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────┘
远程部署（Streamable HTTP / WebSocket）
┌─────────────────────────────────────────────────────┐
│ 你的电脑 (Host) │
│ │
│ ┌──────────────┐ ┌──────────────────┐ │
│ │ Agent Harness │ │ 远程服务器 │ │
│ │ │ │ (另一台电脑/云) │ │
│ │ MCPClient ──┼────────┤ │ │
│ │ │ HTTPS/ │ MCP Server │ │
│ │ │ WebSocket │ │
│ └──────────────┘ └──────────────────┘ │
└─────────────────────────────────────────────────────┘
三种通信方式对比
| 方式 | 适用场景 | 传输协议 | 特点 |
|---|---|---|---|
| stdio | 本地进程通信（同一台电脑） | 标准输入/输出管道 | Server 作为子进程运行，适合本地工具（文件系统、数据库等）；最简单、最低延迟 |
| Streamable HTTP（新版推荐） | 远程 HTTP 通信 | HTTP POST + SSE | Server 运行在远程服务器，客户端通过 HTTP 请求通信；支持流式响应 |
| WebSocket | 双向实时通信 | WebSocket 协议 | 全双工通信，适合需要服务端主动推送的场景 |
SSE vs WebSocket 区别：SSE（Server-Sent Events）是单向的（服务器→客户端），适合服务器推送更新；WebSocket 是双向的，客户端和服务端都能随时发消息。MCP 的 Streamable HTTP 模式使用 SSE 来实现服务端的流式响应推送。
3. 真实 MCP Client 实现
原课程代码的问题
原课程中的 _mock_server_docs() 是一个模拟实现——它直接在进程内创建 MCPClient 并注册假的 handler（lambda 函数返回硬编码字符串）。这在教学中便于理解接口，但跳过了真实的通信过程。
下面补充真实的 MCP 连接和通信代码。
真实的连接过程
# ============================================================
# 真实 MCP Client：建立连接、发现工具、调用工具
# ============================================================
import subprocess
import json
import asyncio
from typing import Any
class RealMCPClient:
"""
真实的 MCP 客户端，负责与 MCP Server 建立通信连接。
支持两种传输方式：
- stdio：本地子进程通信（通过 stdin/stdout 传输 JSON-RPC 消息）
- Streamable HTTP：远程网络通信（通过 HTTP POST + SSE）
"""
def __init__(self, server_name: str, transport: str = "stdio",
command_or_url: str | list[str] | None = None):
"""
初始化 MCP 客户端
Args:
server_name: 服务器名称（用于日志和标识）
transport: 传输方式，"stdio" 或 "streamable-http"
command_or_url:
- stdio 模式：启动 server 子进程的命令（如 ["npx", "-y", "@modelcontextprotocol/server-filesystem", "/path"]
- HTTP 模式：server 的 URL（如 "https://my-mcp-server.example.com/mcp"）
"""
self.server_name = server_name
self.transport = transport
self.command_or_url = command_or_url
self.process: subprocess.Popen | None = None # stdio 模式的子进程
self.session_id: str | None = None # HTTP 模式的会话 ID
self._request_id = 0 # JSON-RPC 请求 ID 计数器
self.tools: list[dict] = [] # 从 server 发现的工具列表
# ---------- 连接管理 ----------
async def connect(self) -> None:
"""建立与 MCP Server 的连接"""
if self.transport == "stdio":
await self._connect_stdio()
elif self.transport == "streamable-http":
await self._connect_http()
else:
raise ValueError(f"Unsupported transport: {self.transport}")
# 连接成功后，自动发现可用工具
await self._discover_tools()
async def _connect_stdio(self) -> None:
"""
stdio 模式连接：启动 server 子进程，通过 stdin/stdout 通信
这是本地 MCP 最常用的方式。server 作为子进程运行，
客户端通过标准输入写 JSON-RPC 请求，从标准输出读响应。
"""
if isinstance(self.command_or_url, str):
cmd = self.command_or_url.split()
else:
cmd = self.command_or_url or []
# 启动子进程
# stdin=PIPE: 我们要往 server 写请求数据
# stdout=PIPE: 我们要从 server 读响应数据
# stderr=PIPE: 方便调试时查看 server 日志
self.process = subprocess.Popen(
cmd,
stdin=subprocess.PIPE,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=False, # 使用二进制模式，避免编码问题
bufsize=0, # 无缓冲，实时通信
)
print(f"[MCP] Connected to '{self.server_name}' via stdio (PID: {self.process.pid})")
async def _connect_http(self) -> None:
"""
Streamable HTTP 模式连接：通过网络请求与远程 server 通信
流程：
1. POST /mcp 发送 initialize 请求
2. Server 返回 session ID 和能力声明
3. 后续请求携带 session ID
4. Server 通过 SSE 推送通知/流式响应
"""
import aiohttp
base_url = self.command_or_url
async with aiohttp.ClientSession() as session:
self._http_session = session
# 1. 发送 initialize 握手
init_response = await self._send_http_request(
base_url, "initialize",
{
"protocolVersion": "2024-11-05",
"capabilities": {},
"clientInfo": {"name": "harness-agent", "version": "1.0.0"},
}
)
# 2. 获取 session ID（用于后续请求）
self.session_id = init_response.get("result", {}).get("session_id")
# 3. 发送 initialized 通知（握手完成）
await self._send_http_notification(base_url, "notifications/initialized")
print(f"[MCP] Connected to '{self.server_name}' via HTTP (session: {self.session_id})")
# ---------- JSON-RPC 通信 ----------
def _next_id(self) -> int:
"""生成下一个 JSON-RPC 请求 ID"""
self._request_id += 1
return self._request_id
async def _send_request(self, method: str, params: dict | None = None) -> dict:
"""
发送 JSON-RPC 请求并等待响应（stdio 模式）
Args:
method: RPC 方法名（如 "tools/list"、"tools/call"）
params: 方法参数
Returns:
服务器的 JSON-RPC 响应（解析后的 dict）
"""
import asyncio
request = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": method,
"params": params or {},
}
message = json.dumps(request) + "\n"
# 写入 server 的 stdin
assert self.process and self.process.stdin
self.process.stdin.write(message.encode("utf-8"))
await self.process.stdin.drain()
# 从 server 的 stdout 读取响应
assert self.process.stdout
raw_response = await asyncio.get_event_loop().run_in_executor(
None, self.process.stdout.readline
)
response = json.loads(raw_response.decode("utf-8"))
if "error" in response:
raise MCPError(response["error"])
return response
async def _send_http_request(self, base_url: str, method: str,
params: dict | None = None) -> dict:
"""发送 HTTP JSON-RPC 请求（Streamable HTTP 模式）"""
url = f"{base_url.rstrip('/')}/"
headers = {"Content-Type": "application/json"}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
body = {
"jsonrpc": "2.0",
"id": self._next_id(),
"method": method,
"params": params or {},
}
async with self._http_session.post(url, json=body, headers=headers) as resp:
return await resp.json()
async def _send_http_notification(self, base_url: str, method: str,
params: dict | None = None) -> None:
"""发送 HTTP JSON-RPC 通知（不需要响应）"""
url = f"{base_url.rstrip('/')}/"
headers = {"Content-Type": "application/json"}
if self.session_id:
headers["Mcp-Session-Id"] = self.session_id
body = {
"jsonrpc": "2.0",
"method": method,
"params": params or {},
}
async with self._http_session.post(url, json=body, headers=headers) as resp:
pass # 通知不需要等待响应
# ---------- 工具发现 ----------
async def _discover_tools(self) -> None:
"""连接后自动调用 tools/list 发现可用工具"""
response = await self._send_request("tools/list")
self.tools = response.get("result", {}).get("tools", [])
tool_names = [t["name"] for t in self.tools]
print(f"[MCP] Discovered {len(self.tools)} tools on '{self.server_name}': {tool_names}")
def list_tools(self) -> list[dict]:
"""返回已发现的工具定义列表"""
return self.tools
# ---------- 工具调用 ----------
async def call_tool(self, tool_name: str, arguments: dict) -> str:
"""
调用 MCP Server 上的工具
Args:
tool_name: 工具名称（必须是 tools/list 返回过的名字）
arguments: 工具参数（符合工具的 inputSchema 定义）
Returns:
工具执行的文本结果
"""
response = await self._send_request("tools/call", {
"name": tool_name,
"arguments": arguments,
})
# 解析 content 数组，提取文本内容
contents = response.get("result", {}).get("content", [])
text_parts = []
for item in contents:
if item.get("type") == "text":
text_parts.append(item.get("text", ""))
elif item.get("type") == "image":
text_parts.append("[image content]")
else:
text_parts.append(json.dumps(item))
return "\n".join(text_parts)
# ---------- 生命周期 ----------
async def close(self) -> None:
"""关闭连接，清理资源"""
if self.process:
# stdio 模式：关闭子进程的 stdin（发 EOF），然后等待退出
assert self.process.stdin
self.process.stdin.close()
try:
self.process.wait(timeout=5)
except subprocess.TimeoutExpired:
self.process.kill()
print(f"[MCP] Disconnected from '{self.server_name}' (stdio)")
if hasattr(self, '_http_session'):
# HTTP 模式：关闭会话
await self._http_session.close()
print(f"[MCP] Disconnected from '{self.server_name}' (HTTP)")
class MCPError(Exception):
"""MCP 协议级错误"""
pass
# ============================================================
# 使用示例
# ============================================================
async def connect_to_real_mcp_server():
"""
真实使用示例：连接到一个 MCP 文件系统 server
这个 server 由 @modelcontextprotocol/server-physics 或类似项目提供，
通过 npx 启动，提供文件系统的读写能力。
"""
# 示例 1：stdio 模式连接本地 server
client_stdio = RealMCPClient(
server_name="filesystem",
transport="stdio",
command_or_url=["npx", "-y", "@modelcontextprotocol/server-filesystem", "/tmp/project"],
)
await client_stdio.connect()
# 查看发现了哪些工具
tools = client_stdio.list_tools()
for t in tools:
print(f" - {t['name']}: {t.get('description', 'no description')}")
# 调用工具
result = await client_stdio.call_tool("read_file", {"path": "/tmp/project/README.md"})
print(f"Result: {result[:200]}")
await client_stdio.close()
# 同步包装器（供 harness 内部使用）
class SyncMCPClient:
"""
同步封装的 MCP Client，适配 harness 的同步执行模型
内部维护一个事件循环来运行异步的 RealMCPClient 操作。
这使得 MCP 工具调用可以像普通函数一样同步使用。
"""
def __init__(self, name: str):
self.name = name
self._async_client: RealMCPClient | None = None
self.tools: list[dict] = []
self._handlers: dict[str, callable] = {} # 工具名 → 处理函数（同步闭包）
def connect(self, command_or_url: str, transport: str = "stdio") -> list[dict]:
"""
同步连接方法：内部启动事件循环执行异步连接
Args:
command_or_url: 启动命令或 URL
transport: 传输方式
Returns:
发现的工具列表
"""
import asyncio
self._async_client = RealMCPClient(
server_name=self.name,
transport=transport,
command_or_url=command_or_url,
)
# 在新的事件循环中运行异步连接
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
try:
loop.run_until_complete(self._async_client.connect())
finally:
loop.close()
self.tools = self._async_client.list_tools()
return self.tools
def register_handlers(self, handlers: dict[str, callable]) -> None:
"""
注册工具处理器映射
每个 handler 是一个同步函数，它内部会调用异步的 call_tool。
这里用闭包把异步调用包装成同步接口。
"""
self._handlers = handlers
def call_tool(self, tool_name: str, args: dict) -> str:
"""
同步调用工具
内部通过新事件循环执行异步调用，对外暴露同步接口。
所有异常都被捕获并转为错误字符串返回（不崩溃）。
"""
if not self._async_client:
return f"MCP error: client '{self.name}' not connected"
import asyncio
loop = asyncio.new_event_loop()
try:
result = loop.run_until_complete(
self._async_client.call_tool(tool_name, args)
)
return result
except Exception as exc:
return f"MCP error: {type(exc).__name__}: {exc}"
finally:
loop.close()
真实 vs Mock 对比
| 维度 | Mock（原课程代码） | 真实实现（上方代码） |
|---|---|---|
| Server 进程 | 不存在，全是假数据 | 真实子进程（stdio）或远程服务（HTTP） |
| 通信方式 | 直接函数调用（无通信） | JSON-RPC over stdin/stdout 或 HTTP |
| 工具发现 | 硬编码 tool_defs 列表 |
调用 tools/list RPC 动态获取 |
| 工具调用 | lambda 返回假字符串 | 调用 tools/call RPC 真正执行 |
| 错误处理 | 无真实错误 | 网络超时、进程崩溃、协议错误等 |
| 资源管理 | 无需清理 | 需要正确关闭子进程/HTTP 会话 |
4. MCPClient 封装（Harness 内部使用的接口）
上面的 RealMCPClient / SyncMCPClient 负责底层通信。Harness 内部在此基础上再做一层封装，统一管理工具注册和调用：
MCPClient 类
代表"一个已连接的外部 server"，保存 server 返回的工具定义列表和每个工具对应的 handler。
核心方法：
register(tool_defs, handlers) — 注册工具
把 server 给的工具定义和 handler 收进来；并做三道校验：
- 名字非空：每个工具必须有非空字符串名称
- 无重名：同一个 server 内工具名不能重复
- handler 完整：每个工具定义都必须有对应的处理函数
任何一项不满足直接抛 ValueError，绝不静默忽略。
def register(self, tool_defs: list[dict], handlers: dict[str, callable]):
names = [tool.get("name") for tool in tool_defs]
# 校验 1：名称非空
if any(not isinstance(name, str) or not name for name in names):
raise ValueError("Every MCP tool needs a non-empty name")
# 校验 2：无重复名称
if len(set(names)) != len(names):
raise ValueError(f"Duplicate MCP tool name on server {self.name!r}")
# 校验 3：每个工具都有对应 handler
missing = [name for name in names if name not in handlers]
if missing:
raise ValueError(f"Missing MCP handlers: {', '.join(missing)}")
self.tools = list(tool_defs)
self._handlers = dict(handlers)
call_tool(tool_name, args) — 调用工具（统一异常处理）
按名字找到 handler 执行；找不到或执行报错，都返回错误字符串，而不是让整个 Agent Loop 崩掉。这正是**“错误留在工具边界内”**的关键设计。
def call_tool(self, tool_name: str, args: dict) -> str:
handler = self._handlers.get(tool_name)
if not handler:
return f"MCP error: unknown tool '{tool_name}'"
try:
return str(handler(**args))
except Exception as exc:
return f"MCP error: {type(exc).__name__}: {exc}"
5. 工具名规范化
为什么需要规范化？
- Anthropic API 对工具名有限制：只允许字母、数字、下划线、连字符
- MCP 服务器来自第三方：工具名可能包含特殊字符（如
/、.、空格等） - 多 server 可能命名冲突：不同 server 可能有同名工具
规范化规则
原始工具名经过处理后变为：mcp__{server}__{tool}
例如：
- server=
docs, tool=search→mcp__docs__search - server=
file-system, tool=read_file→mcp__file-system__read_file - server=
db, tool=query/table→mcp__db__query_table（特殊字符被替换）
工具定义结构
每个 MCP 工具的定义包含以下字段：
| 字段 | 类型 | 说明 |
|---|---|---|
name |
string |
工具名（在 server 内唯一） |
description |
string |
工具描述（供 LLM 理解用途） |
inputSchema |
JSON Schema |
输入参数定义（类型、属性、必填项等） |
annotations |
object |
工具属性提示：readOnlyHint=只读，destructiveHint=有破坏性 |
6. connect_mcp — 连接并发现工具
连接流程（4 步判断）：
| 步骤 | 条件 | 行为 |
|---|---|---|
| 1 | 已连接过 | 返回"已连接"，不重复连 |
| 2 | 配置中找不到 server 名 | 报错：“未知 server” |
| 3 | 找到了 | 调工厂函数造出 server client，存进全局 mcp_clients 字典 |
| 4 | 连接成功 | 打印发现的工具名，返回给模型 |
# 伪代码示意
def connect_mcp(server_name: str) -> list[dict]:
"""连接 MCP server 并发现工具"""
# 1. 检查是否已连接
if server_name in mcp_clients:
print(f"[MCP] Already connected to '{server_name}'")
return mcp_clients[server_name].tools
# 2. 查找 server 配置
if server_name not in SERVER_CONFIGS:
raise ValueError(f"Unknown MCP server: {server_name}")
# 3. 创建 client 并连接
config = SERVER_CONFIGS[server_name]
client = SyncMCPClient(name=server_name)
tools = client.connect(
command_or_url=config["command"], # 或 config["url"]
transport=config.get("transport", "stdio"),
)
# 4. 存入全局字典
mcp_clients[server_name] = client
print(f"[MCP] Connected to '{server_name}', found {len(tools)} tools")
return tools
7. assemble_tool_pool — 工具池组装
做什么
每轮循环调用模型前，将内置工具和所有已连接的 MCP 工具合并，拼成一份全新的 (tools, handlers) 返回给模型。
组装过程中的校验链
遍历每个已连接的 MCP server
│
├─→ 规范化 server 名和工具名
│ （特殊字符替换、长度限制 64 字符）
│
├─→ 拼出前缀名 mcp__{server}__{tool}
│
├─→ 长度检查（>64 字符报错）
│
├─→ 冲突检查（已有同名工具报错）
│
├─→ Schema 验证（inputSchema 必须是合法的 object）
│
├─→ 加入 tools 列表
│
├─→ 闭包绑定 handler（延迟绑定 server 和 tool 引用）
│
└─→ 设置授权策略（默认 confirm，可按 server+tool 覆盖）
代码实现
def assemble_tool_pool() -> tuple[list[dict], dict[str, callable]]:
"""组装完整的工具池：内置工具 + 所有已连接的 MCP 工具"""
global mcp_tool_policies
# 1. 从内置工具开始
tools = list(BUILTIN_TOOLS)
handlers = dict(BUILTIN_HANDLERS)
policies: dict[str, str] = {}
origins = {
tool["name"]: f"built-in tool {tool['name']!r}"
for tool in tools
}
# 2. 遍历每个已连接的 MCP server
for server_name, server in mcp_clients.items():
safe_server = normalize_mcp_name(server_name)
for tool_def in server.tools:
raw_name = tool_def["name"]
safe_tool = normalize_mcp_name(raw_name)
# 拼出前缀名
prefixed = f"mcp__{safe_server}__{safe_tool}"
# 长度检查（API 限制）
if len(prefixed) > 64:
raise ValueError(
f"MCP tool name is longer than 64 characters: {prefixed}"
)
# 冲突检查
origin = f"MCP tool {server_name!r}/{raw_name!r}"
if prefixed in origins:
raise ValueError(
f"MCP tool name collision after normalization: "
f"{prefixed!r} maps both {origins[prefixed]} and {origin}"
)
# Schema 验证
schema = tool_def.get("inputSchema", {})
if not isinstance(schema, dict) or schema.get("type", "object") != "object":
raise ValueError(f"Invalid input schema for {origin}")
origins[prefixed] = origin
# 加入工具定义
tools.append({
"name": prefixed,
"description": tool_def.get("description", ""),
"input_schema": schema,
})
# 闭包绑定：延迟绑定 server 和 tool 引用
# （避免循环引用和 late binding 问题）
handlers[prefixed] = (
lambda *, client=server, tool=raw_name, **kwargs:
client.call_tool(tool, kwargs)
)
# 授权策略：默认 confirm，可按 (server, tool) 覆盖
policies[prefixed] = MCP_HOST_POLICY.get(
(server_name, raw_name), "confirm"
)
mcp_tool_policies = policies
return tools, handlers
8. 权限检查
做什么
决定外部工具能不能直接跑，还是先问用户。授权策略来自宿主配置。
权限钩子
permission_hook 挂在 PreToolUse hook 上，在每次工具调用前执行：
def permission_hook(block):
# ... 内置工具的检查（省略） ...
# MCP 工具的检查：对名字以 "mcp__" 开头的工具
if block.name.startswith("mcp__"):
policy = mcp_tool_policies.get(block.name, "confirm")
if policy != "allow":
print(f"\n[permission] External tool {block.name}({block.input})")
if input("Allow? [y/N] ").strip().lower() not in {"y", "yes"}:
return "Permission denied by user"
return None # None 表示放行
策略配置
MCP_HOST_POLICY = {
# 格式：(server_name, tool_name) → "allow" | "confirm" | "deny"
("docs", "search"): "allow", # docs 的 search 工具：自动放行
("docs", "get_version"): "allow", # docs 的 get_version：自动放行
("deploy", "status"): "allow", # deploy 的 status：自动放行
("deploy", "trigger"): "confirm", # deploy 的 trigger：需要确认
# 未列出的默认为 "confirm"
}
| 策略值 | 行为 |
|---|---|
"allow" |
直接放行，不询问用户 |
"confirm" |
每次调用前询问用户确认 |
"deny" |
直接拒绝（也可不配置，等效于不注册该工具） |
| (未配置) | 默认 "confirm" |
9. 完整执行流程（Agent Loop 集成）
Agent Loop 中 MCP 的位置
def agent_loop(messages: list):
while True:
try:
# ① 每次循环重新组装工具池（MCP 工具可能变化）
tools, handlers = assemble_tool_pool()
# ② 调用 LLM（传入动态工具列表）
response = client.messages.create(
model=MODEL,
system=assemble_system_prompt(), # 动态系统提示
messages=messages,
tools=tools, # ← 包含内置 + MCP 工具
max_tokens=8000,
)
except Exception as exc:
# 错误处理...
return
messages.append({"role": "assistant", "content": response.content})
# ③ 检查是否结束
if response.stop_reason != "tool_use":
trigger_hooks("Stop", messages)
return
# ④ 执行所有工具调用
results = []
for block in response.content:
if block.type != "tool_use":
continue
print(f"> {block.name}")
# ⑤ execute_tool 内部会触发 permission_hook
output = execute_tool(block, handlers) # 使用当前 handlers
print(output[:300])
results.append({
"type": "tool_result",
"tool_use_id": block.id,
"content": output,
})
messages.append({"role": "user", "content": results})
完整调用链示意
① assemble_tool_pool()
├── 合并 BUILTIN_TOOLS + 所有 MCP server 的工具
├── 生成 mcp_tool_policies（全局变量）
└── 返回 (tools, handlers)
↓
② LLM 返回 tool_use block（可能调用 MCP 工具）
↓
③ execute_tool(block, handlers) 被调用
↓
④ trigger_hooks("PreToolUse", block) 触发所有 PreToolUse 钩子
↓
⑤ permission_hook(block) 被调用
│
├── block.name 以 "mcp__" 开头？
│ ├── 否 → 走内置工具权限检查
│ └── 是 → 从 mcp_tool_policies 查找策略
│ ├── "allow" → 直接放行 ✓
│ ├── "confirm" → 询问用户
│ │ ├── 用户同意 → 放行 ✓
│ │ └── 用户拒绝 → 返回 "Permission denied by user" ✗
│ └── 其他 → 询问用户（默认行为）
↓
⑥ 返回结果给 execute_tool
│
├── 被阻止 → 返回错误字符串（Agent 不崩溃）
└── 放行 → 执行 handler（调用真实 MCP Server）
├── 成功 → 返回结果字符串
└── 失败 → 捕获异常，返回错误字符串（Agent 不崩溃）
↓
⑦ tool_result 写入 messages → 下一轮 LLM 看到
本博客所有文章除特别声明外，均采用 CC BY-NC-SA 4.0 许可协议。转载请注明来自 茯茶养生人的博客！
