MCP 全景

8 个模型客户端（Claude、Trae、Cursor、自研 Agent……）
× 10 个服务提供方（GitHub、数据库、天气、Slack……）
= 80 份适配代码

M 个客户端 × N 个服务 → M + N 份实现（各自面向协议编程）

// 请求（客户端 → 服务端）
{
  "jsonrpc": "2.0",
  "id": "req-42",
  "method": "tools/call",
  "params": {
    "name": "get_weather",
    "arguments": { "city": "北京" }
  }
}

// 响应（服务端 → 客户端）
{
  "jsonrpc": "2.0",
  "id": "req-42",
  "result": {
    "content": [
      { "type": "text", "text": "北京 12°C 晴" }
    ]
  }
}

Client                          Server
  │ ── initialize ──────────────→ │   ① 握手：交换版本、能力清单
  │ ←─ initialize result ────── │
  │ ── initialized ────────────→ │   ② 确认（notification）
  │                              │
  │ ── tools/list ─────────────→ │   ③ 发现工具
  │ ←─ 工具清单 + schema ────── │
  │                              │
  │ ── tools/call ────────────→ │   ④ 调用（可重复任意次）
  │ ←─ 执行结果 ──────────────── │
  │                              │
  │ ── （循环：list/call） ──────→ │

# 时间服务（官方示例，uvx 直接拉起）
uvx mcp-server-time

# MCP Inspector：官方调试客户端，图形化查看通信全过程
npx -y @modelcontextprotocol/inspector

