三亩地 三亩地SAN MU DI · CODE DIARY
ARTICLE DETAIL

日记详情

真实记录编程学习的某一天,欢迎挑你感兴趣的翻一翻。

MCP协议实战:构建AI Agent的万能工具箱,实现工具跨语言跨进程调用

MCP协议实战:构建AI Agent的万能工具箱,实现工具跨语言跨进程调用

1. 项目概述:为什么我们需要一个“万能工具箱”?

如果你最近在折腾AI Agent,尤其是想把本地大模型、各种API和工具串联起来,搞点自动化或者智能应用,那你大概率遇到过这个头疼的问题:工具调用太乱了。用Python写个工具函数,想给Node.js的Agent用?得自己封装一层HTTP接口。工具进程挂了,Agent也跟着崩?还得写一堆守护和重启逻辑。更别提不同框架(LangChain、LlamaIndex、AutoGen)之间的工具生态互不兼容,换个框架就得重写一遍工具适配层。

这感觉就像你有一个顶级厨房(大模型),但每个厨具(工具)的电源插头都不一样,有的还只能用特定品牌的插座(特定编程语言或进程)。每次想做道新菜,光折腾插头接线就耗掉大半精力。MCP(Model Context Protocol)协议,就是为了解决这个“插头不通用”的问题而生的。它本质上是一个标准化的“电源转换器”和“通信协议”,让任何工具,无论用什么语言编写、跑在哪个进程里,都能以一种统一的方式被AI Agent发现、描述和调用。

我最初接触MCP,是在尝试将一个用Go写的内部数据清洗工具集成到基于Python的AI Agent里。传统的做法要么用subprocess调命令行(输出解析是噩梦),要么起个HTTP服务(增加部署复杂度)。直到看到MCP,我才意识到,工具调用可以像插件一样即插即用。这个项目,就是一次深入的MCP协议实战。我们将从零搭建一个MCP Server(工具提供方),并集成到一个AI Agent Client中,彻底打破语言和进程的壁垒。你会发现,一旦工具被“MCP化”,你的Agent就真正拥有了一个按需取用、稳定可靠的“万能工具箱”。

2. MCP协议核心思想与架构拆解

在深入代码之前,我们必须先吃透MCP协议的设计哲学。它不是一个具体的库,而是一个开放标准协议,其核心目标可以用三个词概括:标准化、解耦与流式化

2.1 协议的核心:标准化工具描述与调用

MCP定义了一套基于JSON-RPC 2.0的通信规范。所有通过MCP暴露的工具,都必须遵循统一的描述格式。一个工具(在MCP中称为Tool)主要包含以下几个部分:

  • name: 工具的唯一标识符,如search_web
  • description: 给AI模型看的自然语言描述,说明这个工具是干什么的。这是至关重要的一环,描述的质量直接决定了LLM能否正确理解和使用该工具。
  • inputSchema: 定义调用工具时需要输入的参数,遵循JSON Schema规范。这严格约束了输入格式,避免了歧义。

举个例子,一个获取天气的工具描述可能是这样的:

{ "name": "get_weather", "description": "获取指定城市的当前天气情况。", "inputSchema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,例如:北京、Shanghai" } }, "required": ["city"] } }

这种标准化描述,使得任何兼容MCP的Client(如AI Agent)在连接后,都能通过标准的tools/list请求,获取到所有可用工具的清单和用法,无需任何硬编码。

2.2 进程解耦:Server与Client的分离

这是MCP最具魅力的特点。MCP Server和MCP Client运行在完全独立的进程中,它们之间通过标准输入输出(stdio)、HTTP或SSH进行通信。最常见的开发模式是使用stdio。

这种架构带来了巨大优势:

  1. 语言无关性:Server可以用Python、JavaScript、Go、Rust等任何语言编写,只要它遵循MCP协议输出JSON-RPC消息。Client也同样如此。
  2. 稳定性与隔离性:工具进程(Server)的崩溃、内存泄漏、阻塞,不会直接拖垮主Agent进程(Client)。Client可以监控Server状态,必要时重启它。
  3. 动态性与可扩展性:可以随时启动或停止不同的MCP Server来增删工具集,无需重启主Agent。这为实现“工具热插拔”提供了基础。

2.3 流式(Streaming)与资源(Resources)概念

