为了让你直观地理解MCP Client是如何作为“桥梁”工作的我为你准备了两个不同维度的代码示例。第一个示例使用 Python 的mcp官方库展示了 Client 如何“发现能力”和“调用工具”即标准的请求-响应流程第二个示例则展示了 Client 如何处理“服务端推送”即监听 Server 主动发出的日志通知这能很好地体现 Client 的异步事件处理能力。️ 示例一核心功能实现Python这个脚本模拟了一个 MCP Client 的生命周期启动 Server - 握手 - 获取工具列表 - 调用工具 - 关闭。pythonimport asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client # 1. 定义 Server 的启动参数 # 这里我们假设有一个名为 mcp-server-time 的工具用于获取时间 server_params StdioServerParameters( commandpython, args[-m, mcp_server_time], envNone ) async def run(): # 2. 建立连接 (Stdio 模式) # stdio_client 会启动子进程并接管其 stdin/stdout async with stdio_client(server_params) as (read, write): # 3. 创建会话并初始化 # 这里会自动发送 initialize 消息并等待 Server 响应 async with ClientSession(read, write) as session: # 4. 【桥梁作用能力发现】 # Client 询问 Server你都会什么 (对应 tools/list) tools await session.list_tools() print(f 发现可用工具: {[tool.name for tool in tools.tools]}) # 5. 【桥梁作用指令路由】 # 假设我们要调用名为 get_current_time 的工具 # Client 将参数封装为 JSON-RPC 请求发送给 Server try: result await session.call_tool(get_current_time, arguments{timezone: Asia/Shanghai}) # 6. 【桥梁作用结果回传】 # 接收 Server 的 JSON 响应并提取内容 print(f✅ 工具执行结果: {result.content[0].text}) except Exception as e: print(f❌ 调用失败: {e}) if __name__ __main__: # 运行异步主程序 asyncio.run(run())代码解析stdio_client这是 Client 的传输层负责启动 Server 进程并建立管道。ClientSession这是 Client 的协议层负责处理握手initialize和会话管理。list_tools()对应“侦察兵”机制获取 JSON Schema 定义。call_tool()对应“执行器”机制发送tools/call请求。 示例二处理服务端推送监听日志MCP 允许 Server 主动向 Client 发送通知例如日志或进度更新。Client 需要注册回调函数来监听这些事件这体现了 Client 的事件驱动特性。pythonimport asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client from mcp.types import ServerNotification # 假设这是一个会打印日志的 Server server_params StdioServerParameters(commandpython, args[my_logging_server.py]) async def handle_notification(notification: ServerNotification): 当 Server 主动推送消息时Client 会触发此回调 print(f 收到 Server 推送通知: {notification.method}) if notification.method notifications/message: print(f 日志内容: {notification.params.data}) async def run_with_events(): async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: # 【桥梁作用事件监听】 # Client 注册一个钩子专门监听 Server 的日志通知 session.set_notification_handler(handle_notification) # 调用一个会触发日志的工具 print(正在调用工具等待 Server 推送日志...) await session.call_tool(start_long_task, arguments{}) # 保持运行一段时间以接收推送 await asyncio.sleep(2) if __name__ __main__: asyncio.run(run_with_events()) 总结通过这两个示例你可以清晰地看到 MCP Client 的代码结构是如何对应其架构角色的传输层stdio_client或sse_client负责底层的字节流读写。会话层ClientSession负责维护状态和协议握手。业务层list_tools、call_tool以及set_notification_handler负责具体的业务交互。这就是 MCP Client 如何通过几十行代码就构建起连接大模型与外部世界的桥梁。