为了让你直观地理解 MCP Client 是如何作为“桥梁”工作的，我为你准备了两个不同维度的代码示例。

第一个示例使用 Python 的 mcp 官方库，展示了 Client 如何“发现能力”和“调用工具”（即标准的请求-响应流程）；

第二个示例则展示了 Client 如何处理“服务端推送”（即监听 Server 主动发出的日志通知），这能很好地体现 Client 的异步事件处理能力。

🛠️ 示例一：核心功能实现（Python）

这个脚本模拟了一个 MCP Client 的生命周期：启动 Server -> 握手 -> 获取工具列表 -> 调用工具 -> 关闭。

python

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

# 1. 定义 Server 的启动参数
# 这里我们假设有一个名为 "mcp-server-time" 的工具，用于获取时间
server_params = StdioServerParameters(
    command="python",
    args=["-m", "mcp_server_time"], 
    env=None
)

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 的事件驱动特性。

python

import asyncio
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
from mcp.types import ServerNotification

# 假设这是一个会打印日志的 Server
server_params = StdioServerParameters(command="python", 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 如何通过几十行代码，就构建起连接大模型与外部世界的桥梁。