除了工具调用,MCP还引入了两个高级概念:

  • 资源(Resources):可以理解为只读的数据源。例如,一个“当前登录用户信息”资源,或者一个“数据库schema列表”资源。Client可以订阅(resources/subscribe)这些资源,当资源内容变化时,Server会主动推送更新。这非常适合用来为AI Agent提供动态的上下文信息。
  • 流式(Streaming):主要用于read操作(如读取文件内容)和prompt操作(多步对话)。数据可以分块流式传输,避免一次性加载大内容导致的内存压力和延迟。

理解这些核心思想后,我们就能明白,MCP不仅仅是一个“工具调用协议”,它更是一套用于构建复杂、稳定、可扩展AI应用上下文生态的基石。

3. 实战第一步:构建你的第一个MCP Server

理论说得再多,不如动手写一行代码。我们选择用Python来构建第一个MCP Server,因为它生态丰富,入门简单。我们将使用官方推荐的mcpSDK。

3.1 环境准备与SDK安装

首先,创建一个干净的Python虚拟环境是个好习惯。

python -m venv .venv source .venv/bin/activate # Linux/macOS # 或 .venv\Scripts\activate # Windows

接着,安装MCP的Python开发套件。这里我们安装mcpmcp[cli],后者包含了一些有用的命令行工具。

pip install 'mcp[cli]'

注意:MCP的Python库正在快速发展中,API可能会有变动。建议查看其 GitHub仓库 获取最新文档和示例。

3.2 编写一个简单的工具Server

我们的目标是创建一个提供“计算器”和“天气查询”(模拟)功能的MCP Server。创建文件simple_calculator_server.py

import asyncio from mcp import Server, StdioServerParameters from mcp.types import Tool, TextContent, ImageContent import json # 创建Server实例 server = Server("simple-calculator-server") # 1. 定义工具:加法计算器 @server.list_tools() async def handle_list_tools(): # 返回工具列表 return [ Tool( name="add_numbers", description="将两个数字相加。", inputSchema={ "type": "object", "properties": { "a": {"type": "number", "description": "第一个加数"}, "b": {"type": "number", "description": "第二个加数"} }, "required": ["a", "b"] } ), Tool( name="get_weather", description="模拟获取指定城市的天气。返回一个模拟的天气描述。", inputSchema={ "type": "object", "properties": { "city": {"type": "string", "description": "城市名称"} }, "required": ["city"] } ) ] # 2. 实现工具调用处理函数 @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list: if name == "add_numbers": result = arguments["a"] + arguments["b"] return [TextContent(type="text", text=f"计算结果:{result}")] elif name == "get_weather": city = arguments["city"] # 这里模拟一个天气查询,真实场景会调用API weather_info = f"{city}的模拟天气:晴,温度 22°C,湿度 65%。" return [TextContent(type="text", text=weather_info)] else: raise ValueError(f"未知工具:{name}") # 3. 主函数:启动Stdio Server async def main(): async with server.run_stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ == "__main__": asyncio.run(main())

代码解读与实操要点:

  1. Server实例:这是核心对象,用于注册处理函数。
  2. @server.list_tools():这个装饰器注册的函数,用于响应Client的tools/list请求。它返回一个Tool对象的列表。务必把description写清楚,这是AI理解工具用途的唯一依据。
  3. @server.call_tool():这个装饰器注册的函数,用于响应Client的tools/call请求。参数name是工具名,arguments是客户端传入的参数字典。处理完成后,必须返回一个Content列表,目前最常用的是TextContent
  4. run_stdio_server():这是启动为Stdio模式的关键。它设置了标准输入输出作为通信通道。当Client(如AI Agent框架)启动这个Server作为子进程时,它们将通过管道进行JSON-RPC通信。

3.3 测试你的MCP Server

如何验证Server写对了?我们可以使用MCP CLI工具进行手动测试。首先,确保你的CLI工具已安装(包含在mcp[cli]里)。

创建一个Server描述文件server-config.json,告诉CLI如何启动你的Server。

{ "command": "python", "args": ["/ABSOLUTE/PATH/TO/YOUR/simple_calculator_server.py"], "env": {} }

注意:这里必须使用Python脚本的绝对路径env可以设置环境变量。

然后,在终端使用mcp命令进行测试:

# 查看Server提供的工具列表 mcp tools --config server-config.json # 调用 add_numbers 工具 mcp call --config server-config.json --tool add_numbers --arguments '{"a": 5, "b": 3}' # 调用 get_weather 工具 mcp call --config server-config.json --tool get_weather --arguments '{"city": "北京"}'

