MCP Client 侧:Python 实战

前三篇我们一直在写 Server。写完 Tool、写完 Resource、写完 Prompt，然后打开 Claude Desktop，配一下 claude_desktop_config.json，看着桌面客户端把工具跑起来。

但生产里没人靠 Claude Desktop 跑业务。你写了个 MCP Server 连接内部 CRM、连接内部知识库，最终要接到自己的后端服务、自己的 Agent Pipeline、自己的运维脚本里。Server 是复用的接口层，Client 才是把它接进你自己系统的胶水。

这一篇讲：怎么在纯 Python 代码里当 Client，连接 MCP Server、列工具、调工具、读 Resource，最后把这些工具喂给 Claude，让 Claude 自己去调。

不限篇幅，从异步开始讲，一路讲到能跑起来的完整闭环。

一、先把三层架构复位

前面第 01 篇讲过 MCP 三层架构，但那时候我们默认 Host 就是 Claude Desktop。这里要把这个默认拆掉。

MCP 协议的角色定义是这样的：

• Host
  ：承载 LLM 的应用。它决定谁能连、连多少个 Server、怎么把工具展示给用户。
• Client
  ：Host 内部的一个连接实例，一个 Client 对应一个 Server 连接。它负责协议握手、消息编解码、请求/响应管理。
• Server
  ：暴露能力（Tools/Resources/Prompts）的进程。

Claude Desktop 是 Host 的一种实现，它内部帮你管理了 N 个 Client。但 Host 不是必须的——你可以自己写一个 Python 脚本，在这个脚本里手动创建 Client 对象，去连接任意一个 MCP Server。

换句话说，"当 Client"这件事，MCP 官方 SDK 已经给了你完整的类，你只需要拼装。

这一篇要写的东西，本质就是我们自己扮演 Host 的角色，用官方 SDK 提供的 ClientSession 去连 Server。

二、异步是绕不开的：async/await 从零讲

ClientSession 是异步的。所有方法都要 await，所有代码都要写在 async def 函数里。如果你之前只写过同步 Python，这里必须先打通概念，否则后面每一行都会看不懂。

2.1 为什么 MCP Client 是异步的

一次工具调用的完整链路是这样的：

Client.call_tool("query_db", {"sql": "..."})    ↓序列化成 JSON-RPC 消息    ↓通过 stdio / SSE / HTTP 发给 Server    ↓Server 干活（可能查数据库、调 API，几百毫秒到几秒）    ↓Server 返回结果    ↓Client 收到、反序列化、返回给你

中间"等 Server 返回"这一段，Client 什么都不用干，只是在等 I/O。如果用同步代码写，这个线程就阻塞在那儿，不能干任何别的事。

异步的意思是：等 I/O 的时候，线程可以去处理别的任务，等结果就绪了再回来继续。对于 MCP 这种"发请求→等响应"的协议，异步是最合适的抽象。

2.2 三个关键字：async / await / async with

只讲你看懂 MCP Client 代码所需要的最小集：

async def：定义一个协程函数。调用它不会立刻执行，而是返回一个"协程对象"。

async def hello():    return "hi"result = hello()  # result 是协程对象，不是 "hi"

await：把协程"跑起来"，等它结束、拿到返回值。await 只能用在 async def 内部。

async def main() -> None:    result = await hello()  # 现在 result == "hi"    print(result)

async with：异步版的上下文管理器。进入时 await 一次，退出时 await 一次。MCP 里用来管理"连接 Server → 用完关闭"的生命周期。

启动入口：asyncio.run(main())。这一行把整个异步世界跑起来。

import asyncioasync def main() -> None:    ...if __name__ == "__main__":    asyncio.run(main())

记住这四件东西，MCP Client 的代码就能读了。

三、ClientSession：MCP Client 的核心类

MCP Python SDK 里，ClientSession 就是"一个到 Server 的连接"。它管四件事：

