企业官网建设流程全解析

MCP Client 集成：在 Agent 中接入 MCP 工具

摘要：MCP Client 是连接 LLM 应用与 MCP Server 的桥梁。本文深入讲解 Client 的连接管理、工具发现与调用、错误处理机制，并通过一个完整的 Agent 示例展示如何将 MCP 工具集成到实际应用中。

一、前言

如果说 MCP Server 是工具的"提供方"，那么 MCP Client 就是工具的"消费方"。在 MCP 的三方架构（Host-Client-Server）中，Client 承担了最核心的职责：

1. 连接管理：建立和维护与 Server 的通信连接
2. 能力协商：在初始化阶段与 Server 交换能力信息
3. 工具发现：动态获取 Server 暴露的工具列表
4. 工具调用：将 LLM 的决策转化为具体的 MCP 请求
5. 结果处理：将 Server 返回的结果反馈给 LLM

对于 Agent 开发者而言，理解 MCP Client 的工作原理，是将 MCP 生态能力融入自己应用的关键。

二、Client SDK 概览

2.1 Python Client SDK

MCP Python SDK 同时提供了 Server 和 Client 的实现。Client 端的核心类是ClientSession：

frommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_client

2.2 Client 类型

根据传输方式的不同，SDK 提供了多种 Client 连接方式：

2.3 TypeScript/Node.js Client SDK

对于 TypeScript/Node.js 生态：

import{Client}from"@modelcontextprotocol/sdk/client/index.js";import{StdioClientTransport}from"@modelcontextprotocol/sdk/client/stdio.js";

三、连接管理

3.1 Stdio 连接（本地 Server）

Stdio 连接是最简单的方式，适合本地部署的 MCP Server：

# mcp_client_demo.py# MCP Client 基础示例 —— 展示连接管理与工具调用的核心流程importasynciofrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientasyncdefmain():# 定义 Server 启动参数# StdioServerParameters 描述了如何启动一个本地 MCP Server 进程server_params=StdioServerParameters(command="python",# 启动命令args=["-m","dev_assistant.server"],# 命令参数env=None,# 环境变量（None 表示继承当前环境）)# 使用 async context manager 建立连接# stdio_client 会启动 Server 进程并建立 stdin/stdout 通信asyncwithstdio_client(server_params)as(read_stream,write_stream):# 创建 ClientSession，传入读写流asyncwithClientSession(read_stream,write_stream)assession:# === 步骤 1: 初始化握手 ===# 发送 initialize 请求，交换能力信息awaitsession.initialize()print("✅ 连接建立成功")# === 步骤 2: 工具发现 ===# 调用 tools/list 获取 Server 暴露的工具列表tools_response=awaitsession.list_tools()print(f"\n📋 发现{len(tools_response.tools)}个工具:")fortoolintools_response.tools:print(f" -{tool.name}:{tool.description}")# === 步骤 3: 工具调用 ===# 调用具体工具并获取结果result=awaitsession.call_tool("validate_json",arguments={"json_string":'{"name": "MCP", "version": "1.0"}'})print(f"\n🔧 工具调用结果:")forcontentinresult.content:print(f"{content.text}")# === 步骤 4: 资源访问 ===# 列出可用资源resources=awaitsession.list_resources()print(f"\n📚 发现{len(resources.resources)}个资源:")forresourceinresources.resources:print(f" -{resource.uri}:{resource.name}")# 读取指定资源resource_content=awaitsession.read_resource("snippets://library")print(f"\n📖 资源内容:")forcontentinresource_content.contents:print(f"{content.text[:200]}...")# 运行 Clientasyncio.run(main())

代码解读：

1. StdioServerParameters封装了 Server 进程的启动参数，stdio_client会自动管理进程生命周期
2. ClientSession是 Client 的核心，它封装了所有协议交互逻辑
3. initialize()完成握手后，Client 才能调用其他方法
4. list_tools()返回的工具列表包含名称、描述和参数 Schema，这些信息可以直接传给 LLM

3.2 连接生命周期管理

# 连接状态管理示例fromcontextlibimportasynccontextmanager@asynccontextmanagerasyncdefmanaged_mcp_connection(server_params):"""管理 MCP 连接的完整生命周期"""print("🔌 正在建立连接...")try:asyncwithstdio_client(server_params)as(read,write):asyncwithClientSession(read,write)assession:awaitsession.initialize()print("✅ 连接就绪")yieldsessionexceptExceptionase:print(f"❌ 连接异常:{e}")raisefinally:print("🔌 连接已关闭")

四、工具发现与调用

4.1 工具 Schema 解析

MCP Server 返回的工具信息包含完整的 JSON Schema，这对 LLM 工具调用至关重要：

asyncdefparse_tool_schema(session:ClientSession):"""解析工具 Schema，提取 LLM 可用的工具描述"""tools_response=awaitsession.list_tools()llm_tools=[]fortoolintools_response.tools:# 将 MCP 工具描述转换为 LLM Function Calling 格式llm_tool={"type":"function","function":{"name":tool.name,"description":tool.descriptionor"","parameters":tool.inputSchemaor{"type":"object","properties":{}}}}llm_tools.append(llm_tool)returnllm_tools