如果一切正常,你将看到工具列表和正确的调用结果。这个测试步骤极其重要,它能确保你的Server协议实现是正确的,避免在集成到复杂Agent时出现底层通信问题。

4. 进阶实战:集成真实工具与资源订阅

一个只会做加法和模拟天气的Server显然不够看。让我们来点更实用的,集成一个真实的工具:通过SerpAPI进行网络搜索(你需要一个SerpAPI密钥),并暴露一个“系统状态”资源。

4.1 集成第三方API:搜索工具

安装必要的库:

pip install httpx

创建advanced_search_server.py

import asyncio import os import httpx from mcp import Server, StdioServerParameters from mcp.types import Tool, TextContent, Resource, ListResourcesResult server = Server("advanced-search-server") # 从环境变量读取API密钥 SERPAPI_KEY = os.getenv("SERPAPI_KEY") @server.list_tools() async def handle_list_tools(): return [ Tool( name="search_web", description="使用搜索引擎在互联网上搜索信息。对于需要最新、实时信息的问题非常有用。", inputSchema={ "type": "object", "properties": { "query": {"type": "string", "description": "搜索关键词或问题"}, "num_results": {"type": "number", "description": "返回的结果数量,默认为5", "default": 5} }, "required": ["query"] } ) ] @server.call_tool() async def handle_call_tool(name: str, arguments: dict) -> list: if name == "search_web": query = arguments["query"] num = arguments.get("num_results", 5) if not SERPAPI_KEY: return [TextContent(type="text", text="错误:未设置SERPAPI_KEY环境变量。")] async with httpx.AsyncClient() as client: # 调用SerpAPI(示例,请根据实际API调整) params = { "q": query, "api_key": SERPAPI_KEY, "num": num, "engine": "google" } try: resp = await client.get("https://serpapi.com/search", params=params, timeout=30.0) resp.raise_for_status() data = resp.json() # 简化处理,提取有机搜索结果 results = data.get("organic_results", []) summary = f"关于 '{query}' 的搜索结果(共{len(results)}条):\n\n" for i, r in enumerate(results[:num], 1): summary += f"{i}. [{r.get('title', '无标题')}]({r.get('link', '#')})\n" summary += f" {r.get('snippet', '无摘要')}\n\n" return [TextContent(type="text", text=summary)] except Exception as e: return [TextContent(type="text", text=f"搜索请求失败:{str(e)}")] else: raise ValueError(f"未知工具:{name}") # 4.2 暴露资源(Resources) # 定义一个“系统状态”资源 @server.list_resources() async def handle_list_resources(): # 返回资源列表。每个资源有一个唯一的URI。 return ListResourcesResult(resources=[ Resource( uri="file:///sys/status", name="系统状态概览", description="当前服务器的简单状态信息,如时间、工具数量。", mimeType="text/plain" ) ]) @server.read_resource() async def handle_read_resource(uri: str) -> list: if uri == "file:///sys/status": import datetime status_text = f"""系统状态报告 生成时间:{datetime.datetime.now().isoformat()} 可用工具数:1 (search_web) 资源数:1 运行正常。 """ return [TextContent(type="text", text=status_text)] else: raise ValueError(f"未知资源:{uri}") async def main(): async with server.run_stdio_server() as (read_stream, write_stream): await server.run(read_stream, write_stream) if __name__ == "__main__": asyncio.run(main())

关键点解析:

  1. 环境变量管理:像API密钥这样的敏感信息,务必通过环境变量传入,不要硬编码在代码中。在启动前设置export SERPAPI_KEY=your_key_here
  2. 错误处理:在工具调用中,必须用try...except包裹可能失败的第三方API调用,并返回友好的错误信息给Client,而不是让整个Server崩溃。
  3. 资源定义@server.list_resources()@server.read_resource()分别用于列出资源和读取资源内容。资源URI可以自定义格式,通常类似file://http://这样的协议风格,便于区分。
  4. 资源与工具的区别:资源是被动读取的,提供静态或动态数据;工具是主动调用的,执行一个动作并返回结果。Agent可以根据需求选择订阅资源获取背景信息,或调用工具执行具体操作。

4.3 在Client端订阅资源

资源的价值在于可以被Client“订阅”。当Server端资源内容变化时,可以主动通知Client。虽然我们上面的例子是静态资源,但你可以想象一个“股票价格”资源,当价格变动时主动推送更新。在Client端(如一些高级的AI Agent框架),你可以配置订阅这些资源URI,使其内容自动成为LLM上下文的一部分,让Agent始终掌握最新动态信息。

