夜雨聆风前两篇文章，我们分别介绍了MCP的概念与架构，以及如何写MCP服务器。
今天这篇是系列第三篇——MCP客户端开发。搞定这一篇，你就打通了MCP的完整链路：
服务器（暴露能力）← 客户端（连接调用）← AI模型（理解执行）
准备好了吗？开始！
一、什么是MCP客户端？它在整个架构中扮演什么角色？
先搞清楚概念：
| MCP服务器 | ||
| MCP客户端 | ||
| Host（宿主应用） |
一句话概括：
MCP客户端是AI应用与MCP服务器之间的"桥梁"。它负责连接服务器、发现能力、转发工具调用、处理返回结果。
客户端的核心职责
1. 连接管理 — 启动服务器进程，建立通信通道 2. 能力发现 — 列出服务器提供的工具、资源、提示词 3. 请求转发 — 将AI模型的意图转化为MCP协议请求 4. 响应处理 — 将服务器结果转回AI模型可理解的内容 5. 会话管理 — 维护对话上下文，支持多轮交互
客户端运行在哪里？
MCP客户端通常嵌入在Host应用中：
• Claude Desktop — Anthropic官方桌面客户端 • VS Code / Cursor — IDE内置的MCP支持 • 自定义聊天机器人 — 你自己写的Python/TypeScript程序 • 企业AI平台 — 集成MCP的内部系统
二、从零开始：Python实现完整聊天机器人
这是官方教程推荐的入门方式，代码量适中，功能完整。
📋 环境准备
# 1. 创建项目目录mkdir mcp-client && cd mcp-client# 2. 安装MCP SDK和依赖uv pip install "mcp[cli]" anthropic httpx要求： Python 3.10+，最新版本的 uv
🔧 完整代码：MCP聊天机器人客户端
创建 client.py 文件：
import asyncioimport sysfrom contextlib import AsyncExitStackfrom anthropic import AsyncAnthropicfrom mcp.shared.session import Sessionfrom mcp.client.stdio import StdioClientSession, StdioServerParametersclass MCPClient: """MCP客户端 - 连接服务器并调用工具""" def __init__(self): # 会话管理 self.exit_stack = AsyncExitStack() self.session: Session | None = None # AI模型客户端（以Claude为例） self.anthropic = AsyncAnthropic() self.messages = [] # 对话历史 # ===================== # 1. 连接到MCP服务器 # ===================== async def connect_to_server(self, server_script_path: str): """连接到MCP服务器""" # 判断服务器脚本类型 is_python = server_script_path.endswith('.py') is_js = server_script_path.endswith('.js') if not (is_python or is_js): raise ValueError("服务器脚本必须是 .py 或 .js 文件") # 设置启动命令 command = "python" if is_python else "node" server_params = StdioServerParameters( command=command, args=[server_script_path], env=None ) # 建立stdio通信并初始化会话 stdio_transport = await self.exit_stack.enter_async_context( asyncio.create_task( StdioClientSession(server_params) ) ) _, session_service = await stdio_transport.__aenter__() self.session = await session_service # 列出可用工具 response = await self.session.list_tools() tools = response.tools print(f"\n✅ 已连接到服务器，可用工具：") for tool in tools: print(f" 🔧 {tool.name}: {tool.description}") # ===================== # 2. 处理对话循环 # ===================== async def chat_loop(self): """交互式对话循环""" print("\n💬 开始对话！（输入 'quit' 退出）") while True: try: user_input = input("\n👤 你：").strip() if user_input.lower() in ('quit', 'exit', '退出'): break await self.handle_query(user_input) except KeyboardInterrupt: break # ===================== # 3. 处理用户查询（核心逻辑） # ===================== async def handle_query(self, query: str): """处理用户查询，调用AI模型并执行工具""" self.messages.append({ "role": "user", "content": query }) # 将MCP工具格式化为Claude API需要的格式 mcp_tools = [] if self.session: tools_response = await self.session.list_tools() for tool in tools_response.tools: mcp_tools.append({ "name": tool.name, "description": tool.description, "input_schema": tool.inputSchema }) # 调用AI模型 response = await self.anthropic.messages.create( model="claude-sonnet-4-0", max_tokens=4096, messages=self.messages, tools=mcp_tools if mcp_tools else None ) # 处理AI响应 assistant_text = [] tool_results = [] has_text_response = False for content in response.content: if content.type == "text": assistant_text.append(content.text) print(f"\n🤖 AI：{content.text}") has_text_response = True elif content.type == "tool_use": tool_name = content.name tool_args = content.input # 执行工具调用 result = await self.session.call_tool(tool_name, tool_args) tool_results.append({ "call": tool_name, "result": result }) print(f"\n🔧 [调用工具: {tool_name}, 参数: {tool_args}]") # 显示工具结果 for content_item in result.content: if content_item.type == "text": print(f"📋 工具结果：{content_item.text}") # 如果调用了工具，将结果发回给AI进行二次处理 if tool_results: self.messages.append({ "role": "assistant", "content": assistant_text if assistant_text else [ {"type": "tool_use", "name": r["call"], "input": {}, "id": "tool_call_1"} ] }) # 将工具结果作为用户消息返回给AI tool_result_msg = { "role": "user", "content": [ {"type": "tool_result", "content": r["result"].content} for r in tool_results ] } self.messages.append(tool_result_msg) # 二次调用AI模型，生成最终回答 final_response = await self.anthropic.messages.create( model="claude-sonnet-4-0", max_tokens=4096, messages=self.messages ) for content in final_response.content: if content.type == "text": print(f"\n🤖 AI最终回答：{content.text}") # ===================== # 4. 资源清理 # ===================== async def cleanup(self): """清理资源，断开连接""" await self.exit_stack.aclose() print("\n👋 已断开服务器连接")# =====================# 程序入口# =====================async def main(): if len(sys.argv) < 2: print("用法: python client.py <服务器脚本路径>") sys.exit(1) client = MCPClient() try: await client.connect_to_server(sys.argv[1]) await client.chat_loop() finally: await client.cleanup()if __name__ == "__main__": import asyncio asyncio.run(main())🚀 运行客户端
# 假设你有一个服务器脚本 weather.pypython client.py ./weather.py交互效果：
✅ 已连接到服务器，可用工具： 🔧 get-forecast: 获取指定位置的天气预报 🔧 get-alerts: 获取指定位置的天气警报信息💬 开始对话！（输入 'quit' 退出）👤 你：北京的天气怎么样？🤖 AI：让我帮你查一下北京的天气...🔧 [调用工具: get-forecast, 参数: {"latitude": 39.9, "longitude": 116.4}]📋 工具结果：天气预报：晴转多云，气温25-30度...🤖 AI最终回答：北京今天的天气是晴转多云，气温在25-30度之间，适合户外活动。三、关键组件详解
1️⃣ 客户端初始化
class MCPClient: def __init__(self): self.exit_stack = AsyncExitStack() # 资源管理 self.session: Session | None = None # MCP会话 self.anthropic = AsyncAnthropic() # AI模型客户端 self.messages = [] # 对话历史关键组件：
• AsyncExitStack— Python异步上下文管理器，确保资源正确释放• Session— MCP会话对象，管理连接和通信• messages— 对话历史列表，维持多轮对话的上下文
2️⃣ 服务器连接
async def connect_to_server(self, server_script_path: str): # 1. 判断脚本类型（.py → python / .js → node） command = "python" if is_python else "node" # 2. 配置服务器参数 server_params = StdioServerParameters( command=command, args=[server_script_path], env=None ) # 3. 建立stdio传输通道 stdio_transport = await self.exit_stack.enter_async_context(...) # 4. 获取会话对象 self.session = await session_service # 5. 列出可用工具 response = await self.session.list_tools()核心API：
StdioServerParameters | |
StdioClientSession | |
session.initialize() | |
session.list_tools() |
3️⃣ 工具调用流程
用户输入 → AI模型分析意图 → 决定调用工具 → MCP客户端执行 ↓ 服务器返回结果 → 客户端转发给AI ↓ AI结合结果生成最终回答关键代码：
# 将MCP工具格式化为Claude API需要的格式mcp_tools = []for tool in tools_response.tools: mcp_tools.append({ "name": tool.name, # 工具名称 "description": tool.description, # 工具描述 "input_schema": tool.inputSchema # 参数Schema })# 调用AI模型，传入工具列表response = await self.anthropic.messages.create( model="claude-sonnet-4-0", tools=mcp_tools, # ← 告诉AI有哪些工具可用 messages=self.messages)# 如果AI决定调用工具if content.type == "tool_use": result = await self.session.call_tool(tool_name, tool_args)4️⃣ 资源（Resources）访问
除了工具，客户端还可以访问服务器暴露的资源：
# 列出可用资源resources = await self.session.list_resources()# 读取特定资源内容content, mime_type = await self.session.read_resource("file:///some/path")5️⃣ 提示词（Prompts）使用
# 列出可用提示词模板prompts = await self.session.list_prompts()# 获取具体提示词（带参数）prompt_result = await self.session.get_prompt("security-review", { "code": "..."})四、TypeScript实现（前端开发者必看）
如果你更习惯JavaScript/TypeScript，SDK也提供了完整的TS支持：
📋 环境准备
mkdir mcp-client-ts && cd mcp-client-tsnpm init -ynpm install @modelcontextprotocol/sdk node-fetch🔧 TypeScript完整代码
创建 client.ts：
import { Client } from "@modelcontextprotocol/sdk/client/index.js";import { StdioTransport } from "@modelcontextprotocol/sdk/client/stdio.js";import fetch from "node-fetch";async function main() { // 1. 创建客户端实例 const client = new Client({ name: "my-mcp-client", version: "1.0.0" }); // 2. 连接服务器（stdio方式） const transport = new StdioTransport({ command: "node", args: ["./weather.js"] }); await client.connect(transport); // 3. 列出可用工具 const tools = await client.listTools(); console.log("✅ 可用工具："); tools.tools.forEach(tool => { console.log(` 🔧 ${tool.name}: ${tool.description}`); }); // 4. 调用工具 const result = await client.callTool({ name: "get-forecast", arguments: { latitude: 39.9, longitude: 116.4 } }); console.log("\n📋 工具结果："); result.content.forEach(item => { if (item.type === "text") { console.log(item.text); } }); // 5. 断开连接 await client.close();}main().catch(console.error);⚡ TypeScript vs Python对比
| SDK | mcp | @modelcontextprotocol/sdk |
| 连接方式 | StdioClientSession | StdioTransport |
| 列出工具 | await session.list_tools() | await client.listTools() |
| 调用工具 | await session.call_tool(name, args) | await client.callTool({name, arguments}) |
| 资源管理 | AsyncExitStackaclose() | client.close() |
五、两种传输方式：Stdio vs SSE/HTTP
📡 Stdio（标准输入输出）
适用场景： 本地服务器，客户端和服务器在同一台机器上
# Pythonserver_params = StdioServerParameters( command="python", args=["./weather.py"])优点： 简单直接，零网络延迟 限制： 只能连接本地服务器
🌐 SSE（Server-Sent Events / HTTP）
适用场景： 远程服务器，需要跨机器通信
// TypeScript - SSE传输方式import { SSEClientTransport } from "@modelcontextprotocol/sdk/client/sse.js";const transport = new SSEClientTransport( new URL("http://remote-server:3000/mcp"));await client.connect(transport);优点： 支持远程连接，可跨网络 限制： 需要服务器支持HTTP/SSE传输
| 通信方式 | ||
| 适用场景 | ||
| 配置复杂度 | ||
| 延迟 |
六、客户端开发的完整流程总结
构建一个MCP客户端的核心流程如下：
第一步：初始化客户端
• Python: MCPClient()+AsyncExitStack• TypeScript: new Client({...})
第二步：连接服务器
• 配置服务器参数（命令、路径） • 选择传输方式（stdio或SSE） • 建立通信通道并初始化会话
第三步：发现能力
• list_tools()— 列出可用工具• list_resources()— 列出可用资源• list_prompts()— 列出提示词模板
第四步：调用工具/资源
# 调用工具result = await session.call_tool("get-forecast", { "latitude": 39.9, "longitude": 116.4})# 读取资源content, mime_type = await session.read_resource("file:///path/to/file")第五步：整合AI模型
• 将工具列表格式化为AI API需要的格式 • 调用AI模型，传入对话历史和工具定义 • 解析AI响应，判断是否需要调用工具 • 执行工具调用，将结果返回给AI生成最终回答
第六步：资源清理
async def cleanup(self): await self.exit_stack.aclose() # Python七、常见坑和注意事项
⚠️ 服务器路径问题
错误写法：
# Windows - 反斜杠可能导致问题python client.py C:\projects\weather.py正确写法：
# 方式1：正斜杠（推荐）python client.py C:/projects/weather.py# 方式2：转义反斜杠python client.py C:\\projects\\weather.py# 方式3：绝对路径 + 引号python client.py "C:\projects\weather.py"⚠️ 首次响应延迟
现象： 第一次对话可能需要 10-30秒 才有响应
原因：
1. 服务器进程初始化 2. AI模型处理查询 3. 工具执行时间
建议： 在初始等待期间不要中断进程，后续响应会快很多。
⚠️ API Key配置
# Claude API（示例）export ANTHROPIC_API_KEY="sk-ant-xxx"# OpenAI（如果用GPT的话）export OPENAI_API_KEY="sk-xxx"⚠️ 错误处理
MCP协议规定： 工具级别的错误应在结果对象内报告，而不是作为协议级错误。
# ✅ 正确做法：在内容中返回错误return { "content": [{"type": "text", "text": "操作失败：xxx"}], "isError": True}八、不同AI平台的客户端实现
Claude Desktop（无需写代码）
配置文件：~/Library/Application Support/Claude/claude_desktop_config.json
{ "mcpServers": { "weather": { "command": "python", "args": ["/absolute/path/to/weather.py"] } }}VS Code / Cursor（内置支持）
在设置中添加MCP服务器配置即可，无需额外编码。
Google Gemini集成
from google import genai# 将MCP工具转换为Gemini格式mcp_tools = await session.list_tools()gemini_tools = types.Tool(function_declarations=[ { "name": tool.name, "description": tool.description, "parameters": tool.inputSchema } for tool in mcp_tools.tools])# 调用Gemini APIresponse = await client.aio.models.generate_content( contents=[...], tools=[gemini_tools])九、总结
MCP客户端开发的本质就是做一个"翻译官"：
| 对AI模型 | |
| 对服务器 | |
| 对用户 |
核心API总结：
客户端生命周期：├── 初始化 → Client / MCPClient├── 连接 → connect() / StdioTransport├── 发现能力 → list_tools() / list_resources() / list_prompts()├── 调用工具 → call_tool(name, args)├── 读取资源 → read_resource(uri)└── 清理 → close() / aclose()掌握了MCP客户端开发，你就拥有了连接任何MCP服务器的能力。一次编写，到处调用——这就是MCP的魅力。
下一步探索：
• 📚 深入阅读官方SDK文档，了解所有API细节 • 🔌 尝试HTTP/SSE传输，实现远程服务器连接 • 🏗️ 将客户端嵌入你自己的应用（Web、桌面、移动端） • 🧩 同时连接多个MCP服务器，让AI拥有更多能力
现在就开始动手吧——你的第一个MCP聊天机器人，10分钟就能跑起来！
📢 关注"大强哥爱编程"公众号，获取更多AI技术前沿资讯！
🌟 喜欢这篇文章？请点赞、转发、收藏！有任何问题或建议，欢迎在评论区留言交流！