4.2 工具调用与结果解析

asyncdefcall_tool_safely(session:ClientSession,tool_name:str,arguments:dict,timeout:float=30.0)->dict:""" 安全地调用 MCP 工具，包含超时控制和错误处理。 Args: session: MCP Client 会话 tool_name: 工具名称 arguments: 工具参数 timeout: 超时时间（秒） Returns: 包含 success、result、error 的字典 """try:# 使用 asyncio.wait_for 设置超时result=awaitasyncio.wait_for(session.call_tool(tool_name,arguments),timeout=timeout)# 检查是否为错误结果ifresult.isError:error_text=" ".join(c.textforcinresult.contentifhasattr(c,'text'))return{"success":False,"result":None,"error":f"工具执行失败:{error_text}"}# 提取文本内容text_contents=[c.textforcinresult.contentifhasattr(c,'text')]return{"success":True,"result":"\n".join(text_contents),"error":None}exceptasyncio.TimeoutError:return{"success":False,"result":None,"error":f"工具调用超时 ({timeout}s)"}exceptExceptionase:return{"success":False,"result":None,"error":f"工具调用异常:{str(e)}"}

五、错误处理

5.1 错误类型分类

MCP Client 需要处理以下几类错误：

5.2 重连机制

importasynciofromtypingimportOptionalclassResilientMCPClient:"""具备重连能力的 MCP Client"""def__init__(self,server_params,max_retries=3,retry_delay=2.0):self.server_params=server_params self.max_retries=max_retries self.retry_delay=retry_delay self.session:Optional[ClientSession]=Noneself._connection_context=Noneasyncdefconnect(self)->bool:"""建立连接，支持重试"""forattemptinrange(self.max_retries):try:self._connection_context=stdio_client(self.server_params)read,write=awaitself._connection_context.__aenter__()self.session=ClientSession(read,write)awaitself.session.__aenter__()awaitself.session.initialize()returnTrueexceptExceptionase:print(f"连接失败 (尝试{attempt+1}/{self.max_retries}):{e}")ifattempt<self.max_retries-1:awaitasyncio.sleep(self.retry_delay*(2**attempt))returnFalseasyncdefcall_tool_with_retry(self,tool_name:str,arguments:dict)->dict:"""调用工具，连接断开时自动重连"""forattemptinrange(2):# 最多重试 1 次（首次 + 1 次重连）try:ifself.sessionisNone:ifnotawaitself.connect():return{"success":False,"error":"无法建立连接"}result=awaitself.session.call_tool(tool_name,arguments)return{"success":True,"result":result}except(ConnectionError,BrokenPipeError):print("连接断开，尝试重连...")self.session=NonecontinueexceptExceptionase:return{"success":False,"error":str(e)}return{"success":False,"error":"重试次数耗尽"}

六、完整 Agent 集成示例

以下展示如何将 MCP 工具集成到一个简单的 Agent 中：

# mcp_agent.py# 基于 MCP 的简单 Agent 实现 —— 展示完整的 LLM + MCP 工具调用流程importasyncioimportjsonfrommcpimportClientSession,StdioServerParametersfrommcp.client.stdioimportstdio_clientclassMCPAgent:""" 基于 MCP 的 Agent 将 MCP Server 暴露的工具集成到 LLM 对话流程中 """def__init__(self,server_configs:list[dict]):""" 初始化 Agent Args: server_configs: MCP Server 配置列表，每个元素包含: - name: Server 名称 - command: 启动命令 - args: 命令参数 """self.server_configs=server_configs self.sessions:dict[str,ClientSession]={}self.available_tools:list[dict]=[]asyncdefstart(self):"""启动所有 MCP Server 连接"""forconfiginself.server_configs:name=config["name"]params=StdioServerParameters(command=config["command"],args=config["args"],)try:# 建立连接read,write=awaitstdio_client(params).__aenter__()session=ClientSession(read,write)awaitsession.__aenter__()awaitsession.initialize()self.sessions[name]=session# 发现工具tools_resp=awaitsession.list_tools()fortoolintools_resp.tools:self.available_tools.append({"server":name,"name":tool.name,"description":tool.description,"schema":tool.inputSchema,})print(f"✅ 已连接:{name}({len(tools_resp.tools)}个工具)")exceptExceptionase:print(f"❌ 连接失败{name}:{e}")asyncdefhandle_tool_call(self,tool_name:str,arguments:dict)->str:""" 处理 LLM 发出的工具调用请求 Args: tool_name: 工具名称 arguments: 工具参数 Returns: 工具执行结果文本 """# 查找工具所属的 Serverfortool_infoinself.available_tools:iftool_info["name"]==tool_name:server_name=tool_info["server"]session=self.sessions.get(server_name)ifsession:try:result=awaitsession.call_tool(tool_name,arguments)ifresult.isError:returnf"工具执行错误:{result.content}"return"\n".join(c.textforcinresult.contentifhasattr(c,'text'))exceptExceptionase:returnf"工具调用异常:{e}"returnf"未找到工具:{tool_name}"defget_tools_for_llm(self)->list[dict]:""" 获取 LLM Function Calling 格式的工具列表 Returns: 符合 OpenAI/Anthropic Function Calling 格式的工具描述列表 """llm_tools=[]fortoolinself.available_tools:llm_tools.append({"type":"function","function":{"name":tool["name"],"description":tool["description"]or"","parameters":tool["schema"]or{"type":"object","properties":{}}}})returnllm_toolsasyncdefshutdown(self):"""关闭所有连接"""forname,sessioninself.sessions.items():try:awaitsession.__aexit__(None,None,None)print(f"🔌 已断开:{name}")exceptException:passasyncdefdemo():"""演示 Agent 的完整工作流程"""# 配置 MCP Serverconfigs=[{"name":"dev-assistant","command":"python","args":["-m","dev_assistant.server"],}]agent=MCPAgent(configs)awaitagent.start()# 模拟 LLM 决策后的工具调用print("\n🤖 LLM 决定调用 validate_json 工具...")result=awaitagent.handle_tool_call("validate_json",{"json_string":'{"status": "ok", "data": [1, 2, 3]}'})print(f"📊 工具返回:{result}")# 展示可用工具列表tools=agent.get_tools_for_llm()print(f"\n🔧 可用工具 ({len(tools)}个):")fortintools:print(f" -{t['function']['name']}")awaitagent.shutdown()if__name__=="__main__":asyncio.run(demo())

