1. 项目概述为什么我们需要亲手写一个 MCP Server如果你最近在折腾 LangChain 或者 AI Agent大概率已经不止一次看到MCP这个词了。它可能出现在某个开源项目的 README 里或者在 Cursor 这类智能 IDE 的配置文件中一闪而过。MCP全称Model Context Protocol直译过来是“模型上下文协议”。这个名字听起来有点抽象但它的核心目标非常直接为 AI 模型尤其是大语言模型提供一个标准化的方式来发现、连接和使用外部工具与数据源。你可以把它想象成 AI 世界的“USB 协议”。在 USB 出现之前每个外设打印机、鼠标、U盘都需要自己的驱动和连接方式混乱且低效。MCP 想做的是同样的事情——为 AI Agent 定义一套统一的“插口”和“通信规范”让任何符合 MCP 标准的工具Server都能被任何支持 MCP 的 AI 框架Client即插即用。那么为什么我们要从零开始手写一个本地的 MCP Server并把它接入 LangChain Agent 呢原因有三第一打破“工具墙”。目前很多 AI 应用框架都有自己的工具生态比如 LangChain 的 ToolsLlamaIndex 的 Tools。但这些工具往往不能直接互通。通过 MCP我们可以创建一个一次编写、多处运行的工具。今天接入 LangChain明天或许就能无缝接入 Cursor 或者 Claude Desktop极大地提升了开发效率和工具的可复用性。第二深入理解 Agent 的工作机制。仅仅调用initialize_agent函数是远远不够的。通过亲手实现一个 MCP Server你会被迫思考一个工具应该如何向 AI 描述自己AI 调用工具时数据是如何流转的错误该如何处理这个过程会让你对 AI Agent 的“思考-行动-观察”循环有刻骨铭心的理解。第三解锁高度定制化的本地能力。网络上有很多公开的 MCP Server比如查天气的、搜索网页的。但最具价值的工具往往是和你本地环境、私有数据或特定业务逻辑深度绑定的。比如一个能查询你公司内部数据库的 Server一个能控制你智能家居设备的 Server或者一个能读取你特定格式日志文件的 Server。自己写才能完全掌控。本文面向的是已经对 LangChain 有基本了解希望深入 Agent 和工具层构建更强大、更灵活 AI 应用的开发者。我们将从一个最简单的“回声”服务器开始逐步构建一个能处理本地文件查询的实用 Server并最终让它在一个 LangChain Agent 中活起来。你会发现剥离了复杂的概念后其核心不过是基于标准协议的 HTTP 或 STDIO 通信。2. MCP 核心概念与协议基础拆解在动手写代码之前我们必须先搞清楚 MCP 这套协议到底规定了什么。它不是魔法只是一套设计良好的 JSON-RPC 通信约定。理解这一点后续的所有实现都会变得顺理成章。2.1 MCP 的架构角色Server, Client 与 TransportMCP 协议中主要有三个角色MCP Server工具提供方这是我们本文要构建的对象。它本质上是一个能力的提供者。它向外界宣告“我这里有这些工具Tools可用并且我这里有这些数据资源Resources可以读取。” 例如一个“数据库查询 Server”会提供一个“执行 SQL”的工具一个“文件系统 Server”会提供“读取文件”的资源。MCP Client工具使用方这是调用 Server 能力的实体。在我们的场景里LangChain Agent 就是 Client。Client 负责发现 Server 提供了哪些能力和资源并在需要的时候调用它们。更广义的 Client 可以是 Cursor IDE、Claude Desktop 等任何集成了 MCP 客户端的应用。Transport传输层这是连接 Server 和 Client 的桥梁。MCP 协议本身不关心网络细节它定义了消息的格式而 Transport 定义了消息的传递方式。最常见的两种方式是stdio标准输入输出Server 和 Client 作为父子进程运行通过管道通信。这种方式简单、安全适合本地集成。我们本文将主要采用这种方式。HTTP/SSEHTTP 服务器发送事件Server 作为一个 HTTP 服务运行Client 通过 HTTP 请求与之通信。这种方式更适合远程或跨网络调用。协议的核心交互可以简化为Client 启动连接到 Server 的 Transport。随后Client 发送initialize请求Server 回复其提供的“工具列表”和“资源列表”。之后当 Agent作为 Client 的一部分决定使用某个工具时Client 就会向 Server 发送一个tools/call请求Server 执行并返回结果。2.2 核心协议消息剖析MCP 的消息基于 JSON-RPC 2.0。我们不需要实现完整的 JSON-RPC但需要理解几个关键的消息类型。所有消息都有一个jsonrpc: “2.0”字段和一个id字段用于匹配请求和响应。初始化阶段Client - Server:initialize请求{ “jsonrpc”: “2.0”, “id”: 1, “method”: “initialize”, “params”: { “protocolVersion”: “2024-11-05” “capabilities”: { /_ Client 支持的能力 _/ }, “clientInfo”: { “name”: “langchain-mcp-client” } } }Client 告诉 Server 它使用的协议版本和自身信息。Server - Client:initialize响应{ “jsonrpc”: “2.0”, “id”: 1, “result”: { “protocolVersion”: “2024-11-05” “capabilities”: { /_ Server 支持的能力 _/ }, “serverInfo”: { “name”: “my-local-file-server” } } }Server 确认协议版本并返回自身信息。Server - Client:notifications/initialized通知初始化成功后Server 会主动发送一个通知表示自己准备好了。Client - Server:tools/list请求Client 询问 Server 有哪些工具可用。Server - Client:tools/list响应{ “jsonrpc”: “2.0”, “id”: 2, “result”: { “tools”: [ { “name”: “get_current_time” “description”: “获取当前的系统时间” “inputSchema”: { “type”: “object” “properties”: { “format”: { “type”: “string” “description”: “时间格式例如 ‘YYYY-MM-DD HH:mm:ss’” “default”: “%Y-%m-%d %H:%M:%S” } } } } ] } }这是重中之重。Server 返回一个工具列表。每个工具都必须有唯一的name清晰的description这直接决定了 AI 是否会以及如何调用它以及一个inputSchema来定义调用参数。这个 Schema 遵循 JSON Schema 标准AI 模型会根据它来构造调用参数。工具调用阶段Client - Server:tools/call请求{ “jsonrpc”: “2.0”, “id”: 3, “method”: “tools/call” “params”: { “name”: “get_current_time” “arguments”: { “format”: “%H:%M:%S” } } }Client 要求 Server 执行get_current_time工具并传入了参数。Server - Client:tools/call响应{ “jsonrpc”: “2.0”, “id”: 3, “result”: { “content”: [ { “type”: “text” “text”: “当前时间是14:30:25” } ] } }Server 执行工具并将结果放在content字段中返回。结果可以是文本(text)也可以是图像(image)等类型。资源Resources概念除了主动调用的工具ToolsMCP 还有资源Resources的概念。资源更像是一种被动的、可被读取的数据源。Client 可以通过resources/list和resources/read来获取资源列表并读取其内容。例如一个“日志文件 Server”可以将每个日志文件定义为一个资源AI 可以直接请求读取file:///var/log/app.log这个资源的内容而无需通过工具调用。这在提供只读数据时非常有用。注意对于初次实现我建议先聚焦于Tools。这是最常用、也与 LangChain Tool 概念最对齐的部分。Resources 可以在后续需要时再添加。2.3 为什么选择 stdio 作为首个 Transport在开发调试阶段stdio 传输层是首选原因如下依赖极简无需处理网络端口、CORS、认证等复杂问题。只需要启动一个子进程然后读写它的 stdin/stdout。易于调试你可以直接运行你的 Server 脚本然后在终端里手动输入 JSON-RPC 消息模仿 Client观察它的输出。或者你可以将所有的输入输出日志打印到控制台一目了然。LangChain 原生支持LangChain 的 MCP 集成库对 stdio 有很好的支持配置起来非常方便。安全性由于是本地进程间通信避免了网络暴露的风险。理解了这些基础我们就有了清晰的蓝图我们要写一个能解析 JSON-RPC 消息、维护工具列表、并能根据请求执行相应 Python 函数的程序并通过 stdin/stdout 与外界对话。3. 手把手构建一个本地文件查询 MCP Server现在让我们进入实战环节。我们将构建一个名为local_file_mcp_server的服务器它提供一个核心工具read_file允许 AI Agent 读取指定路径的文本文件内容。这是一个极其实用且基础的能力。3.1 项目环境搭建与依赖安装首先创建一个新的项目目录并初始化虚拟环境这是保持环境清洁的好习惯。mkdir local-mcp-server cd local-mcp-server python -m venv venv # 在 Windows 上使用 venv\Scripts\activate source venv/bin/activate接下来安装核心依赖。我们不需要从头实现 JSON-RPC 和 MCP 协议社区已经有优秀的底层库。pip install “mcp[cli]” langchain langchain-communitymcp这是 MCP 协议的官方 Python SDK。它提供了构建 Server 和 Client 所需的所有底层类、类型和工具。[cli]额外安装了一些命令行工具便于调试。langchainlangchain-community用于后续构建 LangChain Agent。实操心得在开发过程中强烈建议同时安装mcp[cli]中的mcp命令行工具。它包含一个mcp dev命令可以像“客户端模拟器”一样测试你的 Server无需等待 LangChain 部分完成能极大提升开发效率。3.2 实现核心 Server 逻辑创建一个名为server.py的文件。我们将从最简单的结构开始逐步完善。第一步导入与基础框架import asyncio import json import sys from typing import Any, List from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import Tool import pydantic # 初始化 MCP Server mcp_server Server(“local-file-server”)我们导入了必要的模块。mcp.server.Server是核心类它封装了协议处理、工具注册等繁琐工作。第二步定义工具及其输入参数模型使用 Pydantic 来定义工具的参数模型这能自动生成符合 JSON Schema 的输入定义并完成数据验证。class ReadFileInput(pydantic.BaseModel): “”“读取文件的输入参数”“” file_path: str pydantic.Field(… description“需要读取的文件的绝对路径或相对路径”) max_lines: int pydantic.Field(100 description“最多读取的行数防止文件过大” ge1 le1000) # 实现工具对应的函数 async def read_file(file_path: str max_lines: int) - str: “”“读取文件内容的实际函数”“” try: with open(file_path ‘r’ encoding‘utf-8’) as f: lines [] for i line in enumerate(f): if i max_lines: lines.append(f“\n…已截断最多显示{max_lines}行) break lines.append(line.rstrip(‘\n’)) content ‘\n’.join(lines) return f“文件 {file_path} 的内容共显示 {len(lines)} 行\n\n{content}” except FileNotFoundError: return f“错误找不到文件 ‘{file_path}’请检查路径是否正确。” except IsADirectoryError: return f“错误’{file_path}’ 是一个目录而非文件。” except PermissionError: return f“错误没有权限读取文件 ‘{file_path}’。” except UnicodeDecodeError: return f“错误无法以 UTF-8 编码解码文件 ‘{file_path}’它可能是一个二进制文件。” except Exception as e: return f“读取文件时发生未知错误{type(e).__name__}: {e}”这里有几个关键点ReadFileInput模型定义了工具需要的参数file_path必填和max_lines可选有默认值和范围限制。description字段非常重要它会被 AI 模型看到用于理解参数含义。read_file函数是工具的实际执行体。它包含了完整的错误处理这是生产级代码的必备项。AI 并不完美它可能会给出错误的路径良好的错误反馈能帮助 AI 进行自我纠正。第三步将工具注册到 MCP Server# 使用装饰器注册工具 mcp_server.tool() async def read_file_tool(input: ReadFileInput) - str: “”“提供给 AI 的‘读取文件’工具。”“” # 这里直接调用实际的函数 return await read_file(input.file_path input.max_lines) # 如果需要可以注册更多工具 mcp_server.tool() async def get_working_directory() - str: “”“获取当前服务器的工作目录。”“” import os return f“当前工作目录是{os.getcwd()}”mcp_server.tool()装饰器是魔法发生的地方。它会自动收集被装饰函数的函数名read_file_tool作为工具名可在装饰器参数中重写。函数的 Docstring 作为工具描述。输入参数的类型注解ReadFileInput来自动生成inputSchema。返回类型信息。第四步实现 stdio 通信的主循环这是 Server 的“耳朵”和“嘴巴”负责与外部 Client 通信。async def main(): # 配置 stdio 传输使用 sys.stdin 和 sys.stdout 作为通信管道 server_params StdioServerParameters( commandsys.executable # 当前 Python 解释器 args[__file__] # 再次运行本脚本用于子进程此处简化实际有更优方式 # 注意对于自包含的 Server通常我们不会在这里启动子进程。 # 更常见的模式是直接使用 mcp.run 或处理 stdio 流。 ) # 为了清晰我们使用 mcp 库提供的更高级运行函数 # 它会处理所有协议握手、消息路由等底层细节 async with mcp_server.run_stdio_server() as (read_stream write_stream): print(“MCP Server 已启动正在通过 stdio 等待连接…” filesys.stderr) # 这里服务器开始运行并阻塞直到连接关闭 await mcp_server._run_forever(read_stream write_stream) # 注意这是一个内部API示例实际请参考官方示例。 if __name__ “__main__”: asyncio.run(main())上面的main函数是一个概念性示例。实际上mcp库提供了更简洁的启动方式。一个更标准、更简单的server.py结尾如下# 更简单直接的启动方式如果库支持 from mcp.server import stdio async def main(): async with stdio.stdio_server() as (read_stream write_stream): await mcp_server.run(read_stream write_stream) if __name__ “__main__”: asyncio.run(main())重要提示mcp库的 API 可能随着版本更新而变化。最可靠的方法是查阅其官方文档或 GitHub 仓库中的示例。核心在于理解我们需要创建一个 Server 对象注册工具然后将其与一个传输层这里是 stdio绑定并运行。3.3 使用 mcp CLI 进行本地测试在接入 LangChain 之前先用官方 CLI 工具测试 Server 是否工作正常。这是排查问题的黄金阶段。首先确保你的server.py脚本可以直接运行并且通过mcp_server.tool()装饰器注册了工具。然后在项目目录下创建一个mcp.json配置文件这是 Cursor、Claude Desktop 等客户端通用的配置方式{ “mcpServers”: { “local-file-server”: { “command”: “python” “args”: [“/绝对路径/到/你的/local-mcp-server/server.py”], “env”: { “PYTHONPATH”: “/绝对路径/到/你的/local-mcp-server” } } } }或者对于开发测试更简单的方法是使用mcp自带的 CLI# 假设你的 server.py 在当前目录 mcp dev server.py运行此命令后CLI 会启动你的 Server 并进入一个交互式界面。它会模拟 Client 的行为自动完成初始化并列出所有工具。你可以在界面中看到类似这样的输出Available tools: - read_file_tool: 提供给 AI 的‘读取文件’工具。 - get_working_directory: 获取当前服务器的工作目录。然后你可以尝试调用工具call read_file_tool {“file_path”: “./server.py” “max_lines”: 5}如果一切正常你将看到 Server 返回的文件内容。这个步骤至关重要它能验证你的 Server 协议实现是否正确工具描述是否清晰功能是否正常。永远不要在 Server 未通过独立测试时就接入复杂的 Agent 环境。4. 将自定义 MCP Server 接入 LangChain AgentServer 测试通过后我们就可以让它为 LangChain Agent 所用了。目标是创建一个能使用我们read_file_tool的 Agent。4.1 在 LangChain 中配置 MCP ClientLangChain-Community 库提供了McpToolkit和create_mcp_toolkit等类来方便地集成 MCP。我们需要创建一个 Client 来连接我们的 Server并将其工具转化为 LangChain 的 Tool 对象。创建一个新的脚本agent_demo.pyimport asyncio from langchain.agents import AgentExecutor create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI # 或其他你用的模型 from langchain_community.tools.mcp import create_mcp_toolkit from langchain_community.agent_toolkits.mcp import McpToolkit async def main(): # 1. 初始化 LLM # 请确保已设置 OPENAI_API_KEY 环境变量 llm ChatOpenAI(model“gpt-4o-mini” temperature0) # 2. 创建 MCP 工具包 # 这里需要指定如何连接到我们的 Server。 # 对于 stdio 传输我们需要提供 Server 的启动命令。 server_parameters { “command”: “python” “args”: [“server.py”] # 假设 server.py 在同一目录 # “env”: {…} 可以传递环境变量 } # 使用 McpToolkit 来管理连接和工具发现 toolkit await McpToolkit.from_server_parameters(server_parameters) # 或者使用更底层的 create_mcp_toolkit # tools await create_mcp_toolkit(server_parameters).get_tools() # 获取所有 MCP 工具转化为 LangChain Tool 列表 tools toolkit.get_tools() print(f“从 MCP Server 加载了 {len(tools)} 个工具”) for tool in tools: print(f” - {tool.name}: {tool.description}”) # 3. 构建 Agent Prompt prompt ChatPromptTemplate.from_messages([ (“system” “你是一个有帮助的助手可以读取本地文件。请根据用户问题谨慎地使用工具。在提供文件内容时注意不要泄露敏感信息。”), (“placeholder” “{chat_history}”), (“human” “{input}”), (“placeholder” “{agent_scratchpad}”), ]) # 4. 创建 Agent 和 Executor agent create_tool_calling_agent(llm tools prompt) agent_executor AgentExecutor(agentagent toolstools verboseTrue handle_parsing_errorsTrue) # 5. 运行测试 print(“\n Agent 测试开始 \n”) result await agent_executor.ainvoke({ “input”: “请帮我读取当前目录下的 ‘server.py’ 文件的前 10 行内容是什么” }) print(f“\n 最终回答 \n{result[‘output’]}”) # 测试一个需要错误处理的场景 print(“\n 测试错误处理 \n”) result2 await agent_executor.ainvoke({ “input”: “读取一个不存在的文件 ‘nonexistent.txt’” }) print(f“\n 最终回答 \n{result2[‘output’]}”) if __name__ “__main__”: asyncio.run(main())代码解析与注意事项连接配置server_parameters字典是关键它定义了如何启动你的 MCP Server 子进程。command和args必须与你在独立测试时使用的命令一致。确保路径正确。异步处理MCP 通信和 LangChain 的某些工具调用是异步的因此我们使用async/await和asyncio.run。McpToolkit.from_server_parameters也是一个异步方法。工具发现toolkit.get_tools()会触发与 Server 的初始化握手并获取到我们在 Server 中注册的所有工具列表自动包装成 LangChain 的Tool对象。Verbose 模式将AgentExecutor的verbose设为True可以在控制台看到 Agent 的完整思考链ReAct这对于调试 Agent 的行为逻辑至关重要。错误处理handle_parsing_errorsTrue可以防止因为工具调用参数解析失败而导致整个 Agent 崩溃让 Agent 有机会重试或向用户澄清。4.2 运行与效果验证运行agent_demo.py。如果一切配置正确你将看到类似以下的输出从 MCP Server 加载了 2 个工具 - read_file_tool: 提供给 AI 的‘读取文件’工具。 - get_working_directory: 获取当前服务器的工作目录。 Agent 测试开始 进入新的 AgentExecutor 链… 思考用户想查看 server.py 文件的前10行。我有一个读取文件的工具。 行动调用工具 read_file_tool 行动输入{“file_path”: “server.py” “max_lines”: 10} 观察文件 server.py 的内容共显示 10 行 这里显示文件内容… 思考我已经获取了文件内容可以直接回复用户。 行动最终答案 文件 server.py 的前10行内容如下 内容展示… 链结束。 最终回答 文件 server.py 的前10行内容如下 内容展示…你会观察到Agent 自动选择了正确的工具并生成了符合inputSchema的 JSON 参数。当测试不存在的文件时Agent 也应该能接收到 Server 返回的错误信息并可能将其转述给用户。至此你已经成功构建了一个闭环手写的 MCP Server - 标准的 MCP 协议 - LangChain MCP Client - LangChain Agent。你的 Agent 现在拥有了读取本地文件的能力。5. 高级话题性能优化、安全与生产化考量一个能跑通的 Demo 和一个健壮的生产级组件之间还有距离。当你打算真正使用这个 MCP Server 时以下几点必须考虑。5.1 Server 实现优化异步与性能我们的工具函数read_file使用了async def但文件 I/O 本身是阻塞操作。对于读取大文件这可能会阻塞整个事件循环。应考虑使用asyncio.to_thread或将文件操作委托给线程池避免影响 Server 处理其他并发请求虽然 stdio 模式下并发请求少但好习惯要养成。import aiofiles # 一个异步文件库 async def read_file_async(file_path: str max_lines: int) - str: try: async with aiofiles.open(file_path ‘r’ encoding‘utf-8’) as f: content await f.read() # … 处理截断逻辑 return result except …: # … 错误处理资源管理与生命周期确保 Server 在客户端断开连接后能正确清理资源。mcp库的Server类通常通过上下文管理器async with来处理。日志与监控在生产环境中需要为 Server 添加详细的日志记录记录收到的请求、调用的工具、执行耗时和任何错误。这有助于后期调试和性能分析。5.2 安全性加固重中之重将本地文件系统暴露给 AI 是极其危险的操作。必须实施严格的沙箱和权限控制。路径限制沙箱绝对不允许 Agent 读取任意路径。必须在 Server 端实施白名单或根目录限制。import os from pathlib import Path ALLOWED_BASE_DIR Path(“/path/to/allowed/directory”).resolve() async def read_file_safe(file_path: str max_lines: int) - str: requested_path Path(file_path).resolve() # 检查请求的路径是否在允许的目录下 try: requested_path.relative_to(ALLOWED_BASE_DIR) except ValueError: return “错误无权访问指定路径。” # 继续原有的读取逻辑…输入验证与净化除了 Pydantic 做的类型检查还要防范目录遍历攻击如../../../etc/passwd。上面的resolve()和relative_to()检查是有效手段。工具权限分级不是所有工具都对所有用户或所有会话开放。可以在 Server 初始化时接收一个令牌或配置根据此来决定暴露哪些工具。这需要更复杂的会话管理。网络传输安全如果未来改用 HTTP Transport必须启用 HTTPS 和身份认证如 API Key。5.3 扩展更多工具类型我们的例子是“读文件”但 MCP Server 的能力远不止于此。你可以轻松扩展写文件工具需要更谨慎的安全控制。执行命令工具风险极高必须限制可执行的命令列表并考虑在 Docker 容器或强隔离环境中运行。查询数据库工具封装 SQL 查询连接你的业务数据库。调用外部 API 工具作为企业内部 API 的网关。资源Resources例如将./docs目录下的所有 Markdown 文件作为资源发布AI 可以直接通过 URIfile://./docs/guide.md来引用其内容无需调用工具。每个新工具都遵循相同的模式定义 Pydantic 模型 - 实现业务函数 - 用mcp_server.tool()装饰器注册。5.4 与现有 LangChain Tools 生态的融合你可能会问既然 LangChain 本身就有Tool的抽象为什么还要绕道 MCPMCP 的核心价值在于标准化和跨平台。但短期内你可能已有大量传统的 LangChain Tools。有两种融合策略MCP Server 包装现有工具写一个 MCP Server其内部调用你现有的 LangChain Tool 函数。这样这些工具就同时拥有了 MCP 接口。双向转换使用一些适配器库在运行时将 MCP Tools 和 LangChain Tools 进行相互转换。社区正在出现这类工具。我个人在实际项目中的体会是对于全新的、希望长期维护且可能被多种 AI 客户端使用的核心能力优先实现为 MCP Server。对于快速原型、一次性任务或强依赖特定 LangChain 生态的功能可以直接使用 LangChain Tool。随着 MCP 生态的成熟其优势会越来越明显。最后再分享一个调试小技巧在开发 Agent 时如果工具调用结果不理想不要只盯着 Agent 的 Prompt 调。首先用mcp dev或手写请求单独测试你的 MCP Server确保它返回的格式和内容符合预期。其次仔细检查工具定义中的description和inputSchema它们是与 AI 模型沟通的“合同”描述的清晰度直接决定了工具被正确调用的概率。有时候优化工具描述比调整 Agent 的 System Prompt 更有效。