1. 握手
   ：告诉 Server 我是谁，问 Server 你支持哪些能力。
2. 列能力
   ：拿到 Server 的 Tools/Resources/Prompts 列表。
3. 调用
   ：发 tools/call、resources/read 等请求，等响应。
4. 通知
   ：处理 Server 主动推送的消息（如资源变更通知）。

ClientSession 本身不管"怎么跟 Server 通信"。通信这件事由 传输层（transport） 负责。stdio、SSE、Streamable HTTP，每种传输都有对应的 client 工厂函数：

• mcp.client.stdio.stdio_client
  ：启动一个子进程，通过 stdin/stdout 通信。
• mcp.client.sse.sse_client
  ：连接一个 HTTP + SSE 端点。
• mcp.client.streamable_http.streamablehttp_client
  ：连接 Streamable HTTP 端点（后面篇章讲）。

这一篇只讲 stdio，因为前三篇写的 Server 都是 stdio 传输。

四、连接一个 stdio Server：完整流程拆解

导入清单先摆出来：

from mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_client

4.1 StdioServerParameters：告诉 Client 怎么启动 Server

Server 是一个独立进程，Client 需要知道用什么命令去启动它、要不要传参数、要不要设环境变量。

server_params = StdioServerParameters(    command="python",    args=["-m", "my_mcp_server"],    env=None,  # 需要注入环境变量时传 dict)

字段含义：

• command
  ：可执行文件。可以是 "python"、"node"、"uvx"、绝对路径的二进制等。
• args
  ：命令行参数列表。上面等价于 python -m my_mcp_server。
• env
  ：环境变量字典。给 None 时 SDK 不会继承父进程的整个环境，需要 PATH 之类的关键变量要手动传（下面完整示例里会处理）。

4.2 stdio_client：拉起子进程 + 建立管道

stdio_client(server_params) 返回一个异步上下文管理器，进入时它会：

1. fork
   /spawn 一个子进程，执行 command args。
2. 抓住这个子进程的 stdin、stdout。
3. 返回一对 (read_stream, write_stream)，这两个流就是跟 Server 通信的双向管道。

async with stdio_client(server_params) as (read_stream, write_stream):    ...

退出 async with 块时，SDK 会关闭管道、终止子进程。你不用手动 process.kill()。

4.3 ClientSession：在管道上跑 MCP 协议

拿到管道之后，把它交给 ClientSession。ClientSession 也是个异步上下文管理器：

async with ClientSession(read_stream, write_stream) as session:    await session.initialize()    ...

session.initialize() 是 MCP 协议规定的握手动作：Client 报告自己的协议版本和能力，Server 报告它的版本和能力，双方对齐后才能继续。这一步不能省，否则后续调用会被 Server 拒绝。

到这一步，session 就可以用了。

五、四个最常用的 session 方法

5.1 list_tools：列工具

tools_result = await session.list_tools()for tool in tools_result.tools:    print(tool.name, "→", tool.description)    print("  inputSchema:", tool.inputSchema)

返回的 tool.inputSchema 是标准的 JSON Schema。这个 Schema 就是我们下一节要转成 Anthropic tool format 的原料。

5.2 call_tool：调工具

result = await session.call_tool(    name="query_db",    arguments={"sql": "SELECT COUNT(*) FROM users"},)for block in result.content:    if block.type == "text":        print(block.text)

关键点：

• arguments
   必须是 dict，字段必须匹配 Server 声明的 inputSchema。
• 返回的 result.content 是一个列表，每个元素是内容块（text / image / resource 等）。绝大多数工具只返回一个 text 块。
• 如果 Server 抛异常，result.isError 为 True，result.content 里是错误信息。

5.3 list_resources / read_resource：读 Resource

resources_result = await session.list_resources()for res in resources_result.resources:    print(res.uri, "→", res.name)content = await session.read_resource(uri="file:///data/report.md")for block in content.contents:    if hasattr(block, "text"):        print(block.text)

Resource 和 Tool 的差别，第 03 篇讲过：Tool 是"让模型调"的，Resource 是"让 Host 主动读、塞进上下文"的。Client 侧只是提供读的能力，怎么用是你自己决定。

5.4 list_prompts / get_prompt

prompts_result = await session.list_prompts()for p in prompts_result.prompts:    print(p.name, p.arguments)prompt = await session.get_prompt(    name="daily_summary",    arguments={"date": "2026-08-14"},)for msg in prompt.messages:    print(msg.role, msg.content)

这个不常用，但知道有就行。

六、把 MCP 工具喂给 Claude：完整闭环

前面所有铺垫，都是为了这一节。MCP Client 单独跑没意义，它的价值是把 Server 的能力接进 LLM 的推理循环。

流程：

1. 启动一个 MCP Server（我们前面篇章写好的）2. 用 ClientSession 连上3. 调 list_tools() 拿到工具列表4. 把 MCP 的 tool schema 转成 Anthropic API 的 tool format5. 调 anthropic.messages.create()，把工具列表传进去6. Claude 返回 tool_use，我们用 session.call_tool() 真的调7. 把工具结果作为 tool_result 塞回对话，让 Claude 继续8. 直到 Claude 返回 stop_reason == "end_turn"

6.1 Schema 转换：MCP tool → Anthropic tool

两边格式几乎一样，只是字段名不同：

def mcp_tool_to_anthropic(mcp_tool) -> dict:    """把 MCP SDK 的 Tool 对象转成 Anthropic API 需要的 tool 定义"""    return {        "name": mcp_tool.name,        "description": mcp_tool.description or "",        "input_schema": mcp_tool.inputSchema,    }

一行 dict comprehension 就能批量转，简单到不像话，但这个"简单"正是 MCP 的价值——它用的就是 JSON Schema 这个业界通用协议。

6.2 主循环：让 Claude 反复调工具

Agent 循环的骨架是"call model → if tool_use then call tool → feed result back → repeat"：

async def chat_with_mcp_tools(    session: ClientSession,    user_message: str,    max_iterations: int = 10,) -> str:    """用 MCP 工具跟 Claude 对话，返回最终文本回答"""    import anthropic    client = anthropic.Anthropic()    # 1. 拉工具列表并转格式    tools_result = await session.list_tools()    anthropic_tools = [mcp_tool_to_anthropic(t) for t in tools_result.tools]    # 2. 初始化对话    messages: list[dict] = [{"role": "user", "content": user_message}]    for _ in range(max_iterations):        # 3. 调 Claude        response = client.messages.create(            model="claude-sonnet-4-6",            max_tokens=4096,            tools=anthropic_tools,            messages=messages,        )        # 4. 把 assistant 回复原样塞回对话历史        messages.append({"role": "assistant", "content": response.content})        # 5. 没有工具调用就结束        if response.stop_reason != "tool_use":            return "".join(                block.text for block in response.content                if block.type == "text"            )        # 6. 执行所有工具调用        tool_results: list[dict] = []        for block in response.content:            if block.type != "tool_use":                continue            print(f"[tool_use] {block.name}({block.input})")            mcp_result = await session.call_tool(                name=block.name,                arguments=block.input,            )            # 只取 text 内容，简化处理；生产里要处理多种 content block            result_text = "".join(                c.text for c in mcp_result.content                if getattr(c, "type", None) == "text"            )            tool_results.append({                "type": "tool_result",                "tool_use_id": block.id,                "content": result_text,                "is_error": mcp_result.isError,            })        # 7. 把工具结果塞回，进入下一轮        messages.append({"role": "user", "content": tool_results})    return "[reached max iterations without end_turn]"

几个容易踩坑的点：

• messages.append({"role": "assistant", "content": response.content})
   里的 response.content 是 SDK 的对象列表，不是字符串，必须原样塞回，不然 tool_use_id 会丢。
• tool_result
   里的 tool_use_id 必须跟对应的 tool_use.id 精确匹配，Anthropic API 会校验。
• max_iterations
   是硬上限，防止工具循环调用把 token 烧穿。生产里这个值要根据业务定，纯 QA 场景 5 就够，复杂 Agent 可能要 20+。

七、完整可运行代码：Server + Client 一起跑

把前面所有片段拼起来。两个文件，一个 Server（写数据库查询工具），一个 Client（连 Server + 接 Claude）。你把 ANTHROPIC_API_KEY 塞好就能跑。

7.1 server_demo.py：一个假数据库工具

"""极简 MCP Server：模拟数据库查询"""from mcp.server.fastmcp import FastMCPmcp = FastMCP("demo-db-server")# 假装这是数据库_FAKE_DB: dict[str, dict] = {    "u001": {"name": "Alice", "role": "engineer", "team": "infra"},    "u002": {"name": "Bob",   "role": "pm",       "team": "growth"},    "u003": {"name": "Carol", "role": "engineer", "team": "infra"},}@mcp.tool()def get_user(user_id: str) -> dict:    """按 user_id 查询用户信息。找不到时返回 {'error': ...}"""    user = _FAKE_DB.get(user_id)    if user is None:        return {"error": f"user_id {user_id} not found"}    return user@mcp.tool()def list_users_by_team(team: str) -> list[dict]:    """列出某个 team 下的所有用户"""    return [        {"user_id": uid, **info}        for uid, info in _FAKE_DB.items()        if info["team"] == team    ]if __name__ == "__main__":    mcp.run(transport="stdio")

7.2 client_demo.py：连 Server + 接 Claude

"""MCP Client：连接 server_demo.py，用 Claude 驱动工具调用"""import asyncioimport osimport sysimport anthropicfrom mcp import ClientSession, StdioServerParametersfrom mcp.client.stdio import stdio_clientdef mcp_tool_to_anthropic(mcp_tool) -> dict:    return {        "name": mcp_tool.name,        "description": mcp_tool.description or "",        "input_schema": mcp_tool.inputSchema,    }async def chat_with_mcp_tools(    session: ClientSession,    user_message: str,    max_iterations: int = 10,) -> str:    client = anthropic.Anthropic()    tools_result = await session.list_tools()    anthropic_tools = [mcp_tool_to_anthropic(t) for t in tools_result.tools]    print(f"[client] loaded {len(anthropic_tools)} tools from MCP server:")    for t in anthropic_tools:        print(f"  - {t['name']}: {t['description']}")    messages: list[dict] = [{"role": "user", "content": user_message}]    for turn in range(max_iterations):        response = client.messages.create(            model="claude-sonnet-4-6",            max_tokens=4096,            tools=anthropic_tools,            messages=messages,        )        messages.append({"role": "assistant", "content": response.content})        if response.stop_reason != "tool_use":            return "".join(                b.text for b in response.content if b.type == "text"            )        tool_results: list[dict] = []        for block in response.content:            if block.type != "tool_use":                continue            print(f"[turn {turn}] tool_use: {block.name}({block.input})")            mcp_result = await session.call_tool(                name=block.name,                arguments=block.input,            )            result_text = "".join(                c.text for c in mcp_result.content                if getattr(c, "type", None) == "text"            )            print(f"[turn {turn}] tool_result: {result_text}")            tool_results.append({                "type": "tool_result",                "tool_use_id": block.id,                "content": result_text,                "is_error": mcp_result.isError,            })        messages.append({"role": "user", "content": tool_results})    return "[reached max iterations without end_turn]"async def main() -> None:    server_params = StdioServerParameters(        command=sys.executable,          # 用当前 Python 解释器，避免 PATH 问题        args=["server_demo.py"],        env={"PATH": os.environ.get("PATH", "")},    )    async with stdio_client(server_params) as (read_stream, write_stream):        async with ClientSession(read_stream, write_stream) as session:            await session.initialize()            user_message = "帮我查一下 infra 团队都有谁，然后告诉我 u002 是谁。"            answer = await chat_with_mcp_tools(session, user_message)            print("\n=== FINAL ANSWER ===")            print(answer)if __name__ == "__main__":    asyncio.run(main())

跑起来：

export ANTHROPIC_API_KEY=sk-ant-...pip install "mcp[cli]" anthropicpython client_demo.py

你会看到类似输出：

[client] loaded 2 tools from MCP server:  - get_user: 按 user_id 查询用户信息。找不到时返回 {'error': ...}  - list_users_by_team: 列出某个 team 下的所有用户[turn 0] tool_use: list_users_by_team({'team': 'infra'})[turn 0] tool_result: [{"user_id": "u001", "name": "Alice", ...}, ...][turn 1] tool_use: get_user({'user_id': 'u002'})[turn 1] tool_result: {"name": "Bob", "role": "pm", "team": "growth"}=== FINAL ANSWER ===infra 团队有 Alice（engineer）和 Carol（engineer）；u002 是 Bob，PM，属于 growth 团队。

从代码到跑通，你自己完全掌控了 Client 侧——Claude Desktop 只是众多 Host 中的一种，你的 Python 脚本也可以是。

八、工程师反思

写完这篇，做几点诚实的自我审查。

1. 单 Server 假设是被简化的

上面的完整示例只连了一个 Server。真实场景常常是"一个 Agent 同时连 3 个 Server"：一个连 CRM、一个连日历、一个连内部知识库。这时候你需要维护多个 ClientSession，工具列表要合并、工具名要防冲突（比如两个 Server 都有 search 工具就得加前缀）。这一块留到后面"多 Server 编排"篇专门讲，本篇没铺开是刻意为之——先把单连接讲透。

2. tool_result 的内容处理被简化了

代码里我只把 content 里的 text 块拼起来当结果，忽略了 image、resource 等其他块类型。生产里如果工具返回图片（比如画个图表返回 PNG），这里会丢信息。修法是把 tool_result 的 content 也做成结构化的 block 列表，Anthropic API 支持这样传。

3. 没有错误重试与超时

stdio_client 启动 Server 失败（比如可执行文件不存在）、session.call_tool 中途 Server 挂掉、Claude API 限流——这三类错误在示例里都会直接崩溃。生产里至少要给 call_tool 加 asyncio.wait_for 超时，给 Anthropic 调用加指数退避重试。这块用 tenacity 库套一下就行，本文没写是为了让主逻辑更清晰。

4. env=None 在 stdio 里的坑

MCP SDK 早期版本里，StdioServerParameters(env=None) 会导致子进程完全没有环境变量，python 命令都可能找不到。我用 sys.executable 做 command、显式传 PATH 是对这个坑的绕过。如果你的 Server 需要读 ANTHROPIC_API_KEY、数据库连接串等敏感环境变量，记得在 env dict 里显式带上，不要指望自动继承。

5. 为什么强调"自己写 Client"

有读者可能会问：既然 Claude Desktop、Cursor、Cline 都能当 MCP Host，为什么还要自己写？答案是——MCP 的目标是"能力协议化"。你把公司的内部工具做成 MCP Server，Claude Desktop 可以接、Agent Pipeline 可以接、Slack Bot 可以接、CI/CD 脚本也可以接。写 Client 不是替代 Claude Desktop，是在 Claude Desktop 覆盖不到的场景里让 MCP Server 继续产生价值。

下一篇：「用 MCP 接数据库：把 SQL 能力安全地暴露给 LLM」——真的接一个 PostgreSQL 或 SQLite，讲连接池、只读隔离、SQL 注入防护、结果集分页。从 demo 迈向能进生产的第一个真实 Server。

*《MCP 工程实战》系列每两天更新一篇，下篇预告：*

*05 — MCP × 数据库：把 SQL 能力安全地暴露给 LLM*