AI 应用的 USB-C — 一次握手, 把任意 server 的工具/资源/模板标准化接进你的 agent
MCP 解决的是 M×N 接口爆炸: M 个 AI 应用对接 N 种工具, 过去要写 M×N 个胶水层; 有了 MCP, 两边都只实现一次协议, 变成 M+N。它像 USB-C: 设备(工具提供方)与电脑(Agent 应用)各出一个标准口, 插上即用——tools 是插上就能被模型调用的"外设", resources 是可被应用挂载的"移动硬盘", prompts 是预设的"快捷键"。学会这一页, 你就能自己写 server 把公司内部系统接进任何 agent。
# 没有协议: 每个应用×每个工具都要写适配 glue = apps * tools # → M×N 份胶水代码 # 有了 MCP: 双方各实现一次协议 glue = apps + tools # → M+N 份, 复用即插即用
tools # → 模型自主调用: 需要参数校验+权限闸门 resources # → App 决定读什么: file://orders/42 注入上下文 prompts # → 用户输入 /review 触发: 模板由 server 提供
{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}
# ↑ client 写进 server 的 stdin, server 从 stdout 回一行 JSON
# 关键: stdout 是协议专线, print 调试日志 = 毁掉通道POST /mcp Authorization: Bearer <token> # → Content-Type: text/event-stream (流式) 或 application/json # 关键: 会话用 Mcp-Session-Id 头关联, 重启会丢
protocolVersion 与 capabilities。capabilities 决定"这个 server 能干什么", 没声明的能力 client 不会用。 {"method": "initialize",
"params": {"protocolVersion": "2025-06-18",
"capabilities": {"tools": {}, "resources": {}}}}
# 关键: 忘声明 capabilities → client 直接不发起对应请求page = await session.list_tools() page.tools[0].name # → "query_orders" page.tools[0].inputSchema # → JSON Schema, 转成模型 tools 参数 page.nextCursor # → 有值就继续翻页, 否则漏一半
content 数组(text/image/resource)与 isError。业务失败用 isError=true + 文本说明, 让模型读到失败原因。 result = await session.call_tool("query_orders", {"status": "paid"}) result.isError # → False result.content[0].text # → '[{"order_id": "A102", ...}]'
readOnlyHint(只读)/destructiveHint(破坏性)/idempotentHint(幂等)。 # readOnlyHint=true → host 可以放心在自动模式放行 # destructiveHint=true → host 应挂起等人工确认 # 关键: 注解是"提示"不是"保证", 高危操作仍要服务端校验
401 + WWW-Authenticate → client 引导用户授权
Authorization: Bearer eyJhbGciOiJSUzI1NiIs...
# 关键: token 过期要能触发重新授权, 而不是静默 401 循环roots 告诉 server "你只能在这些目录/URI 范围内活动", 是文件类 server 的核心安全边界。 roots = ["file:///Users/luca/ZhongQiuProject"] # server 侧: 每次文件操作前校验路径前缀 # user 传 "../etc/passwd" → 拒绝: 越出 root 范围
# server: "帮我总结这段 diff" sampling/createMessage → client 弹窗确认 → LLM 生成 → 回给 server # 关键: sampling 让 server 不必自带 API key
notifications/tools/list_changed, client 重新 list。热更新不重启。 # server 新增工具后: notifications/tools/list_changed # → client 重拉 tools/list # 不发通知 → 新工具永远不出现在模型面前
print("query took 0.3s") # → client: Invalid Request: 解析失败 logging.basicConfig(stream=sys.stderr) # 对: 日志走 stderr
客服 agent 要查订单库, 但不能给写权限。FastMCP 装饰器把普通函数一键注册成带 Schema 的工具。
from mcp.server.fastmcp import FastMCP mcp = FastMCP("orders") @mcp.tool() def query_orders(status: str, limit: int = 20) -> str: """按状态查询订单, 返回 JSON 数组。status: pending/paid/shipped""" rows = db.fetch("SELECT id, amount, status FROM orders" " WHERE status = %s LIMIT %s", (status, limit)) return json.dumps(rows, ensure_ascii=False) if __name__ == "__main__": mcp.run() # 默认 stdio 传输, 日志记得走 stderr
本地 server 靠一份 JSON 声明接入。command+args+env 三要素, 密钥从 env 注入而不是写死在代码里。
// claude_desktop_config.json / 通用 MCP 客户端配置 { "mcpServers": { "orders": { "command": "uv", "args": ["run", "orders_server.py"], "env": {"DB_DSN": "postgres://readonly:***@10.0.0.3/orders"} } } } // 关键: DB 账号用 readonly 角色 — server 被攻破也写不了库
给全团队的共享 server 要跨网部署: streamable-http 传输 + 反代 + 域名, 一处部署处处接入。
mcp = FastMCP("orders", host="0.0.0.0", port=8808) # ...工具注册同上... mcp.run(transport="streamable-http") # 客户端侧连接: params = {"url": "https://mcp.internal.corp/orders", "headers": {"Authorization": f"Bearer {token}"}} # nginx 反代记得: proxy_buffering off; SSE 流不能被缓冲
host 侧集成 MCP 的标准流程, 全异步。initialize 必须先于一切。
from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client params = StdioServerParameters(command="uv", args=["run", "orders_server.py"]) async with stdio_client(params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # ① 握手, 必须第一 tools = await session.list_tools() # ② 发现 result = await session.call_tool( # ③ 调用 "query_orders", {"status": "pending"}) return result.content[0].text
大文档不进工具参数, 做成 resource 由应用按需挂载——注入什么由 App 决定, 不是模型。
@mcp.resource("file://orders/{order_id}") def order_detail(order_id: str) -> str: """单个订单的完整档案, 供应用按需注入""" return render(order_detail_md(order_id)) # host 侧: contents = await session.read_resource("file://orders/A102") # → 应用决定塞进上下文还是丢弃, 模型只看到最终上下文
把"代码评审""写周报"沉淀成 server 端模板, 全团队 agent 都能 /review 直呼。
@mcp.prompt() def review(code: str, focus: str = "正确性与边界条件") -> str: """生成一次代码评审请求""" return f"""请评审以下代码, 重点关注 {focus}: {code} 输出: 问题清单(按严重度排序) + 修复建议""" # 用户输入 /review → client 列出该模板 → 按参数填充后发送
host 的自动放行策略不能拍脑袋, 靠 server 显式声明只读/破坏性, 再配白名单策略。
from mcp.types import ToolAnnotations @mcp.tool(annotations=ToolAnnotations(readOnlyHint=True)) def query_orders(status: str) -> str: ... @mcp.tool(annotations=ToolAnnotations(destructiveHint=True, idempotentHint=False)) def purge_cache(pattern: str) -> str: ... # host 策略: readOnly 自动放行; destructive 一律挂起等审批
server 校验 Bearer token 的 issuer/audience/scope, 401 响应带授权指引, 而不是裸奔。
# server 侧每个请求: auth = request.headers.get("Authorization", "") if not auth.startswith("Bearer "): return JSONResponse(status_code=401, headers={ "WWW-Authenticate": f'Bearer resource_metadata=' f'"https://mcp.corp/.well-known/oauth-protected-resource"'}) claims = jwt.decode(auth[7:], JWKS, audience="mcp-orders") # 校验 iss/aud/exp/scope, 缺一不可
server 进程挂了工具表就失效。容器化部署配重启策略与健康检查, host 侧重连有退避。
# docker-compose.yml services: mcp-orders: build: ./mcp-orders command: ["python", "server.py"] # streamable-http 模式 restart: unless-stopped # 崩溃自动拉起 healthcheck: test: ["CMD", "curl", "-sf", "http://localhost:8808/health"] interval: 30s timeout: 3s environment: - DB_DSN=postgres://readonly:***@db/orders - MCP_AUTH_AUDIENCE=mcp-orders
host 同时接 5 个 server, 三个都有 get_user。注册时统一加前缀, 冲突启动即报错。
def aggregate(servers: dict[str, McpClient]): registry = {} for name, client in servers.items(): for t in client.list_tools(): fq = f"{name}__{t.name}" # github__get_user if fq in registry: raise ToolConflict(fq) # 冲突启动即失败, 别带病上线 registry[fq] = (client, t) return registry # 模型看到的每个工具都可追溯到唯一 server
print 打调试日志, 混进 JSON-RPC 帧. 正解: 日志一律走 stderr。 # 错: print("cache hit") # → stdout 混入非 JSON 行, 解析崩 # 对: logging.basicConfig(stream=sys.stderr)
# 错: capabilities: {} # → client 不发起 tools/list # 对: capabilities: {"tools": {}, "resources": {}}
# 错: return "2024-11-05" # → client 要求 2025-06-18, 握手失败 # 对: return negotiated(requested_version)
# 错: 跑 10 分钟一声不吭 # → client 60s 超时, 任务白跑 # 对: await ctx.report_progress(40, 100) # 阶段性汇报
# 错: return open(f).read() # → 50MB 注入上下文 # 对: return head(f, 200) + "\n...(用 read_lines 工具分段读)"
# 错: app.run(host="0.0.0.0") # → 无鉴权裸奔公网 # 对: Bearer 校验 + securityGroup 限源
# 错: rows = psycopg2.cursor().fetchall() # → 事件循环卡 5s # 对: rows = await asyncio.to_thread(sync_query, sql)
# 错: tools = (await session.list_tools()).tools # → 只有第一页 # 对: while page.nextCursor: page = await list_tools(cursor)
# 错: registry[t.name] = client # → 静默覆盖 # 对: fq = f"{server}__{t.name}"; 冲突即 raise
# 错: 第三方 server 工具无条件信任 + 全权限 # 对: 未知来源 server → 沙箱权限 + 每次调用审计
// 错: "env": {"DB_DSN": "postgres://user:p@ssw0rd@..."} // 对: "env": {"DB_DSN": "${DB_DSN}"} // 真值在环境/密钥管理
# 错: sys.stdout.write(frame) # → Windows 下 \r\n 污染帧 # 对: sys.stdout.buffer.write(frame.encode() + b"\n")
# 错: spawn 后不监控 # → 挂了就是永久失效 # 对: restart: unless-stopped + client 指数退避重连
# 错: return json.dumps(all_rows) # → 80k tokens 入上下文 # 对: return json.dumps([slim(r) for r in rows[:20]])
'; DROP TABLE, 直通 SQL. 原因: 以为"是 MCP 就安全", server 内没闸门. 正解: server 内照常 pydantic 校验 + 参数化查询。 # 错: db.execute(f"SELECT ... {args['q']}") # → 注入直达 # 对: db.execute("SELECT ... WHERE q = %s", (validated.q,))
# 错: 注册新工具后闷声不响 # → host 缓存里没有它 # 对: notify("notifications/tools/list_changed")
# 错: server 里写"退款必须先查风控"业务流 # 对: server 暴露 check_risk + refund 两个原子工具, 流程归 host
# 错: async with connect(): await call_one_tool() # 每次新建会话 # 对: 会话保活 + 断线重连, initialize 只在握手时发生
# 错: 只支持 file://orders/{id} 直读, 无法发现 # 对: resources/list 返回可枚举清单 + 模板双通道
# 错: sys.stderr.write(jsonrpc_frame) # → 协议帧进了日志黑洞 # 对: stdout 写帧, stderr 写日志, 单测双流各断言