# uv add "mcp[cli]"
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("demo-server")          # 参数是 server 名称

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个整数之和"""
    return a + b

if __name__ == "__main__":
    mcp.run()                          # 默认 stdio 传输

npx -y @modelcontextprotocol/inspector uv run server.py
# Inspector 界面里：Connect → Tools → List Tools → 就能看到 add 了

# file_server.py
from pathlib import Path
from mcp.server.fastmcp import FastMCP

# 允许访问的根目录：把权限写死在 server 侧（第 15 章的最小权限原则预演）
ALLOWED_ROOT = Path("/Users/you/sandbox").resolve()

mcp = FastMCP("file-assistant")

def _safe_path(path: str) -> Path:
    """路径安全检查：禁止逃逸出允许的目录"""
    p = (ALLOWED_ROOT / path).resolve()
    if not str(p).startswith(str(ALLOWED_ROOT)):
        raise ValueError(f"禁止访问沙箱外的路径：{path}")
    return p

@mcp.tool()
def read_file(path: str, max_chars: int = 4000) -> str:
    """读取沙箱内文本文件的内容。文件过大时自动截断。

    Args:
        path: 相对沙箱根目录的文件路径
        max_chars: 最多返回的字符数
    """
    p = _safe_path(path)
    if not p.is_file():
        return f"文件不存在：{path}"
    content = p.read_text(encoding="utf-8")
    if len(content) > max_chars:
        return content[:max_chars] + f"\n...(已截断，共 {len(content)} 字符)"
    return content

@mcp.tool()
def write_file(path: str, content: str) -> str:
    """将文本内容写入沙箱内的文件（覆盖写入）。

    Args:
        path: 相对沙箱根目录的目标路径
        content: 要写入的完整内容
    """
    p = _safe_path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(content, encoding="utf-8")
    return f"已写入 {path}（{len(content)} 字符）"

@mcp.tool()
def list_dir(path: str = ".") -> str:
    """列出沙箱内指定目录的文件和子目录。"""
    p = _safe_path(path)
    if not p.is_dir():
        return f"目录不存在：{path}"
    entries = sorted(p.iterdir(), key=lambda x: x.name)
    lines = [f"{'[dir] ' if e.is_dir() else '      '}{e.name}" for e in entries]
    return "\n".join(lines) or "（空目录）"

if __name__ == "__main__":
    mcp.run()

@mcp.resource("config://app")
def get_config() -> str:
    """应用配置（应用程序控制何时读取）"""
    return "当前环境: 生产\n版本: 2.3.1"

@mcp.prompt()
def review_code(code: str) -> str:
    """代码评审提示词模板（用户通过斜杠命令触发）"""
    return f"请评审以下代码，关注安全性和可读性：\n\n{code}"

{
  "mcpServers": {
    "file-assistant": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/project", "file_server.py"]
    },
    "time": {
      "command": "uvx",
      "args": ["mcp-server-time"]
    }
  }
}

import asyncio
import json

class StdioTransport:
    """MCP stdio 传输层：管理子进程和 JSON-RPC 报文读写"""

    def __init__(self, command: list[str]):
        self.command = command
        self.process: asyncio.subprocess.Process | None = None
        self._next_id = 0

    async def start(self):
        self.process = await asyncio.create_subprocess_exec(
            *self.command,
            stdin=asyncio.subprocess.PIPE,
            stdout=asyncio.subprocess.PIPE,
            stderr=asyncio.subprocess.DEVNULL,   # server 的日志不混入协议流
        )

    async def send(self, method: str, params: dict | None = None) -> dict:
        """发一条 Request 并等待对应 id 的 Response"""
        self._next_id += 1
        req_id = self._next_id
        message = {"jsonrpc": "2.0", "id": req_id, "method": method,
                   "params": params or {}}
        # 按行分帧：官方 SDK 默认用 Content-Length 头分帧（同 LSP）；
        # 自实现按行分帧更直观，但要求 server 也支持——官方 Python SDK
        # 的 stdio server 同时兼容两种，所以能跑通。若连接其他实现，
        # 以对方文档为准（用官方 SDK 客户端则完全无需关心）
        line = json.dumps(message, ensure_ascii=False) + "\n"
        self.process.stdin.write(line.encode())
        await self.process.stdin.drain()
        return await self._read_response(req_id)

    async def notify(self, method: str, params: dict | None = None) -> None:
        """发一条 Notification（不等待回复）"""
        message = {"jsonrpc": "2.0", "method": method, "params": params or {}}
        line = json.dumps(message, ensure_ascii=False) + "\n"
        self.process.stdin.write(line.encode())
        await self.process.stdin.drain()

    async def _read_response(self, req_id: int) -> dict:
        while True:
            raw = await self.process.stdout.readline()
            if not raw:
                raise ConnectionError("server 进程已退出")
            data = json.loads(raw)
            if data.get("id") == req_id:        # 跳过通知等无关消息
                if "error" in data:
                    raise RuntimeError(f"MCP 错误：{data['error']}")
                return data["result"]

    async def close(self):
        if self.process:
            self.process.terminate()
            await self.process.wait()

class McpClient:
    def __init__(self, command: list[str]):
        self.transport = StdioTransport(command)

    async def connect(self):
        """① initialize 握手 ② initialized 确认"""
        await self.transport.start()
        result = await self.transport.send("initialize", {
            "protocolVersion": "2025-06-18",
            "capabilities": {},                  # 客户端能力：不声明任何高级特性
            "clientInfo": {"name": "handmade-client", "version": "0.1.0"},
        })
        # result 里有 server 的能力声明，可按需保存
        await self.transport.notify("notifications/initialized")

    async def list_tools(self) -> list[dict]:
        """③ 发现工具"""
        result = await self.transport.send("tools/list", {})
        return result.get("tools", [])

    async def call_tool(self, name: str, arguments: dict) -> str:
        """④ 调用工具"""
        result = await self.transport.send("tools/call", {
            "name": name, "arguments": arguments,
        })
        # 结果是 content 数组，取全部 text 拼接
        parts = [c.get("text", "") for c in result.get("content", [])
                 if c.get("type") == "text"]
        return "\n".join(parts) or str(result)

    async def close(self):
        await self.transport.close()

async def main():
    client = McpClient(["uvx", "mcp-server-time"])

    await client.connect()
    tools = await client.list_tools()
    for t in tools:
        print(f"- {t['name']}: {t['description']}")

    result = await client.call_tool("get_current_time", {"timezone": "Asia/Shanghai"})
    print(result)

    await client.close()

asyncio.run(main())

- get_current_time: Get current time in a specific timezone
- convert_time: Convert time between timezones
当前时间: 2026-08-26T15:30:00+08:00 (Asia/Shanghai)

{
    "name": "get_current_time",
    "description": "Get current time in a specific timezone",
    "inputSchema": {"type": "object",
                    "properties": {"timezone": {"type": "string"}},
                    "required": ["timezone"]},
}

{
    "type": "function",
    "function": {
        "name": "get_current_time",
        "description": "Get current time in a specific timezone",
        "parameters": {"type": "object",
                       "properties": {"timezone": {"type": "string"}},
                       "required": ["timezone"]},
    },
}

def mcp_to_openai(mcp_tool: dict, server_name: str) -> dict:
    """MCP 工具 → OpenAI tools 格式；加 server 前缀防重名"""
    return {
        "type": "function",
        "function": {
            "name": f"{server_name}__{mcp_tool['name']}",   # 前缀：多个 server 同名工具不冲突
            "description": mcp_tool.get("description", ""),
            "parameters": mcp_tool.get("inputSchema", {}),
        },
    }

class McpRegistry:
    """聚合多个 MCP server 的工具注册器（实现 Registry 同构接口）"""

    def __init__(self):
        self.servers: dict[str, McpClient] = {}
        self.tools: dict[str, tuple[str, str]] = {}   # 工具全名 → (server名, 原名)

    async def add_server(self, name: str, command: list[str]):
        client = McpClient(command)
        await client.connect()
        self.servers[name] = client
        for t in await client.list_tools():
            full_name = f"{name}__{t['name']}"
            self.tools[full_name] = (name, t["name"])

    def schemas(self) -> list[dict]:
        # 把所有 server 的工具转成 OpenAI 格式（此处省略 inputSchema 缓存细节）
        return [mcp_to_openai_cached(t) for t in ...]

    def dispatch(self, full_name: str, arguments_json: str) -> str:
        server_name, original = self.tools[full_name]
        client = self.servers[server_name]
        return asyncio.run(client.call_tool(
            original, json.loads(arguments_json)))

registry = McpRegistry()
asyncio.run(registry.add_server("time", ["uvx", "mcp-server-time"]))
# 现在模型可以用 time__get_current_time 等工具了
agent = Agent(system="你可以查询各时区的时间。", registry=registry)
print(agent.run("现在上海几点？"))

模型 ←(OpenAI tools 格式)→ Agent 循环 ←(JSON-RPC)→ MCP servers
        ↑ 第 7 章的协议            ↑ MCP 协议（本章第一部分）

# uv add mcp
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

async def with_sdk():
    params = StdioServerParameters(command="uvx", args=["mcp-server-time"])
    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(
                "get_current_time", {"timezone": "Asia/Shanghai"})
            print(result.content[0].text)

asyncio.run(with_sdk())

              ┌────────────────────────────┐
              │      Agent 主循环(第8章)     │
              │   tools = local + remote   │
              └──────────┬─────────────────┘
                         │ dispatch(名字, 参数)
              ┌──────────▼─────────────────┐
              │      混合 Registry          │
              │  ┌─────────┐ ┌──────────┐ │      ┌──────────────┐
              │  │本地工具表│ │MCP 工具表 │ │ stdio│ mcp-server-A │
              │  └─────────┘ └────┬─────┘ │─────→└──────────────┘
              └───────────────────┼───────┘ stdio┌──────────────┐
                                  │─────────────→│ mcp-server-B │
                                                 └──────────────┘

{
  "mcpServers": {
    "time": {
      "command": "uvx",
      "args": ["mcp-server-time"]
    },
    "file-assistant": {
      "command": "uv",
      "args": ["run", "--directory", "/path/to/project", "file_server.py"]
    }
  }
}

import asyncio
import json
from pathlib import Path

class McpManager:
    """管理所有 MCP server 的连接生命周期"""

    def __init__(self, config_path: str):
        config = json.loads(Path(config_path).read_text(encoding="utf-8"))
        self.servers_cfg: dict[str, dict] = config.get("mcpServers", {})
        self.sessions: dict[str, "ClientSession"] = {}
        self.tool_index: dict[str, dict] = {}    # 全名 → 工具元数据

    async def connect_all(self):
        from mcp import ClientSession, StdioServerParameters
        from mcp.client.stdio import stdio_client

        for name, cfg in self.servers_cfg.items():
            params = StdioServerParameters(
                command=cfg["command"], args=cfg.get("args", []),
                env=cfg.get("env"),
            )
            # 注意：stdio_client 返回的异步上下文管理器必须保持存活
            # （存住 CM 对象本身，防止流被提前关闭）。这里写法偏底层，
            # 生产建议把整个 Agent 跑在一个事件循环里用 async with。
            cm = stdio_client(params)
            read, write = await cm.__aenter__()
            session_cm = ClientSession(read, write)
            session = await session_cm.__aenter__()
            await session.initialize()

            tools = await session.list_tools()
            for t in tools.tools:
                full_name = f"{name}__{t.name}"
                self.tool_index[full_name] = {
                    "server": name,
                    "original": t.name,
                    "schema": t.inputSchema,
                    "description": t.description or "",
                }
            self.sessions[name] = session
            # 生产代码还应保存 cm / session_cm 以便退出时 __aexit__

    def schemas(self) -> list[dict]:
        """全部 MCP 工具 → OpenAI tools 格式（第三部分的转换思路）"""
        result = []
        for full_name, meta in self.tool_index.items():
            result.append({
                "type": "function",
                "function": {
                    "name": full_name,
                    "description": meta["description"],
                    "parameters": meta["schema"],
                },
            })
        return result

    def dispatch(self, full_name: str, arguments_json: str) -> str:
        """按前缀路由到对应 server 执行"""
        meta = self.tool_index.get(full_name)
        if not meta:
            return f"错误：不存在工具 {full_name}"
        session = self.sessions[meta["server"]]
        # 注意：同一线程内 asyncio.run 不可重入（嵌套调用会报错）。
        # 示例为求简洁；生产做法见 4.5 节的说明。
        result = asyncio.run(session.call_tool(
            meta["original"], json.loads(arguments_json)))
        return "\n".join(
            c.text for c in result.content if hasattr(c, "text"))

class HybridRegistry:
    """本地工具 + MCP 工具的统一注册器"""

    def __init__(self, local: Registry, mcp: McpManager | None):
        self.local = local
        self.mcp = mcp

    def schemas(self) -> list[dict]:
        result = self.local.schemas() if self.local else []
        if self.mcp:
            result += self.mcp.schemas()
        return result

    def dispatch(self, name: str, arguments_json: str) -> str:
        # MCP 工具名带 server 前缀（time__get_current_time）
        if self.mcp and name in self.mcp.tool_index:
            return self.mcp.dispatch(name, arguments_json)
        # 本地工具原样分发（第 7 章的 Registry 接口）
        return self.local.dispatch(name, arguments_json)

async def main():
    # 本地工具（第 7 章）
    local = Registry()
    local.register(Tool(calculate, CalculateArgs,
        "精确计算算术表达式。任何数学计算都必须用本工具。"))

    # MCP 工具
    mcp = McpManager("mcp.json")
    await mcp.connect_all()

    # 合体
    registry = HybridRegistry(local=local, mcp=mcp)

    agent = Agent(
        system="你是一个实用助手，能精确计算、查询全球时间。",
        registry=registry,
        max_rounds=6,
    )
    agent.on(logger)
    print(agent.run("帮我算 987 * 654，再告诉我现在纽约几点"))

    # 关闭连接（真实代码应处理优雅退出，这里示意）

asyncio.run(main())

  ⚙ calculate({"expression": "987*654"})
  ↳ 645498
  ⚙ time__get_current_time({"timezone": "America/New_York"})
  ↳ 当前时间: 2026-08-26T03:30:00-04:00 (America/New_York)
  ✔ 2 轮完成

def dispatch(self, full_name, arguments_json):
    try:
        return self.mcp.dispatch(full_name, arguments_json)
    except Exception as e:
        return f"MCP 工具调用失败（{e}）。可告知用户该能力暂不可用。"