5. 将MCP Server集成到AI Agent Client

Server准备好了,现在需要让AI Agent能用上它。这里我们以Claude Desktop(一个集成了Claude模型的桌面应用,原生支持MCP)和Cursor(一个AI驱动的代码编辑器)为例,演示如何集成。

5.1 配置Claude Desktop使用自定义MCP Server

Claude Desktop允许通过配置文件添加自定义MCP Server。

  1. 找到配置文件位置

    • macOS:~/Library/Application Support/Claude/claude_desktop_config.json
    • Windows:%APPDATA%\Claude\claude_desktop_config.json
  2. 编辑配置文件:如果文件不存在就创建它。添加mcpServers配置项。

    { "mcpServers": { "my-calculator": { "command": "python", "args": ["/ABSOLUTE/PATH/TO/YOUR/simple_calculator_server.py"] }, "my-web-searcher": { "command": "python", "args": ["/ABSOLUTE/PATH/TO/YOUR/advanced_search_server.py"], "env": { "SERPAPI_KEY": "your_actual_serpapi_key_here" } } } }

    重要提示args中的路径必须是绝对路径env字段用于设置Server进程的环境变量。

  3. 重启Claude Desktop:保存配置文件并完全重启Claude Desktop应用。

  4. 验证集成:重启后,在Claude的聊天界面,你应该能看到一个“工具”图标(可能是个扳手)。点击它,如果配置成功,你会看到my-calculatormy-web-searcher下的工具列表(如add_numbers,search_web)。现在,你可以直接对Claude说:“请用add_numbers工具计算一下123加456”,或者“搜索一下最新的MCP协议动态”,Claude就会自动调用你编写的工具并返回结果。

5.2 在Cursor中配置MCP Server

Cursor编辑器同样支持MCP。配置方式类似,通常在其设置(Settings)中寻找“MCP Servers”或“Advanced”相关选项,添加类似的命令配置。具体路径可能随版本更新而变化,请参考Cursor官方文档。

5.3 编程式集成:在自定义Python Agent中使用

如果你想在自己的Python AI Agent项目(比如使用LangChain、LlamaIndex)中集成MCP Server,可以使用mcp库的Client功能。下面是一个极简示例:

import asyncio from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client async def main(): # 1. 定义如何启动Server进程(与Claude配置类似) server_params = StdioServerParameters( command="python", args=["/path/to/your/advanced_search_server.py"], env={"SERPAPI_KEY": "your_key"} ) # 2. 创建客户端会话并连接 async with stdio_client(server_params) as (read, write): async with ClientSession(read, write) as session: await session.initialize() # 初始化连接 # 3. 列出所有可用工具 tools_result = await session.list_tools() print("可用工具:", [t.name for t in tools_result.tools]) # 4. 调用工具 call_result = await session.call_tool( name="search_web", arguments={"query": "MCP protocol latest news", "num_results": 3} ) for content in call_result.content: if content.type == "text": print("搜索结果:", content.text) # 5. (可选)列出并读取资源 resources_result = await session.list_resources() for resource in resources_result.resources: print("发现资源:", resource.uri) # 可以进一步 session.read_resource(resource.uri) if __name__ == "__main__": asyncio.run(main())

通过这种方式,你可以将任意MCP Server无缝嵌入到你的自动化脚本或智能体应用中,实现强大的工具扩展能力。

6. 性能优化、调试与常见问题排查

在实际生产环境中使用MCP,你会遇到一些挑战。以下是我踩过坑后总结的经验。

6.1 性能考量与优化建议

  1. Server启动开销:每个MCP Server都是一个独立进程。频繁启动销毁开销很大。对于需要长期使用的工具集,应该设计为长生命周期的Server,由Agent Client在启动时连接,而不是每次调用都新建。
  2. 工具调用延迟:跨进程通信(尤其是stdio)会引入毫秒级延迟。对于延迟极度敏感的工具(如简单计算),可以考虑将其实现为Client内的本地函数。MCP更适合用于I/O密集型(网络请求、数据库查询)或计算密集型(需要隔离)的工具。
  3. 流式传输:对于可能返回大量数据的工具(如读取长文档),务必在Server端实现流式响应(使用@server.call_tool(streaming=True)),并在Client端流式读取。这可以显著提升用户体验,避免长时间等待。
  4. 连接管理:实现Client端的连接池和健康检查。定期向Server发送心跳或测试请求,确保连接可用,并在Server无响应时优雅地重连或报警。

