Python 客户端连接服务器

服务器写好后，还得有"客户端"去消费它的工具。官方 Python SDK（包名 mcp）同时提供 FastMCP（写服务器）和一套异步客户端：本地用 stdio 传输，远程用 Streamable HTTP。本章给出两端都能跑通的最小示例。

安装官方 SDK

pip install "mcp[cli]"      # 官方 Python SDK；[cli] 额外带上 mcp 命令行工具

客户端 API 随 SDK 版本快速演进，写法如有出入，一律以 modelcontextprotocol.io 的官方 SDK 文档为准。

准备一个测试服务器

先写一个极简单的 stdio 服务器，后面所有客户端都连它：

# server_echo.py：一个只提供 add 工具的 FastMCP 服务器
from mcp.server.fastmcp import FastMCP

mcp = FastMCP("echo")

@mcp.tool()
def add(a: int, b: int) -> int:
    """两数相加"""
    return a + b

if __name__ == "__main__":
    mcp.run()       # FastMCP 默认以 stdio 方式运行

用 stdio_client 连接本地服务器

stdio 客户端把"命令行 + 参数"描述进 StdioServerParameters，SDK 负责拉起子进程并与之双向通信：

# client_stdio.py：完整可运行的 stdio 客户端
import asyncio
import sys
from mcp import ClientSession
from mcp.client.stdio import StdioServerParameters, stdio_client

async def main():
    params = StdioServerParameters(
        command=sys.executable,   # 用当前 Python 解释器去启动服务器
        args=["server_echo.py"],  # 服务器脚本路径（与本文件同目录）
    )
    async with stdio_client(params) as (read, write):    # 建立两条传输流
        async with ClientSession(read, write) as session:
            await session.initialize()            # ① 握手初始化（必须先做）
            tools = await session.list_tools()    # ② 列出服务器声明的工具
            print(f"可用工具：{[t.name for t in tools.tools]}")
            # 输出：可用工具：['add']
            result = await session.call_tool(     # ③ 调用工具，实参按 JSON Schema 传
                "add", {"a": 2, "b": 3}
            )
            print(f"返回：{result.content[0].text}")   # 输出：返回：5

asyncio.run(main())

要点：stdio_client 负责进程生命周期，ClientSession 负责协议；list_tools 返回工具清单（含名称、描述、inputSchema），call_tool 的第一个参数是工具名、第二个是参数字典，结果内容在 result.content 的 TextContent 块里。

连接远程 Streamable HTTP 服务器

服务器通过 HTTP 部署后（见第 15 章），客户端换成 streamablehttp_client，其余流程一致。官方已弃用旧的 SSE 传输，新项目一律用 Streamable HTTP：

# client_http.py：连接远程 Streamable HTTP 服务器
import asyncio
from mcp import ClientSession
from mcp.client.streamable_http import streamablehttp_client

URL = "http://127.0.0.1:8000/mcp"    # 远程服务地址，需先按第 15 章部署

async def main():
    async with streamablehttp_client(URL) as (read, write, _get_session_id):
        async with ClientSession(read, write) as session:
            await session.initialize()
            result = await session.call_tool("add", {"a": 10, "b": 5})
            print(f"远程返回：{result.content[0].text}")   # 输出：远程返回：15

asyncio.run(main())

两点提示：个别 SDK 版本的 streamablehttp_client 只返回 (read, write) 两条流，按你安装的版本解包即可；需要带鉴权头时给 streamablehttp_client 传 headers 参数，字段名以官方文档为准。 小结：客户端三板斧是 stdio_client（或 streamablehttp_client）→ ClientSession → initialize，之后 list_tools 看工具、call_tool 调工具；stdio 连本机子进程、Streamable HTTP 连远程服务，SSE 传输已弃用。