代码解读：

1. MCPAgent类管理多个 MCP Server 连接，实现了工具的统一发现和调用
2. get_tools_for_llm()将 MCP 工具转换为标准的 Function Calling 格式，可以直接传给 LLM API
3. handle_tool_call()根据工具名自动路由到对应的 Server，对 LLM 透明
4. 完善的生命周期管理（start/shutdown）确保资源正确释放

七、集成最佳实践

7.1 工具选择策略

当连接了多个 MCP Server（数十甚至上百个工具）时，如何让 LLM 高效选择工具？

1. 工具分组：按领域对工具进行分组，每次只暴露相关的工具子集
2. 工具描述优化：确保每个工具的描述简洁明确，避免歧义
3. 动态工具过滤：根据用户意图动态过滤不相关的工具
4. 工具推荐：维护一个工具推荐引擎，根据上下文推荐最相关的工具

7.2 安全最佳实践

1. 最小权限原则：只暴露 LLM 实际需要的工具
2. 用户确认：对敏感操作（写入、删除、发送）要求用户确认
3. 输入验证：在 Client 端对工具参数进行基本验证
4. 结果审计：记录所有工具调用的输入和输出

7.3 性能优化

1. 连接复用：保持与 Server 的长连接，避免频繁创建销毁
2. 并行调用：当多个工具调用相互独立时，使用asyncio.gather并行执行
3. 结果缓存：对幂等的工具调用结果进行缓存
4. 懒连接：仅在首次使用时建立与 Server 的连接

八、总结

MCP Client 是 Agent 与工具世界之间的桥梁。一个优秀的 Client 实现需要：

1. 稳定的连接管理：处理连接建立、维持、断开、重连的完整生命周期
2. 高效的工具发现：动态获取并缓存 Server 暴露的工具列表
3. 安全的工具调用：包含超时控制、错误处理、结果验证
4. 灵活的集成能力：将 MCP 工具无缝接入 LLM 的 Function Calling 流程

通过本文的示例代码，你应该已经掌握了 MCP Client 的核心实现。在下一篇文章中，我们将深入 MCP 最新的 2026-07-28 规范，了解协议的重大变化和迁移策略。

参考资料

1. Anthropic. “MCP Client Development Guide.”modelcontextprotocol.io, https://modelcontextprotocol.io/quickstart/client
2. MCP Python SDK — Client Examples.github.com, https://github.com/modelcontextprotocol/python-sdk
3. Anthropic. “MCP Client Features — Sampling.”modelcontextprotocol.io, https://modelcontextprotocol.io/specification/2025-11-25/client/sampling
4. Anthropic. “MCP Architecture — Roles.”modelcontextprotocol.io, https://modelcontextprotocol.io/specification/2025-11-25/architecture
5. OpenAI. “Function Calling Guide.”platform.openai.com, https://platform.openai.com/docs/guides/function-calling

本系列覆盖AI 大模型基础、Agent 开发、MCP 协议、Skill 开发、RAG、模型微调、部署推理七大方向，从入门到实战的全栈内容持续更新中。

所有文章的 Markdown 源文件、可运行代码、高清配图已整理成完整资料包。

👍 点赞 + ⭐ 关注，评论区扣「1」，挨个发你领取方式 👇