6.2 调试技巧与工具

  1. 使用MCP Inspector:这是一个官方的图形化调试工具。你可以运行mcp devtools来启动它,然后加载你的Server配置文件。它可以直观地展示Server提供的工具和资源,并允许你手动调用工具、查看原始JSON-RPC请求和响应,是调试协议问题的利器。
  2. 启用日志:在Server和Client代码中增加详细日志,记录收到的请求、发出的响应以及错误信息。Python的logging模块是好朋友。
    import logging logging.basicConfig(level=logging.DEBUG)
  3. Stdio调试:在开发时,可以暂时修改Server,将通信数据打印到标准错误输出(sys.stderr),以便观察原始数据流。
  4. 超时设置:务必在Client端为工具调用设置合理的超时(如30秒),防止因某个工具挂起而导致整个Agent卡死。

6.3 常见问题与解决方案速查表

问题现象可能原因排查步骤与解决方案
Claude/Cursor中看不到工具1. 配置文件路径错误。
2. Server启动失败。
3. Server未遵循MCP协议。
1. 使用mcp tools --config命令测试Server是否正常。
2. 检查配置文件JSON格式和绝对路径
3. 查看应用日志或系统控制台(如终端)是否有Server报错信息。
工具调用失败,返回“未知工具”1.@server.call_tool()装饰的函数未正确定义或名称不匹配。
2. 工具名拼写错误。
1. 确保handle_call_tool函数能正确匹配name参数。
2. 使用mcp call命令进行手动调用测试,确认Server本身无误。
工具调用超时或无响应1. Server端工具函数执行阻塞或死循环。
2. 网络请求(如调用API)超时。
3. Client-Server进程通信中断。
1. 在Server工具函数内增加超时控制(asyncio.wait_for)。
2. 优化工具逻辑,避免长时间同步操作。
3. 检查Client端的超时设置是否合理。
返回结果乱码或格式错误1. 返回的Content对象格式不符合MCP协议。
2. 文本中包含控制字符或非法JSON。
1. 确保返回的是List[TextContent]等标准类型。
2. 对返回的文本进行必要的清理和转义。
Server进程意外退出1. Server代码中存在未捕获的异常。
2. 环境依赖缺失。
1. 用try...except包裹所有工具和资源处理逻辑。
2. 确保Server运行环境已安装所有依赖包。在配置中可指定完整Python路径。

7. 生态展望与项目进阶方向

MCP协议之所以被称为“万能工具箱”的基石,是因为它背后正在形成一个蓬勃发展的生态。

现有的MCP Server生态:社区已经创建了大量开箱即用的MCP Server,极大丰富了AI Agent的能力边界。例如:

  • 文件系统操作:读写本地文件、列出目录。
  • 数据库连接:查询SQLite、PostgreSQL、MySQL等数据库。
  • 版本控制:与Git仓库交互,执行commit、diff等操作。
  • 云服务:操作AWS S3、Google Cloud Storage等。
  • 专业工具:如Figma(设计)、Brave Search(搜索)、Playwright(浏览器自动化)等都有对应的MCP Server。

你的项目可以如何进阶?

  1. 封装内部工具:将你团队内部常用的脚本、数据处理器、审批接口等全部封装成MCP Server。这样,无论是通过Claude、Cursor,还是你们自研的Agent平台,都能以统一、安全的方式调用这些能力。
  2. 构建工具市场/网关:设计一个中心化的MCP Server管理网关。Agent只需连接这个网关,网关背后动态管理着数十个不同的工具Server,实现负载均衡、权限控制、调用审计和缓存。
  3. 实现动态工具组合:基于MCP,可以开发一个“元Agent”,它的核心能力是分析用户需求,然后动态选择、组合并调用多个底层MCP Server提供的工具来完成任务链。这真正实现了“工具箱”的智能调度。
  4. 与本地大模型深度结合:将MCP Server与Ollama、LM Studio等本地大模型管理工具结合。让完全离线运行的本地大模型,也能拥有联网搜索、操作文件、查询数据库等强大能力,打造真正私密、强大的个人AI助手。

MCP协议解耦的不仅是进程和语言,更是AI能力与具体实现的绑定。它让AI Agent的“身体”(执行能力)可以独立于“大脑”(推理模型)进行进化和发展。当你熟练掌握了MCP的实战,你就为你的AI项目插上了无限扩展的翅膀。

← 返回列表