1. 从“协议”到“集成”为什么MCP值得你投入时间如果你最近在关注AI应用开发尤其是围绕大型语言模型LLM构建智能体Agent或工具链那么“MCP”这个词大概率已经在你眼前晃过好几次了。它可能出现在Claude Desktop的配置里出现在Cursor IDE的插件市场或者出现在一些前沿开发者的技术分享中。但当你真正想去了解它时却发现资料零散官方文档偏向协议规范社区案例又多是某个具体工具的配置片段。你可能会困惑这个协议到底解决了什么痛点我为什么要花时间搭建一个自己的MCP Server它和我的Agent项目又该怎么结合这正是我写这篇完整指南的初衷。在过去几个月里我深度参与了几个基于MCP的AI工具链项目从最初对着协议文档挠头到成功自建Server并集成进复杂的Agent工作流踩了不少坑也积累了一套行之有效的方法论。MCP全称Model Context Protocol它本质上是一个标准化的“通信层”。你可以把它想象成AI世界里的USB协议或者蓝牙协议——它定义了一套规则让不同的“设备”在这里是LLM和外部工具/数据源能够用一种彼此都能理解的方式对话和交换数据。它的核心价值在于“解耦”和“标准化”。在没有MCP之前如果你想让你的大模型比如GPT-4、Claude 3去调用一个外部API、查询一个数据库或者读取一个本地文件你需要为每一个工具编写特定的适配代码、设计提示词Prompt来教模型如何使用整个过程繁琐且难以复用。而MCP提供了一套统一的描述语言通过JSON Schema来声明工具Tools和资源Resources以及一个标准的传输层比如Stdio或SSE来传递请求和结果。这意味着工具提供者只需按照MCP协议打包好自己的能力就能被任何兼容MCP的客户端如Claude Desktop、Cursor、你自建的Agent框架直接发现和使用无需为每个客户端做重复的适配工作。所以无论你是想为你的团队内部工具构建AI入口还是想开发一个能联网、能查数据库、能操作文件的智能助手亦或是想理解像Cursor这类现代IDE是如何实现“AI赋能”的深入掌握MCP都是一个极具性价比的投资。接下来我将抛开那些泛泛而谈的概念带你从协议的本质开始一步步走到搭建你自己的Server并最终将其无缝集成到一个功能性的Agent中。我们会用到真实的代码示例、可运行的配置并重点分享那些官方文档里不会写的“踩坑”经验。2. 拆解MCP协议不止是JSON更是设计哲学很多人一听到“协议”就觉得头大认为是一堆枯燥的字段定义。但理解MCP协议的设计哲学远比死记硬背几个JSON结构重要这能帮助你在后续开发和调试中做出正确的设计决策。2.1 核心组件Tools、Resources与PromptsMCP协议的核心是三大抽象工具Tools、资源Resources和提示Prompts。它们共同构成了模型与外部世界交互的上下文。工具Tools这是最常用、最直观的概念。一个Tool就是一个模型可以调用的函数。比如“获取天气”、“发送邮件”、“查询数据库”。在MCP中每个Tool都必须用JSON Schema清晰地定义其输入参数。这不仅仅是技术规范更是一种强制性的“设计文档”。它要求开发者必须事先想清楚这个工具叫什么名字需要哪些参数每个参数是什么类型、是否必填、有什么描述例如一个搜索工具的Schema可能包含query字符串必填和max_results整数可选字段。这种严格的声明式定义使得客户端如LLM能够在运行时动态地、准确地理解如何调用工具而无需硬编码的逻辑。资源Resources这是MCP中一个非常精妙的设计。Resource代表一个可供读取的、具有统一标识符URI的数据单元。它可以是一个本地文件file:///path/to/doc.md一个数据库表db://sales/customers甚至是一个远程API的某个端点api://weather/current。与Tool的“主动调用”不同Resource是“被动提供”的。Server向客户端声明自己有哪些Resources客户端或用户可以选择将哪些Resources的“内容”作为上下文附加给模型。这解决了大模型上下文窗口有限但又需要参考大量背景信息的问题。例如你可以将项目文档、API说明书作为Resources提供给模型模型在回答问题时就能基于这些资料生成更准确的答案。提示PromptsPrompt在MCP中也被对象化了。Server可以预定义一些高质量的提示模板比如“代码审查”、“撰写周报”、“分析数据”。客户端可以列出这些Prompt用户可以选择一个并填入变量如代码片段、周报日期来快速生成高质量的指令。这促进了最佳实践的沉淀和复用。2.2 传输层Stdio与SSE的抉择协议定义了“说什么”传输层则定义“怎么说”。MCP主要支持两种传输方式标准输入输出Stdio和服务器发送事件SSE。选择哪一种取决于你的部署场景。Stdio标准输入输出这是最简单、最常用的模式尤其适用于本地或紧密集成的场景。MCP Server作为一个独立的进程启动客户端如Claude Desktop通过标准输入stdin向它发送JSON-RPC请求并通过标准输出stdout读取响应。它的优点是零网络配置、启动快、依赖少。你在本地调试一个自定义的MCP Server时几乎总是从Stdio模式开始。它的缺点是Server进程的生命周期通常与客户端绑定且难以被多个远程客户端同时访问。SSEServer-Sent Events这是一种基于HTTP的轻量级服务器推送技术。在这种模式下MCP Server作为一个HTTP服务器运行。客户端通过HTTP POST发送请求并通过一个长期的SSE连接接收来自服务器的异步事件流如Tool调用的结果、Resource内容的更新。SSE模式的优点是支持远程访问、一个Server可服务多个客户端并且更符合云原生的部署方式。缺点是设置稍复杂需要处理HTTP服务器、CORS等问题。我的经验之谈对于个人工具或IDE集成优先从Stdio开始简单粗暴。当你需要将能力以服务的形式提供给团队或云端Agent时再考虑迁移到SSE。很多成熟的MCP Server实现比如官方提供的示例都同时支持两种模式通过一个启动参数来切换。2.3 请求-响应循环一个完整的交互示例理论说了这么多我们来看一个缩略的、但完整的JSON-RPC交互流程这能帮你建立直观感受。假设我们有一个“天气查询”Tool。初始化客户端启动通过Stdio连接到Server。双方交换initialize和initialized消息协商协议版本和能力。列出工具客户端发送tools/list请求。Server回复一个列表其中包含我们定义的get_weather工具及其详细的输入JSON Schema。// Server响应 tools/list 的示例片段 { jsonrpc: 2.0, id: 1, result: { tools: [{ name: get_weather, description: 获取指定城市的当前天气, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称例如北京 } }, required: [city] } }] } }调用工具用户向LLM提问“北京天气怎么样”。LLM分析后决定调用get_weather工具并通过客户端发送tools/call请求。{ jsonrpc: 2.0, id: 2, method: tools/call, params: { name: get_weather, arguments: { city: 北京 } } }执行并返回Server收到请求执行真正的天气查询逻辑可能是调用第三方API然后通过tools/call的响应返回结果。{ jsonrpc: 2.0, id: 2, result: { content: [{ type: text, text: 北京当前天气晴气温 22°C北风2级。 }] } }结果交付客户端将结果返回给LLMLLM整合信息后生成最终回答给用户“北京现在是晴天22度比较舒适。”这个流程清晰地展示了MCP如何作为“翻译官”和“邮差”在LLM和外部能力之间建立了一条标准化通道。理解了这一点我们就能动手构建自己的“能力提供方”——MCP Server了。3. 自建MCP Server实战从零到一暴露你的第一个工具现在我们进入实战环节。我将以最流行的Python环境为例带你创建一个最简单的MCP Server它提供一个“计算器”工具和一个“服务器状态”资源。我们将使用官方推荐的mcpSDK它封装了协议细节让我们能专注于业务逻辑。3.1 环境准备与项目初始化首先确保你的Python版本在3.8以上。创建一个新的项目目录并设置虚拟环境是良好的习惯。mkdir my-mcp-server cd my-mcp-server python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来安装核心的MCP开发库。mcp是协议实现的底层库而mcp[cli]则包含了用于测试和运行的命令行工具。pip install mcp[cli]注意网络环境可能导致安装较慢可以使用国内镜像源如pip install mcp[cli] -i https://pypi.tuna.tsinghua.edu.cn/simple。另外mcp库更新较快如果遇到接口变化请参考其GitHub仓库的最新文档。3.2 编写你的第一个ServerCalculator Status创建一个名为server.py的文件我们将从这里开始。import asyncio from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types # 创建MCP Server实例 app Server(my-first-mcp-server) # 1. 定义一个工具Tool加法计算器 app.list_tools() async def handle_list_tools() - list[types.Tool]: return [ types.Tool( nameadd_numbers, description将两个数字相加, inputSchema{ type: object, properties: { a: {type: number, description: 第一个加数}, b: {type: number, description: 第二个加数}, }, required: [a, b], }, ) ] # 处理该工具的调用 app.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any]) - list[types.TextContent]: if name add_numbers: a arguments.get(a, 0) b arguments.get(b, 0) result a b return [types.TextContent(typetext, textf{a} {b} {result})] # 如果收到未知工具名可以抛出错误 raise ValueError(fUnknown tool: {name}) # 2. 定义一个资源Resource服务器状态 app.list_resources() async def handle_list_resources() - list[types.Resource]: return [ types.Resource( urifile:///server/status, name服务器状态, description当前MCP服务器的运行状态信息, mimeTypetext/plain, ) ] # 处理对该资源的读取请求 app.read_resource() async def handle_read_resource(uri: str) - str: if uri file:///server/status: # 这里可以返回更复杂的动态信息比如内存使用率、启动时间等 status_info 服务器状态报告 - 名称: My First MCP Server - 运行中: 是 - 工具数量: 1 (add_numbers) - 资源数量: 1 (本状态文档) - 最后更新: 2023-10-27 10:00:00 return status_info raise ValueError(fUnknown resource: {uri}) # 主异步函数启动Stdio服务器 async def main(): # 配置Stdio服务器参数 server_params StdioServerParameters( commandpython, # 解释器 args[server.py], # 脚本自身当以模块运行时可能需要调整 ) async with mcp.server.stdio.stdio_server(server_params) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize(InitializationOptions(root_urifile:///, capabilitiesapp.get_capabilities())) # 运行应用开始处理请求 await app.run(session, read_stream, write_stream) if __name__ __main__: asyncio.run(main())这段代码做了以下几件事创建了一个名为my-first-mcp-server的Server实例。通过app.list_tools()装饰器声明了一个工具列表其中只有一个工具add_numbers并定义了其输入参数Schema。通过app.call_tool()装饰器处理工具调用请求当收到add_numbers调用时执行加法并返回结果。通过app.list_resources()和app.read_resource()类似地声明并提供了一个只读的资源。main()函数配置了Stdio传输层并启动了服务器事件循环。3.3 本地测试与调试使用MCP CLI如何验证我们的Server是否正常工作官方mcp包附带的CLI工具mcp是的名字一样是一个绝佳的测试客户端。首先我们需要一个配置文件来告诉CLI如何连接我们的Server。创建一个名为mcp-config.json的文件通常位于用户目录下的.config/mcp/但测试时可以在当前目录{ mcpServers: { my-calculator: { command: python, args: [/绝对路径/to/your/project/server.py], env: { PYTHONPATH: /绝对路径/to/your/project } } } }然后在终端运行MCP CLI的检查命令mcp inspect my-calculator如果一切正常你将看到类似下面的输出清晰地列出了你的Server提供的所有Tools和ResourcesServer: my-calculator Tools: - add_numbers: 将两个数字相加 Resources: - file:///server/status (text/plain): 服务器状态你还可以使用mcp call命令直接调用工具进行测试mcp call my-calculator add_numbers --arg a5 --arg b3预期输出Tool call result: 5 3 8踩坑实录在配置command和args时路径问题是最常见的错误来源。特别是当你的脚本有相对路径导入时确保args中的路径是绝对路径或者通过env设置正确的PYTHONPATH。另一个常见问题是虚拟环境确保CLI是在激活了正确虚拟环境的终端中运行的或者command直接指向虚拟环境中的Python解释器如venv/bin/python。通过CLI测试我们确认Server协议层面已完全正确。接下来我们要把它集成到一个真正的“消费者”——AI Agent中。4. 集成MCP到AI Agent以LangGraph为例构建智能工作流拥有一个MCP Server只是第一步让AI Agent能够利用它才是目标。这里我选择以LangGraph框架为例因为它能很好地表达带有循环、分支的复杂Agent工作流并且与MCP的集成路径清晰。我们的目标是构建一个Agent它能理解用户需求自动判断是否需要调用我们的计算器工具并给出答案。4.1 搭建基础的LangGraph Agent首先安装必要的库。我们将使用LangChain的LangGraph实现并需要OpenAI或其他LLM的API。pip install langgraph langchain-openai mcp假设我们已经有了一个可用的OpenAI API Key。接下来我们编写一个基础的、集成了MCP的Agent。关键点在于我们需要一个“MCP Client”来与我们的Server通信并将Server提供的Tools动态地注入到LLM的调用选项中。import asyncio from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool from mcp import ClientSession, StdioServerParameters import mcp.client.stdio # 1. 定义Agent的状态 class AgentState(TypedDict): messages: Annotated[list, 对话消息历史] # 可以添加其他状态如用户查询、中间结果等 # 2. 创建MCP Client并获取工具列表 async def get_mcp_tools(): 连接到我们的MCP Server并获取其工具定义 server_params StdioServerParameters( commandpython, args[/绝对路径/to/your/project/server.py], ) # 注意这里我们使用client的stdio来连接 async with mcp.client.stdio.stdio_client(server_params) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() # 获取Server声明的所有工具 tools_response await session.list_tools() mcp_tools tools_response.tools # 将MCP工具转换为LangChain Tool对象 # 注意这里需要将MCP的调用方式适配到LangChain的Tool格式 # 为了简化我们先手动创建一个实际项目需要写一个通用的适配器 # 假设我们只处理 add_numbers from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class AddInput(BaseModel): a: float Field(description第一个加数) b: float Field(description第二个加数) async def add_numbers(a: float, b: float) - str: # 这里实际上应该通过MCP session去调用为演示我们直接计算 # 真实集成需要复用上面的session进行 tools/call return f{a} {b} {a b} # 创建LangChain Tool lc_tool StructuredTool.from_function( funcadd_numbers, nameadd_numbers, description将两个数字相加, args_schemaAddInput, ) return [lc_tool] # 3. 定义Agent的节点Nodes def call_model(state: AgentState): 调用LLM决定下一步行动回复或调用工具 # 从异步函数同步获取工具在实际应用中需要处理异步同步化 # 此处为示例假设我们已经通过其他方式将tools注入到了llm中 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 绑定工具到LLM llm_with_tools llm.bind_tools(tools) # 假设tools已定义 message llm_with_tools.invoke(state[messages]) return {messages: [message]} def execute_tools(state: AgentState): 执行LLM选择要调用的工具 last_message state[messages][-1] tool_calls last_message.tool_calls if not tool_calls: return {messages: []} # 没有工具调用 tool_messages [] for tc in tool_calls: tool_name tc[name] tool_args tc[args] # 这里根据tool_name分发到具体的工具函数执行 if tool_name add_numbers: result f{tool_args[a]} {tool_args[b]} {tool_args[a] tool_args[b]} else: result fError: Unknown tool {tool_name} # 构造ToolMessage返回给LLM tool_messages.append(ToolMessage(contentresult, tool_call_idtc[id])) return {messages: tool_messages} # 4. 构建图Graph workflow StateGraph(AgentState) workflow.add_node(agent, call_model) # LLM思考节点 workflow.add_node(action, execute_tools) # 执行工具节点 workflow.set_entry_point(agent) # 定义条件边如果最后一条消息有tool_calls就去执行工具否则结束 def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return action return END workflow.add_conditional_edges( agent, should_continue, ) workflow.add_edge(action, agent) # 执行完工具后回到agent继续思考 app workflow.compile() # 5. 运行Agent async def main(): # 在实际启动前先获取MCP工具并绑定到LLM # 注意此示例简化了异步获取工具并同步注入流程真实场景需更严谨处理 tools await get_mcp_tools() llm ChatOpenAI(modelgpt-4o-mini, temperature0).bind_tools(tools) # 更新call_model函数中的llm绑定此处为概念演示需重构代码结构 # 假设我们已重构现在可以运行 initial_state AgentState(messages[HumanMessage(content请帮我计算42加18等于多少)]) # 注意app.invoke是同步的我们的工具获取是异步的这里存在不匹配。 # 解决方案在应用启动前异步获取工具或使用异步版本的invoke。 result app.invoke(initial_state) for msg in result[messages]: print(f{msg.type}: {msg.content}) if __name__ __main__: asyncio.run(main())这段代码勾勒出了集成的核心骨架状态定义使用TypedDict定义Agent的对话状态。MCP工具获取get_mcp_tools函数异步连接到我们的MCP Server获取工具列表并将其转换为LangChain能识别的Tool对象。这是集成的关键桥梁。节点与图call_model节点让LLM根据对话历史和可用工具决定行动execute_tools节点负责执行具体的工具调用。通过条件边我们构建了经典的“思考-行动”循环。运行将用户查询“计算42加18”放入初始状态运行图Agent会自动调用add_numbers工具并给出答案。核心难点与解决方案上述示例最大的简化在于“异步获取工具”与“同步执行图”之间的矛盾。在生产环境中你需要更优雅地处理方案A推荐在Agent应用启动时异步初始化所有MCP Client并获取工具将其缓存起来。确保call_model节点使用的LLM已经绑定了这些工具。方案B使用LangGraph的异步API如ainvoke并重写节点为异步函数在execute_tools节点内直接使用MCP Client会话进行调用。工具适配器需要编写一个通用的MCPToolAdapter类它能将任意的MCP Tool定义动态地转换为LangChain Tool并处理调用转发。这涉及到对MCPargumentsJSON Schema到PydanticBaseModel的映射。4.2 进阶处理复杂工具与资源集成我们的计算器工具很简单。但现实中的工具可能更复杂比如需要一个多步认证的API或者返回结构化数据如图表、JSON。MCP的TextContent和ImageContent等类型可以很好地支持这些。在Server端你可以返回types.TextContent(typetext, textjson.dumps(data))甚至types.ImageContent(typeimage, database64_encoded_image, mimeTypeimage/png)。对于**资源Resources**的集成在Agent场景下通常用于“知识注入”。例如你有一个file:///project/docs/api.md资源。在启动Agent处理特定任务前你可以通过MCP Client的read_resource方法读取该文档内容然后将其作为系统提示词System Prompt或上下文的一部分预先提供给LLM。这样Agent在回答关于API的问题时就有了准确的参考依据。# 伪代码在Agent初始化时注入资源内容 async def enrich_context_with_resources(): async with mcp_client_session() as session: resources await session.list_resources() relevant_uris [r.uri for r in resources if api in r.name] # 筛选资源 contexts [] for uri in relevant_uris: content await session.read_resource(uri) contexts.append(content) return \n\n.join(contexts) system_message f你是一个助手请参考以下项目文档来回答问题 {await enrich_context_with_resources()} # 将system_message放入对话历史开头这种模式将动态的、可管理的外部知识库与Agent的推理能力结合了起来非常强大。5. 生产级考量安全、性能与可观测性当你准备将集成了MCP Server的Agent投入生产环境时以下几个方面的考量至关重要。5.1 安全与权限控制MCP Server可能暴露敏感操作如数据库写、发送邮件或数据如内部文档。必须实施严格的权限控制。Server端校验在每个Tool的执行函数和Resource的读取函数中加入身份验证和授权逻辑。例如可以从初始化请求的InitializationOptions中传递用户令牌Server端进行验证。最小权限原则为不同的客户端如不同部门的Agent配置不同的MCP Server配置文件仅暴露其必需的工具和资源。输入验证与净化尽管有JSON SchemaServer端仍应对输入参数进行严格的业务逻辑验证防止注入攻击等。5.2 性能优化与连接管理连接池对于SSE模式的Server需要考虑客户端连接数。使用合适的异步服务器框架如FastAPI并管理好连接生命周期。工具调用超时与重试在Agent端调用MCP Tool时必须设置超时如10秒并设计重试逻辑对于幂等操作。避免因为一个慢速工具阻塞整个Agent工作流。资源缓存对于不常变化的Resource如静态文档可以在Client端或Server端实现缓存机制避免频繁读取。5.3 日志、监控与调试可观测性是排查线上问题的生命线。结构化日志在Server和Agent端使用结构化日志JSON格式记录每个请求的ID、工具名、参数、执行时间、结果状态和错误信息。这便于使用ELK、Loki等工具进行聚合分析。关键指标监控工具调用频率、平均响应时间、错误率。对于SSE Server监控活跃连接数。调试技巧在开发阶段可以启用MCP协议的详细日志。许多MCP客户端库支持设置环境变量如MCP_LOGdebug来打印所有进出的JSON-RPC消息这对于理解通信过程和无理。5.4 部署模式Sidecar与集中式如何部署你的MCP ServerSidecar模式一对一每个运行Agent的容器或Pod都附带一个专用的MCP Server Sidecar容器。它们通过本地Stdio或localhost SSE通信。优点是隔离性好资源独享缺点是资源消耗倍增管理成本高。集中式服务一对多部署一个或多个强大的MCP Server实例作为集群内的服务。所有Agent通过内网SSE连接到此服务。优点是资源利用率高易于统一升级和管理缺点是引入了网络依赖需要处理服务发现、负载均衡和高可用。我的经验是对于轻量级、个性化的工具如个人文件搜索适合用Sidecar模式通过Stdio与Agent绑定。对于重量级、共享的数据源或服务如公司客户数据库、统一搜索API则适合构建集中式的MCP Server服务。6. 生态与展望不止于Claude和Cursor目前MCP最知名的应用是集成在Claude Desktop和Cursor IDE中。用户只需在配置文件中添加MCP Server地址就能立即在Chat界面或编辑器中使用这些扩展能力。但这只是冰山一角。MCP的潜力在于其协议的中立性。任何兼容MCP的客户端都可以无缝接入任何兼容MCP的Server。这意味着你的自定义Agent框架可以轻松获得海量能力。企业内部可以构建统一的能力中台各个AI应用客服、编码助手、数据分析都通过MCP协议来消费这些能力避免重复建设。开源社区会出现越来越多高质量的、针对特定领域的MCP Server比如tavily-mcp网络搜索、brave-search-mcp搜索、filesystem-mcp文件系统。集成它们就像安装插件一样简单。我个人的体会是MCP正在做的是为AI应用定义“外围设备”的即插即用标准。今天你为Claude写的一个Server明天可能就能被LangGraph、AutoGen、甚至是未来某个新框架直接使用。这种投资的长尾效应非常可观。最后再分享一个实用小技巧当你自建的Server工具越来越多时管理工具列表会成为挑战。可以考虑在Server中实现一个“工具分类”或“标签”机制并在list_tools的返回中通过description或自定义的元数据字段来体现。这样在客户端如Claude Desktop中用户能更好地理解和使用你的工具集。构建和集成MCP Server的过程是一个深入思考如何将AI能力与具体业务逻辑结合的过程。它迫使你清晰地定义接口、考虑边界情况、设计用户交互。这份经历本身就是对于构建可靠AI应用的一次极佳训练。 
长河君 · 2026/8/7 5:08:32 · 阅读 1.2k MCP协议实战指南：从原理到AI Agent集成开发
1. 从“协议”到“集成”为什么MCP值得你投入时间如果你最近在关注AI应用开发尤其是围绕大型语言模型LLM构建智能体Agent或工具链那么“MCP”这个词大概率已经在你眼前晃过好几次了。它可能出现在Claude Desktop的配置里出现在Cursor IDE的插件市场或者出现在一些前沿开发者的技术分享中。但当你真正想去了解它时却发现资料零散官方文档偏向协议规范社区案例又多是某个具体工具的配置片段。你可能会困惑这个协议到底解决了什么痛点我为什么要花时间搭建一个自己的MCP Server它和我的Agent项目又该怎么结合这正是我写这篇完整指南的初衷。在过去几个月里我深度参与了几个基于MCP的AI工具链项目从最初对着协议文档挠头到成功自建Server并集成进复杂的Agent工作流踩了不少坑也积累了一套行之有效的方法论。MCP全称Model Context Protocol它本质上是一个标准化的“通信层”。你可以把它想象成AI世界里的USB协议或者蓝牙协议——它定义了一套规则让不同的“设备”在这里是LLM和外部工具/数据源能够用一种彼此都能理解的方式对话和交换数据。它的核心价值在于“解耦”和“标准化”。在没有MCP之前如果你想让你的大模型比如GPT-4、Claude 3去调用一个外部API、查询一个数据库或者读取一个本地文件你需要为每一个工具编写特定的适配代码、设计提示词Prompt来教模型如何使用整个过程繁琐且难以复用。而MCP提供了一套统一的描述语言通过JSON Schema来声明工具Tools和资源Resources以及一个标准的传输层比如Stdio或SSE来传递请求和结果。这意味着工具提供者只需按照MCP协议打包好自己的能力就能被任何兼容MCP的客户端如Claude Desktop、Cursor、你自建的Agent框架直接发现和使用无需为每个客户端做重复的适配工作。所以无论你是想为你的团队内部工具构建AI入口还是想开发一个能联网、能查数据库、能操作文件的智能助手亦或是想理解像Cursor这类现代IDE是如何实现“AI赋能”的深入掌握MCP都是一个极具性价比的投资。接下来我将抛开那些泛泛而谈的概念带你从协议的本质开始一步步走到搭建你自己的Server并最终将其无缝集成到一个功能性的Agent中。我们会用到真实的代码示例、可运行的配置并重点分享那些官方文档里不会写的“踩坑”经验。2. 拆解MCP协议不止是JSON更是设计哲学很多人一听到“协议”就觉得头大认为是一堆枯燥的字段定义。但理解MCP协议的设计哲学远比死记硬背几个JSON结构重要这能帮助你在后续开发和调试中做出正确的设计决策。2.1 核心组件Tools、Resources与PromptsMCP协议的核心是三大抽象工具Tools、资源Resources和提示Prompts。它们共同构成了模型与外部世界交互的上下文。工具Tools这是最常用、最直观的概念。一个Tool就是一个模型可以调用的函数。比如“获取天气”、“发送邮件”、“查询数据库”。在MCP中每个Tool都必须用JSON Schema清晰地定义其输入参数。这不仅仅是技术规范更是一种强制性的“设计文档”。它要求开发者必须事先想清楚这个工具叫什么名字需要哪些参数每个参数是什么类型、是否必填、有什么描述例如一个搜索工具的Schema可能包含query字符串必填和max_results整数可选字段。这种严格的声明式定义使得客户端如LLM能够在运行时动态地、准确地理解如何调用工具而无需硬编码的逻辑。资源Resources这是MCP中一个非常精妙的设计。Resource代表一个可供读取的、具有统一标识符URI的数据单元。它可以是一个本地文件file:///path/to/doc.md一个数据库表db://sales/customers甚至是一个远程API的某个端点api://weather/current。与Tool的“主动调用”不同Resource是“被动提供”的。Server向客户端声明自己有哪些Resources客户端或用户可以选择将哪些Resources的“内容”作为上下文附加给模型。这解决了大模型上下文窗口有限但又需要参考大量背景信息的问题。例如你可以将项目文档、API说明书作为Resources提供给模型模型在回答问题时就能基于这些资料生成更准确的答案。提示PromptsPrompt在MCP中也被对象化了。Server可以预定义一些高质量的提示模板比如“代码审查”、“撰写周报”、“分析数据”。客户端可以列出这些Prompt用户可以选择一个并填入变量如代码片段、周报日期来快速生成高质量的指令。这促进了最佳实践的沉淀和复用。2.2 传输层Stdio与SSE的抉择协议定义了“说什么”传输层则定义“怎么说”。MCP主要支持两种传输方式标准输入输出Stdio和服务器发送事件SSE。选择哪一种取决于你的部署场景。Stdio标准输入输出这是最简单、最常用的模式尤其适用于本地或紧密集成的场景。MCP Server作为一个独立的进程启动客户端如Claude Desktop通过标准输入stdin向它发送JSON-RPC请求并通过标准输出stdout读取响应。它的优点是零网络配置、启动快、依赖少。你在本地调试一个自定义的MCP Server时几乎总是从Stdio模式开始。它的缺点是Server进程的生命周期通常与客户端绑定且难以被多个远程客户端同时访问。SSEServer-Sent Events这是一种基于HTTP的轻量级服务器推送技术。在这种模式下MCP Server作为一个HTTP服务器运行。客户端通过HTTP POST发送请求并通过一个长期的SSE连接接收来自服务器的异步事件流如Tool调用的结果、Resource内容的更新。SSE模式的优点是支持远程访问、一个Server可服务多个客户端并且更符合云原生的部署方式。缺点是设置稍复杂需要处理HTTP服务器、CORS等问题。我的经验之谈对于个人工具或IDE集成优先从Stdio开始简单粗暴。当你需要将能力以服务的形式提供给团队或云端Agent时再考虑迁移到SSE。很多成熟的MCP Server实现比如官方提供的示例都同时支持两种模式通过一个启动参数来切换。2.3 请求-响应循环一个完整的交互示例理论说了这么多我们来看一个缩略的、但完整的JSON-RPC交互流程这能帮你建立直观感受。假设我们有一个“天气查询”Tool。初始化客户端启动通过Stdio连接到Server。双方交换initialize和initialized消息协商协议版本和能力。列出工具客户端发送tools/list请求。Server回复一个列表其中包含我们定义的get_weather工具及其详细的输入JSON Schema。// Server响应 tools/list 的示例片段 { jsonrpc: 2.0, id: 1, result: { tools: [{ name: get_weather, description: 获取指定城市的当前天气, inputSchema: { type: object, properties: { city: { type: string, description: 城市名称例如北京 } }, required: [city] } }] } }调用工具用户向LLM提问“北京天气怎么样”。LLM分析后决定调用get_weather工具并通过客户端发送tools/call请求。{ jsonrpc: 2.0, id: 2, method: tools/call, params: { name: get_weather, arguments: { city: 北京 } } }执行并返回Server收到请求执行真正的天气查询逻辑可能是调用第三方API然后通过tools/call的响应返回结果。{ jsonrpc: 2.0, id: 2, result: { content: [{ type: text, text: 北京当前天气晴气温 22°C北风2级。 }] } }结果交付客户端将结果返回给LLMLLM整合信息后生成最终回答给用户“北京现在是晴天22度比较舒适。”这个流程清晰地展示了MCP如何作为“翻译官”和“邮差”在LLM和外部能力之间建立了一条标准化通道。理解了这一点我们就能动手构建自己的“能力提供方”——MCP Server了。3. 自建MCP Server实战从零到一暴露你的第一个工具现在我们进入实战环节。我将以最流行的Python环境为例带你创建一个最简单的MCP Server它提供一个“计算器”工具和一个“服务器状态”资源。我们将使用官方推荐的mcpSDK它封装了协议细节让我们能专注于业务逻辑。3.1 环境准备与项目初始化首先确保你的Python版本在3.8以上。创建一个新的项目目录并设置虚拟环境是良好的习惯。mkdir my-mcp-server cd my-mcp-server python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # macOS/Linux: source venv/bin/activate接下来安装核心的MCP开发库。mcp是协议实现的底层库而mcp[cli]则包含了用于测试和运行的命令行工具。pip install mcp[cli]注意网络环境可能导致安装较慢可以使用国内镜像源如pip install mcp[cli] -i https://pypi.tuna.tsinghua.edu.cn/simple。另外mcp库更新较快如果遇到接口变化请参考其GitHub仓库的最新文档。3.2 编写你的第一个ServerCalculator Status创建一个名为server.py的文件我们将从这里开始。import asyncio from typing import Any from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio import mcp.types as types # 创建MCP Server实例 app Server(my-first-mcp-server) # 1. 定义一个工具Tool加法计算器 app.list_tools() async def handle_list_tools() - list[types.Tool]: return [ types.Tool( nameadd_numbers, description将两个数字相加, inputSchema{ type: object, properties: { a: {type: number, description: 第一个加数}, b: {type: number, description: 第二个加数}, }, required: [a, b], }, ) ] # 处理该工具的调用 app.call_tool() async def handle_call_tool(name: str, arguments: dict[str, Any]) - list[types.TextContent]: if name add_numbers: a arguments.get(a, 0) b arguments.get(b, 0) result a b return [types.TextContent(typetext, textf{a} {b} {result})] # 如果收到未知工具名可以抛出错误 raise ValueError(fUnknown tool: {name}) # 2. 定义一个资源Resource服务器状态 app.list_resources() async def handle_list_resources() - list[types.Resource]: return [ types.Resource( urifile:///server/status, name服务器状态, description当前MCP服务器的运行状态信息, mimeTypetext/plain, ) ] # 处理对该资源的读取请求 app.read_resource() async def handle_read_resource(uri: str) - str: if uri file:///server/status: # 这里可以返回更复杂的动态信息比如内存使用率、启动时间等 status_info 服务器状态报告 - 名称: My First MCP Server - 运行中: 是 - 工具数量: 1 (add_numbers) - 资源数量: 1 (本状态文档) - 最后更新: 2023-10-27 10:00:00 return status_info raise ValueError(fUnknown resource: {uri}) # 主异步函数启动Stdio服务器 async def main(): # 配置Stdio服务器参数 server_params StdioServerParameters( commandpython, # 解释器 args[server.py], # 脚本自身当以模块运行时可能需要调整 ) async with mcp.server.stdio.stdio_server(server_params) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize(InitializationOptions(root_urifile:///, capabilitiesapp.get_capabilities())) # 运行应用开始处理请求 await app.run(session, read_stream, write_stream) if __name__ __main__: asyncio.run(main())这段代码做了以下几件事创建了一个名为my-first-mcp-server的Server实例。通过app.list_tools()装饰器声明了一个工具列表其中只有一个工具add_numbers并定义了其输入参数Schema。通过app.call_tool()装饰器处理工具调用请求当收到add_numbers调用时执行加法并返回结果。通过app.list_resources()和app.read_resource()类似地声明并提供了一个只读的资源。main()函数配置了Stdio传输层并启动了服务器事件循环。3.3 本地测试与调试使用MCP CLI如何验证我们的Server是否正常工作官方mcp包附带的CLI工具mcp是的名字一样是一个绝佳的测试客户端。首先我们需要一个配置文件来告诉CLI如何连接我们的Server。创建一个名为mcp-config.json的文件通常位于用户目录下的.config/mcp/但测试时可以在当前目录{ mcpServers: { my-calculator: { command: python, args: [/绝对路径/to/your/project/server.py], env: { PYTHONPATH: /绝对路径/to/your/project } } } }然后在终端运行MCP CLI的检查命令mcp inspect my-calculator如果一切正常你将看到类似下面的输出清晰地列出了你的Server提供的所有Tools和ResourcesServer: my-calculator Tools: - add_numbers: 将两个数字相加 Resources: - file:///server/status (text/plain): 服务器状态你还可以使用mcp call命令直接调用工具进行测试mcp call my-calculator add_numbers --arg a5 --arg b3预期输出Tool call result: 5 3 8踩坑实录在配置command和args时路径问题是最常见的错误来源。特别是当你的脚本有相对路径导入时确保args中的路径是绝对路径或者通过env设置正确的PYTHONPATH。另一个常见问题是虚拟环境确保CLI是在激活了正确虚拟环境的终端中运行的或者command直接指向虚拟环境中的Python解释器如venv/bin/python。通过CLI测试我们确认Server协议层面已完全正确。接下来我们要把它集成到一个真正的“消费者”——AI Agent中。4. 集成MCP到AI Agent以LangGraph为例构建智能工作流拥有一个MCP Server只是第一步让AI Agent能够利用它才是目标。这里我选择以LangGraph框架为例因为它能很好地表达带有循环、分支的复杂Agent工作流并且与MCP的集成路径清晰。我们的目标是构建一个Agent它能理解用户需求自动判断是否需要调用我们的计算器工具并给出答案。4.1 搭建基础的LangGraph Agent首先安装必要的库。我们将使用LangChain的LangGraph实现并需要OpenAI或其他LLM的API。pip install langgraph langchain-openai mcp假设我们已经有了一个可用的OpenAI API Key。接下来我们编写一个基础的、集成了MCP的Agent。关键点在于我们需要一个“MCP Client”来与我们的Server通信并将Server提供的Tools动态地注入到LLM的调用选项中。import asyncio from typing import Annotated, TypedDict from langgraph.graph import StateGraph, END from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, ToolMessage from langchain_core.tools import tool from mcp import ClientSession, StdioServerParameters import mcp.client.stdio # 1. 定义Agent的状态 class AgentState(TypedDict): messages: Annotated[list, 对话消息历史] # 可以添加其他状态如用户查询、中间结果等 # 2. 创建MCP Client并获取工具列表 async def get_mcp_tools(): 连接到我们的MCP Server并获取其工具定义 server_params StdioServerParameters( commandpython, args[/绝对路径/to/your/project/server.py], ) # 注意这里我们使用client的stdio来连接 async with mcp.client.stdio.stdio_client(server_params) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: await session.initialize() # 获取Server声明的所有工具 tools_response await session.list_tools() mcp_tools tools_response.tools # 将MCP工具转换为LangChain Tool对象 # 注意这里需要将MCP的调用方式适配到LangChain的Tool格式 # 为了简化我们先手动创建一个实际项目需要写一个通用的适配器 # 假设我们只处理 add_numbers from langchain_core.tools import StructuredTool from pydantic import BaseModel, Field class AddInput(BaseModel): a: float Field(description第一个加数) b: float Field(description第二个加数) async def add_numbers(a: float, b: float) - str: # 这里实际上应该通过MCP session去调用为演示我们直接计算 # 真实集成需要复用上面的session进行 tools/call return f{a} {b} {a b} # 创建LangChain Tool lc_tool StructuredTool.from_function( funcadd_numbers, nameadd_numbers, description将两个数字相加, args_schemaAddInput, ) return [lc_tool] # 3. 定义Agent的节点Nodes def call_model(state: AgentState): 调用LLM决定下一步行动回复或调用工具 # 从异步函数同步获取工具在实际应用中需要处理异步同步化 # 此处为示例假设我们已经通过其他方式将tools注入到了llm中 llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 绑定工具到LLM llm_with_tools llm.bind_tools(tools) # 假设tools已定义 message llm_with_tools.invoke(state[messages]) return {messages: [message]} def execute_tools(state: AgentState): 执行LLM选择要调用的工具 last_message state[messages][-1] tool_calls last_message.tool_calls if not tool_calls: return {messages: []} # 没有工具调用 tool_messages [] for tc in tool_calls: tool_name tc[name] tool_args tc[args] # 这里根据tool_name分发到具体的工具函数执行 if tool_name add_numbers: result f{tool_args[a]} {tool_args[b]} {tool_args[a] tool_args[b]} else: result fError: Unknown tool {tool_name} # 构造ToolMessage返回给LLM tool_messages.append(ToolMessage(contentresult, tool_call_idtc[id])) return {messages: tool_messages} # 4. 构建图Graph workflow StateGraph(AgentState) workflow.add_node(agent, call_model) # LLM思考节点 workflow.add_node(action, execute_tools) # 执行工具节点 workflow.set_entry_point(agent) # 定义条件边如果最后一条消息有tool_calls就去执行工具否则结束 def should_continue(state: AgentState) - str: last_message state[messages][-1] if last_message.tool_calls: return action return END workflow.add_conditional_edges( agent, should_continue, ) workflow.add_edge(action, agent) # 执行完工具后回到agent继续思考 app workflow.compile() # 5. 运行Agent async def main(): # 在实际启动前先获取MCP工具并绑定到LLM # 注意此示例简化了异步获取工具并同步注入流程真实场景需更严谨处理 tools await get_mcp_tools() llm ChatOpenAI(modelgpt-4o-mini, temperature0).bind_tools(tools) # 更新call_model函数中的llm绑定此处为概念演示需重构代码结构 # 假设我们已重构现在可以运行 initial_state AgentState(messages[HumanMessage(content请帮我计算42加18等于多少)]) # 注意app.invoke是同步的我们的工具获取是异步的这里存在不匹配。 # 解决方案在应用启动前异步获取工具或使用异步版本的invoke。 result app.invoke(initial_state) for msg in result[messages]: print(f{msg.type}: {msg.content}) if __name__ __main__: asyncio.run(main())这段代码勾勒出了集成的核心骨架状态定义使用TypedDict定义Agent的对话状态。MCP工具获取get_mcp_tools函数异步连接到我们的MCP Server获取工具列表并将其转换为LangChain能识别的Tool对象。这是集成的关键桥梁。节点与图call_model节点让LLM根据对话历史和可用工具决定行动execute_tools节点负责执行具体的工具调用。通过条件边我们构建了经典的“思考-行动”循环。运行将用户查询“计算42加18”放入初始状态运行图Agent会自动调用add_numbers工具并给出答案。核心难点与解决方案上述示例最大的简化在于“异步获取工具”与“同步执行图”之间的矛盾。在生产环境中你需要更优雅地处理方案A推荐在Agent应用启动时异步初始化所有MCP Client并获取工具将其缓存起来。确保call_model节点使用的LLM已经绑定了这些工具。方案B使用LangGraph的异步API如ainvoke并重写节点为异步函数在execute_tools节点内直接使用MCP Client会话进行调用。工具适配器需要编写一个通用的MCPToolAdapter类它能将任意的MCP Tool定义动态地转换为LangChain Tool并处理调用转发。这涉及到对MCPargumentsJSON Schema到PydanticBaseModel的映射。4.2 进阶处理复杂工具与资源集成我们的计算器工具很简单。但现实中的工具可能更复杂比如需要一个多步认证的API或者返回结构化数据如图表、JSON。MCP的TextContent和ImageContent等类型可以很好地支持这些。在Server端你可以返回types.TextContent(typetext, textjson.dumps(data))甚至types.ImageContent(typeimage, database64_encoded_image, mimeTypeimage/png)。对于**资源Resources**的集成在Agent场景下通常用于“知识注入”。例如你有一个file:///project/docs/api.md资源。在启动Agent处理特定任务前你可以通过MCP Client的read_resource方法读取该文档内容然后将其作为系统提示词System Prompt或上下文的一部分预先提供给LLM。这样Agent在回答关于API的问题时就有了准确的参考依据。# 伪代码在Agent初始化时注入资源内容 async def enrich_context_with_resources(): async with mcp_client_session() as session: resources await session.list_resources() relevant_uris [r.uri for r in resources if api in r.name] # 筛选资源 contexts [] for uri in relevant_uris: content await session.read_resource(uri) contexts.append(content) return \n\n.join(contexts) system_message f你是一个助手请参考以下项目文档来回答问题 {await enrich_context_with_resources()} # 将system_message放入对话历史开头这种模式将动态的、可管理的外部知识库与Agent的推理能力结合了起来非常强大。5. 生产级考量安全、性能与可观测性当你准备将集成了MCP Server的Agent投入生产环境时以下几个方面的考量至关重要。5.1 安全与权限控制MCP Server可能暴露敏感操作如数据库写、发送邮件或数据如内部文档。必须实施严格的权限控制。Server端校验在每个Tool的执行函数和Resource的读取函数中加入身份验证和授权逻辑。例如可以从初始化请求的InitializationOptions中传递用户令牌Server端进行验证。最小权限原则为不同的客户端如不同部门的Agent配置不同的MCP Server配置文件仅暴露其必需的工具和资源。输入验证与净化尽管有JSON SchemaServer端仍应对输入参数进行严格的业务逻辑验证防止注入攻击等。5.2 性能优化与连接管理连接池对于SSE模式的Server需要考虑客户端连接数。使用合适的异步服务器框架如FastAPI并管理好连接生命周期。工具调用超时与重试在Agent端调用MCP Tool时必须设置超时如10秒并设计重试逻辑对于幂等操作。避免因为一个慢速工具阻塞整个Agent工作流。资源缓存对于不常变化的Resource如静态文档可以在Client端或Server端实现缓存机制避免频繁读取。5.3 日志、监控与调试可观测性是排查线上问题的生命线。结构化日志在Server和Agent端使用结构化日志JSON格式记录每个请求的ID、工具名、参数、执行时间、结果状态和错误信息。这便于使用ELK、Loki等工具进行聚合分析。关键指标监控工具调用频率、平均响应时间、错误率。对于SSE Server监控活跃连接数。调试技巧在开发阶段可以启用MCP协议的详细日志。许多MCP客户端库支持设置环境变量如MCP_LOGdebug来打印所有进出的JSON-RPC消息这对于理解通信过程和无理。5.4 部署模式Sidecar与集中式如何部署你的MCP ServerSidecar模式一对一每个运行Agent的容器或Pod都附带一个专用的MCP Server Sidecar容器。它们通过本地Stdio或localhost SSE通信。优点是隔离性好资源独享缺点是资源消耗倍增管理成本高。集中式服务一对多部署一个或多个强大的MCP Server实例作为集群内的服务。所有Agent通过内网SSE连接到此服务。优点是资源利用率高易于统一升级和管理缺点是引入了网络依赖需要处理服务发现、负载均衡和高可用。我的经验是对于轻量级、个性化的工具如个人文件搜索适合用Sidecar模式通过Stdio与Agent绑定。对于重量级、共享的数据源或服务如公司客户数据库、统一搜索API则适合构建集中式的MCP Server服务。6. 生态与展望不止于Claude和Cursor目前MCP最知名的应用是集成在Claude Desktop和Cursor IDE中。用户只需在配置文件中添加MCP Server地址就能立即在Chat界面或编辑器中使用这些扩展能力。但这只是冰山一角。MCP的潜力在于其协议的中立性。任何兼容MCP的客户端都可以无缝接入任何兼容MCP的Server。这意味着你的自定义Agent框架可以轻松获得海量能力。企业内部可以构建统一的能力中台各个AI应用客服、编码助手、数据分析都通过MCP协议来消费这些能力避免重复建设。开源社区会出现越来越多高质量的、针对特定领域的MCP Server比如tavily-mcp网络搜索、brave-search-mcp搜索、filesystem-mcp文件系统。集成它们就像安装插件一样简单。我个人的体会是MCP正在做的是为AI应用定义“外围设备”的即插即用标准。今天你为Claude写的一个Server明天可能就能被LangGraph、AutoGen、甚至是未来某个新框架直接使用。这种投资的长尾效应非常可观。最后再分享一个实用小技巧当你自建的Server工具越来越多时管理工具列表会成为挑战。可以考虑在Server中实现一个“工具分类”或“标签”机制并在list_tools的返回中通过description或自定义的元数据字段来体现。这样在客户端如Claude Desktop中用户能更好地理解和使用你的工具集。构建和集成MCP Server的过程是一个深入思考如何将AI能力与具体业务逻辑结合的过程。它迫使你清晰地定义接口、考虑边界情况、设计用户交互。这份经历本身就是对于构建可靠AI应用的一次极佳训练。