S19: MCP Plugin — 代码解析
相对于 S18 新增：MCPClient 类 + normalize_mcp_name + 两个 Mock 服务器 + connect_mcp + assemble_tool_pool + Agent Loop 动态工具池更新
1. 整体功能概览
01
解决什么问题
S18 的工具集在启动时固定。S19 引入 MCP（Model Context Protocol）：运行时动态连接外部服务器，发现其工具，并将它们合并进工具池。LLM 可以在对话过程中使用新连接的工具。
02
系统中的角色
MCP 是「插件系统」：docs 服务器提供文档搜索，deploy 服务器提供部署控制。工具名称遵循 mcp__{server}__{tool} 命名约定，避免与内置工具冲突。连接后工具池动态扩展。
03
新增模块
MCPClient 类（发现 + 调用）、normalize_mcp_name()、_mock_server_docs/deploy()、MOCK_SERVERS 注册表、connect_mcp()、assemble_tool_pool()、BUILTIN_TOOLS/BUILTIN_HANDLERS（将原来的 TOOLS 拆分为 builtin + MCP）。Agent Loop 连接后立即重组工具池。
2. MCPClient 类 — MCP 服务器客户端
新增教学版用 mock handler 模拟真实 MCP 通讯（真实 CC 会通过 stdio/SSE 与服务器通讯）。
| class MCPClient:
"""Discovers and calls tools on an MCP server (mock for teaching).""" |
普通 Python 类（非 dataclass）；有 __init__、register、call_tool 三个方法 |
| def __init__(self, name: str):
self.name = name
self.tools: list[dict] = []
self._handlers: dict[str, callable] = {} |
self.tools 存工具定义（schema）；self._handlers 存实现函数；_ 前缀表示私有 |
| def register(self, tool_defs: list[dict],
handlers: dict[str, callable]): |
调用方式：client.register(tool_defs=[...], handlers={...})；教学版替代真实 MCP 工具发现 |
| self.tools = tool_defs
self._handlers = handlers |
直接赋值；覆盖 __init__ 中的空列表/字典 |
| def call_tool(self, tool_name: str, args: dict) -> str: |
统一入口；调用方无需知道具体 handler |
| handler = self._handlers.get(tool_name)
if not handler:
return f"MCP error: unknown tool '{tool_name}'" |
dict.get() 返回 None；None 为 falsy；优雅错误处理 |
| try:
return handler(**args)
except Exception as e:
return f"MCP error: {e}" |
**args 解包 dict 为关键字参数；try/except 捕获 handler 内部异常，转为字符串返回 |
示例 — call_tool 调用：
client = MCPClient("docs")
client.register(
tool_defs=[{"name":"search",...}],
handlers={"search": lambda query: f"[docs] Found 3 results for '{query}'"}
)
client.call_tool("search", {"query": "anthropic api"}) → "[docs] Found 3 results for 'anthropic api'"
client.call_tool("unknown", {}) → "MCP error: unknown tool 'unknown'"
3. normalize_mcp_name — 工具名称规范化
新增MCP 服务器名称和工具名称可能含有非法字符（空格、斜线、特殊符号）。Claude API 的工具名称只允许 [a-zA-Z0-9_-]，需要规范化。
| _DISALLOWED_CHARS = re.compile(r'[^a-zA-Z0-9_-]') |
模块级正则；[^...] 否定字符类，匹配不在集合内的字符；pre-compile 性能优化 |
| def normalize_mcp_name(name: str) -> str:
"""Replace non [a-zA-Z0-9_-] with underscore."""
return _DISALLOWED_CHARS.sub('_', name) |
Pattern.sub(replacement, string)；将所有不合规字符替换为 _；docstring 解释行为 |
| safe_server = normalize_mcp_name(server_name) # e.g. "docs"
safe_tool = normalize_mcp_name(tool_def["name"]) # e.g. "search"
prefixed = f"mcp__{safe_server}__{safe_tool}" |
双下划线 __ 分隔符；mcp__ 前缀标识来源；最终名称如 "mcp__docs__search" |
规范化示例：
normalize_mcp_name("docs") → "docs"
normalize_mcp_name("my-server") → "my-server" (连字符合法)
normalize_mcp_name("my server") → "my_server" (空格→_)
normalize_mcp_name("get/version") → "get_version" (斜线→_)
normalize_mcp_name("do.it!") → "do_it_" (点→_ 感叹号→_)
完整名称构造：
server="docs", tool="search" → "mcp__docs__search"
server="my server", tool="get/docs" → "mcp__my_server__get_docs"
4. Mock 服务器 — docs & deploy
_mock_server_docs
| def _mock_server_docs():
client = MCPClient("docs")
client.register(
tool_defs=[
{"name": "search",
"description": "Search documentation. (readOnly)",
"inputSchema": {"type": "object",
"properties": {"query": {"type": "string"}},
"required": ["query"]}},
{"name": "get_version", ...},
],
handlers={
"search": lambda query: f"[docs] Found 3 results for '{query}'",
"get_version": lambda: "[docs] API v2.1.0",
})
return client |
工厂函数（每次调用创建新实例）；inputSchema 注意：MCP 用 inputSchema（驼峰），而 Claude API 内置工具用 input_schema（下划线）；assemble_tool_pool 负责转换；lambda 短函数 |
_mock_server_deploy
| def _mock_server_deploy():
client = MCPClient("deploy")
client.register(
tool_defs=[
{"name": "trigger",
"description": "Trigger deployment. (destructive — requires approval in real CC)",
...},
{"name": "status",
"description": "Check status. (readOnly)", ...},
],
handlers={
"trigger": lambda service: f"[deploy] Triggered: {service}",
"status": lambda service: f"[deploy] {service}: running (v1.4.2)",
})
return client |
描述中标注 (destructive) — 真实 CC 会拦截此类工具要求用户确认；教学版不阻塞 |
MOCK_SERVERS 注册表
| MOCK_SERVERS = {
"docs": _mock_server_docs,
"deploy": _mock_server_deploy,
} |
dict 映射服务器名 → 工厂函数；注意值是函数本身（不加括号），由 connect_mcp 调用时再实例化 |
5. connect_mcp — 连接服务器并发现工具
新增Agent 调用此工具连接 MCP 服务器。连接后该服务器的工具在下次 assemble_tool_pool() 时自动加入工具池。
| mcp_clients: dict[str, MCPClient] = {} |
全局注册表；key = 服务器名；value = MCPClient 实例 |
| def connect_mcp(name: str) -> str: |
工具处理函数；被 run_connect_mcp 包装后加入 BUILTIN_HANDLERS |
| if name in mcp_clients:
return f"MCP server '{name}' already connected" |
幂等保护；in 操作符检查 dict key 是否存在 |
| factory = MOCK_SERVERS.get(name)
if not factory:
available = ", ".join(MOCK_SERVERS.keys())
return f"Unknown server '{name}'. Available: {available}" |
在 MOCK_SERVERS 查找工厂函数；dict.keys() 返回所有键；", ".join() 拼接为字符串 |
| mcp_client = factory() |
调用工厂函数（加括号）；返回配置好的 MCPClient 实例 |
| mcp_clients[name] = mcp_client |
注册到全局字典；此后 assemble_tool_pool 能找到它 |
| tool_names = [t["name"] for t in mcp_client.tools] |
列表推导；取每个 tool_def 的 name 字段 |
| return (f"Connected to MCP server '{name}'. "
f"Discovered {len(mcp_client.tools)} tools: {', '.join(tool_names)}") |
括号内隐式字符串拼接；f-string；len()计数；', '.join() 拼接工具名 |
示例：
connect_mcp("docs")
→ factory = _mock_server_docs (函数对象)
→ mcp_client = _mock_server_docs() (创建实例)
→ mcp_clients["docs"] = mcp_client
→ tool_names = ["search", "get_version"]
返回: "Connected to MCP server 'docs'. Discovered 2 tools: search, get_version"
connect_mcp("unknown") → "Unknown server 'unknown'. Available: docs, deploy"
connect_mcp("docs") (再次调用) → "MCP server 'docs' already connected"
6. assemble_tool_pool — 动态工具池组装
核心新增每次调用时，将固定的 BUILTIN_TOOLS + 已连接的所有 MCP 工具合并为一个工具池，传给 Claude API。
| def assemble_tool_pool() -> tuple[list[dict], dict]: |
返回 (工具定义列表, 处理器字典) 元组；两者一一对应 |
| """Assemble builtin tools + all MCP tools into one pool.""" |
docstring 说明函数用途 |
| tools = list(BUILTIN_TOOLS)
handlers = dict(BUILTIN_HANDLERS) |
浅拷贝：list() 和 dict() 创建新容器，不修改原始常量；追加 MCP 工具不影响 BUILTIN_TOOLS |
| for server_name, mcp_client in mcp_clients.items(): |
dict.items() 返回 (key, value) 对；遍历所有已连接服务器 |
| safe_server = normalize_mcp_name(server_name) |
规范化服务器名（防特殊字符） |
| for tool_def in mcp_client.tools: |
遍历该服务器的所有工具定义 |
| safe_tool = normalize_mcp_name(tool_def["name"])
prefixed = f"mcp__{safe_server}__{safe_tool}" |
构造完整工具名；f-string 双下划线分隔 |
| tools.append({
"name": prefixed,
"description": tool_def.get("description", ""),
"input_schema": tool_def.get("inputSchema", {}),
}) |
注意：MCP 原始格式用 inputSchema；Claude API 需要 input_schema（转换在这里完成）；dict.get() 带默认值 |
| handlers[prefixed] = (
lambda *, c=mcp_client, t=tool_def["name"], **kw: c.call_tool(t, kw)) |
lambda 默认参数捕获闭包变量（c=mcp_client, t=tool_def["name"] 避免循环变量陷阱）；* 强制关键字参数；**kw 接收任意关键字参数后转为 dict 传给 call_tool |
| return tools, handlers |
Python 多值返回；调用方用 tools, handlers = assemble_tool_pool() |
Lambda 闭包陷阱详解：
错误写法（循环变量陷阱）：
handlers[prefixed] = lambda **kw: mcp_client.call_tool(tool_def["name"], kw)
正确写法（默认参数捕获）：
handlers[prefixed] = (lambda *, c=mcp_client, t=tool_def["name"], **kw: c.call_tool(t, kw))
assemble_tool_pool 输出示例：
连接了 docs 后：
tools = [...BUILTIN_TOOLS..., {name:"mcp__docs__search",...}, {name:"mcp__docs__get_version",...}]
handlers = {...BUILTIN_HANDLERS..., "mcp__docs__search": lambda ..., "mcp__docs__get_version": lambda ...}
7. Agent Loop — 连接后动态更新工具池
变更S18 用固定 TOOL_HANDLERS。S19 的 agent_loop 在每次 connect_mcp 调用后立即重组工具池。
| def agent_loop(messages: list, context: dict):
tools, handlers = assemble_tool_pool()
system = assemble_system_prompt(context) |
进入循环前组装一次工具池；注意 assemble_system_prompt（非 get_system_prompt）— S19 不用缓存，因为工具池变化时 prompt 也需要更新 |
| response = client.messages.create(
model=MODEL, system=system, messages=messages,
tools=tools, max_tokens=8000) |
使用当前 tools（可能含 MCP 工具）；API 调用 |
| results = []
for block in response.content:
if block.type != "tool_use":
continue
handler = handlers.get(block.name)
output = handler(**block.input) if handler else "Unknown" |
handlers 包含内置 + MCP 所有处理器；block.name 如 "mcp__docs__search"；**block.input 解包参数 |
| if any(b.name == "connect_mcp" for b in response.content
if b.type == "tool_use"):
tools, handlers = assemble_tool_pool()
context = update_context(context, messages)
system = assemble_system_prompt(context) |
any() + generator expression；检查本轮是否调用了 connect_mcp；如果是，立即重组工具池和 prompt，下一轮 LLM 就能使用新工具 |
动态扩展示例（对话流程）：
Turn 1: 用户: "连接 docs 服务器并搜索 API 文档"
→ LLM tool_use: {name:"connect_mcp", input:{name:"docs"}}
→ connect_mcp("docs") → mcp_clients["docs"] = MCPClient
→ 检测到 connect_mcp → assemble_tool_pool()
tools 现在包含: [...BUILTIN..., "mcp__docs__search", "mcp__docs__get_version"]
→ system prompt 更新: "Connected MCP servers: docs"
Turn 2: 同一次 agent_loop 内继续
→ LLM 现在看到 mcp__docs__search 工具
→ LLM tool_use: {name:"mcp__docs__search", input:{query:"authentication"}}
→ handlers["mcp__docs__search"](**{"query":"authentication"})
→ mcp_client.call_tool("search", {"query":"authentication"})
→ "[docs] Found 3 results for 'authentication'"
8. BUILTIN_TOOLS / BUILTIN_HANDLERS 拆分
重构S18 用 TOOLS + TOOL_HANDLERS。S19 重命名为 BUILTIN_TOOLS + BUILTIN_HANDLERS，作为 assemble_tool_pool 的基础，不再直接传给 API。
新增工具：connect_mcp
| {"name": "connect_mcp",
"description": "Connect to an MCP server (docs, deploy) and discover tools.",
"input_schema": {"type": "object",
"properties": {"name": {"type": "string"}},
"required": ["name"]}}, |
加入 BUILTIN_TOOLS；LLM 可以主动调用此工具来连接服务器；description 列出可用服务器名帮助 LLM 选择 |
assemble_system_prompt 新增 MCP 状态
| def assemble_system_prompt(context: dict) -> str:
...
mcp_names = list(mcp_clients.keys())
if mcp_names:
sections.append(f"Connected MCP servers: {', '.join(mcp_names)}") |
动态读取 mcp_clients 全局状态；每次调用时反映当前连接状态；让 LLM 知道哪些 MCP 工具已可用 |
9. 完整 MCP 流程图
connect_mcp("docs") 调用链：
connect_mcp("docs")
→ MOCK_SERVERS["docs"] = _mock_server_docs (工厂函数)
→ mcp_client = _mock_server_docs()
→ mcp_client.tools = [{"name":"search",...}, {"name":"get_version",...}]
→ mcp_client._handlers = {"search": λ, "get_version": λ}
→ mcp_clients["docs"] = mcp_client
→ 返回 "Connected... Discovered 2 tools: search, get_version"
assemble_tool_pool() 调用链：
tools = [bash, read_file, ..., connect_mcp] (BUILTIN_TOOLS 拷贝)
handlers = {bash:fn, ..., connect_mcp:fn} (BUILTIN_HANDLERS 拷贝)
for "docs", mcp_client in mcp_clients.items():
safe_server = "docs"
for {"name":"search",...} in mcp_client.tools:
prefixed = "mcp__docs__search"
tools.append({"name":"mcp__docs__search","description":"...","input_schema":{...}})
handlers["mcp__docs__search"] = lambda *, c=mcp_client, t="search", **kw: c.call_tool(t, kw)
for {"name":"get_version",...} in mcp_client.tools:
prefixed = "mcp__docs__get_version"
...同上...
return tools (len = N_builtin + 2), handlers
LLM 调用 mcp__docs__search({"query":"anthropic"}):
block.name = "mcp__docs__search"
handler = handlers["mcp__docs__search"]
handler(**{"query":"anthropic"})
→ c.call_tool("search", {"query":"anthropic"})
→ mcp_client._handlers["search"](**{"query":"anthropic"})
→ lambda query: f"[docs] Found 3 results for '{query}'"
→ "[docs] Found 3 results for 'anthropic'"
两个 Mock 服务器对比：
┌─────────────────┬──────────────────┬──────────────────────────────────┐
│ Server │ Tools │ 特性 │
├─────────────────┼──────────────────┼──────────────────────────────────┤
│ docs │ search │ readOnly - 无副作用 │
│ │ get_version │ readOnly - 无副作用 │
├─────────────────┼──────────────────┼──────────────────────────────────┤
│ deploy │ trigger │ destructive - 真实CC需要确认 │
│ │ status │ readOnly - 无副作用 │
└─────────────────┴──────────────────┴──────────────────────────────────┘