1. 项目概述为什么我们需要手写一个 MCP Server如果你最近在折腾 LangChain 或者 AI Agent大概率已经不止一次看到MCP这个词了。它可能出现在 Cursor 的配置里也可能在某个开源 Agent 项目的 README 中一闪而过。MCP全称Model Context Protocol直译过来是“模型上下文协议”。这个名字听起来有点抽象但它的目标却非常具体让大语言模型LLM能够安全、标准化地访问和使用外部工具与数据。你可以把它想象成 LLM 世界的“USB 协议”。在 USB 出现之前每个外设打印机、鼠标、U盘都需要自己的驱动和接口混乱不堪。MCP 想做类似的事情——为 LLM 定义一套标准化的“插拔”接口。通过 MCP一个 LangChain Agent 可以像即插即用一样动态发现并使用来自文件系统、数据库、搜索引擎甚至内部业务系统的能力而无需在 Agent 的代码里写死一堆if-else和 API 调用。那么为什么要“手写”一个本地的 MCP Server 呢市面上不是已经有很多现成的了吗比如搜索类的tavily-mcp文件操作类的filesystem-mcp。原因有三定制化、数据安全与学习深度。现成的 Server 解决的是通用问题但你的业务数据、内部工具往往是独特的。将敏感数据托付给第三方云服务存在风险而在本地部署一个专属的 MCP Server数据不出域可控性极高。更重要的是通过从零构建你能彻底吃透 MCP 的工作原理、数据流转和与 LangChain 的集成机制这是单纯调用 API 无法获得的认知。本文将带你完整走一遍这个过程从协议理解、Server 手写到接入 LangChain Agent 并实现一个真实的“本地知识库查询”功能。2. MCP 核心原理与架构拆解在动手写代码之前我们必须先搞清楚 MCP 到底规定了什么。它不是魔法而是一套基于 JSON-RPC 的轻量级通信规范。整个体系通常涉及三个角色MCP Client 通常是 LangChain、LlamaIndex 这类 AI 应用框架或者 Cursor、Claude Desktop 这类客户端。它负责向 Server 发起请求。MCP Server 我们即将要手写的部分。它封装了对特定资源如文件、数据库、API的操作能力并以“工具Tools”和“资源Resources”的形式暴露给 Client。Transport Layer 传输层定义了 Client 和 Server 之间如何通信。常见的有stdio标准输入输出和SSEServer-Sent Events。本地开发调试时stdio 最为简单直接。MCP 协议的核心交互可以概括为以下几个步骤初始化握手 Client 启动 Server 进程并通过 stdio 发送initialize请求。Server 回复自身的能力清单比如支持哪些工具。列出能力 Client 随后会调用tools/list或resources/list来获取 Server 提供的具体工具和资源列表。调用工具 当 Agent 需要执行某个动作时例如“读取 project/readme.md 文件”Client 会通过tools/call请求调用对应的工具。Server 执行实际操作如读取文件并将结果返回。推送通知可选 Server 可以主动向 Client 推送资源变更通知如文件被修改这是通过notifications实现的。一个关键设计是工具Tools的抽象。每个工具都有name、description和inputSchema。inputSchema是一个符合 JSON Schema 的对象用于严格定义调用此工具时需要传入的参数。这相当于给 LLM 提供了一份详细的“工具说明书”LLM 可以根据当前对话内容自动匹配并填充参数来调用工具。理解了这些我们就可以把目标具体化手写一个提供“本地文件读取”和“目录列表”工具的 MCP Server然后让一个 LangChain Agent 学会使用它来回答关于我们本地项目的问题。3. 手写本地 MCP Server 实战我们将使用 Python 来实现因为它与 LangChain 生态结合最紧密。核心是实现 MCP 协议定义的几个必需和可选的 JSON-RPC 方法。3.1 项目环境搭建与依赖安装首先创建一个干净的 Python 虚拟环境是个好习惯。这里我们使用venv。# 创建项目目录并进入 mkdir local-mcp-server cd local-mcp-server # 创建虚拟环境 python -m venv .venv # 激活虚拟环境 # Windows: .venv\Scripts\activate # Linux/Mac: source .venv/bin/activate接下来安装核心依赖。我们将使用官方提供的mcpSDK 来简化协议层的处理。同时为了构建一个简单的 CLI 或 Server 应用click库很有用。pip install mcp click注意mcp库是快速构建 MCP 组件的利器它封装了协议细节让我们可以更关注业务逻辑。确保你的 Python 版本在 3.8 以上。3.2 实现核心 Server 类我们创建一个名为server.py的文件。首先从mcp库导入必要的组件并定义一个FileServer类。import json import os from pathlib import Path from typing import Any, List import click from mcp import ClientSession, StdioServerParameters from mcp.server import Server from mcp.server.models import InitializationOptions import mcp.server.stdio from mcp.shared.exceptions import McpError # 初始化 MCP Server 实例 mcp_server Server(local-file-server)现在我们需要为 Server 注册工具。使用mcp_server.tool()装饰器来定义。mcp_server.tool() async def list_directory(path: str) - str: 列出指定目录下的文件和文件夹。 Args: path: 要列出的目录绝对路径。 Returns: 一个格式化的字符串包含目录下的项目列表。 try: target_path Path(path).resolve() # 安全检查确保路径存在且是目录 if not target_path.exists(): raise McpError(code404, messagef路径不存在: {path}) if not target_path.is_dir(): raise McpError(code400, messagef路径不是目录: {path}) items [] for item in target_path.iterdir(): # 简单标记目录和文件 item_type if item.is_dir() else items.append(f{item_type} {item.name}) if not items: return f目录 {path} 为空。 return f路径 {path} 下的内容\n \n.join(items) except PermissionError: raise McpError(code403, messagef无权限访问目录: {path}) except Exception as e: raise McpError(code500, messagef列出目录时出错: {str(e)}) mcp_server.tool() async def read_file(filepath: str) - str: 读取指定文件的内容。 Args: filepath: 要读取的文件的绝对路径。 Returns: 文件的文本内容。对于非文本文件会提示错误。 try: target_file Path(filepath).resolve() # 安全检查 if not target_file.exists(): raise McpError(code404, messagef文件不存在: {filepath}) if not target_file.is_file(): raise McpError(code400, messagef路径不是文件: {filepath}) # 简单判断是否为文本文件可根据需求扩展 if target_file.suffix.lower() in [.png, .jpg, .jpeg, .pdf, .zip]: raise McpError(code415, messagef不支持读取二进制文件: {filepath}) # 读取文件内容 content target_file.read_text(encodingutf-8) return content except UnicodeDecodeError: raise McpError(code415, messagef文件编码不是 UTF-8无法读取: {filepath}) except PermissionError: raise McpError(code403, messagef无权限读取文件: {filepath}) except Exception as e: raise McpError(code500, messagef读取文件时出错: {str(e)})代码解读与注意事项工具定义每个工具都是一个async函数。mcp_server.tool()装饰器会自动收集函数的名称、文档字符串和参数信息并将其转化为 MCP 协议所需的工具描述。输入验证与安全这是 Server 实现中最关键的一环。我们绝不能盲目信任 Client 传入的路径。Path(...).resolve()解析路径消除..和符号链接得到绝对路径。路径遍历攻击防护上述代码是基础防护。在生产环境中你必须添加更严格的检查例如将可访问的路径限制在某个“根目录”下防止用户通过../../etc/passwd这样的路径访问系统文件。一个简单的改进是添加if not str(resolved_path).startswith(ALLOWED_ROOT): raise ...。存在性与类型检查确保路径存在并且是预期的类型文件或目录。错误处理我们使用mcp.shared.exceptions.McpError来抛出标准化的错误。这比普通的 Python 异常更能让 Client 理解错误原因。错误码参考了 HTTP 状态码这是一种通用惯例。异步支持MCP SDK 基于异步asyncio。即使我们的文件操作是同步的工具函数也需要定义为async。如果工具涉及网络 I/O如调用另一个 API这里就能真正发挥异步的优势。3.3 实现 Server 主循环与 stdio 通信工具定义好了接下来需要让 Server 能够通过标准输入输出与 Client 对话。我们创建一个run_server函数并使用click来构建一个简单的命令行入口。async def run_server(): 运行 MCP Server 的主异步函数。 # 创建 stdio 通信层 async with mcp.server.stdio.stdio_server() as (read_stream, write_stream): # 创建 ClientSession 来处理与 Client 的会话 async with ClientSession( read_stream, write_stream, InitializationOptions(server_namelocal-file-server) ) as session: # 这里 session 会自动处理握手和请求分发。 # 我们只需要启动 Server让它开始处理来自 session 的请求。 await mcp_server.run( session, # 这里可以传入 InitializationOptions但我们在 ClientSession 中已经指定了 ) # Server 会一直运行直到连接断开。 click.command() def cli(): 启动本地文件 MCP Server。 import asyncio click.echo(本地文件 MCP Server 启动... (使用 CtrlC 退出)) try: asyncio.run(run_server()) except KeyboardInterrupt: click.echo(\nServer 已停止。) except Exception as e: click.echo(fServer 运行出错: {e}, errTrue) if __name__ __main__: cli()关键点解析mcp.server.stdio.stdio_server()这是一个上下文管理器它为我们配置好了标准输入和输出流这是 MCP over stdio 传输的标准方式。ClientSession它代表一个与 Client 的连接会话。它封装了 JSON-RPC 的通信细节并将接收到的tools/call请求路由给我们之前用mcp_server.tool()注册的函数。mcp_server.run(session, ...)这是启动 Server 处理循环的核心。它将 Server 实例与 Session 绑定开始监听并处理请求。运行方式最终这个 Server 将通过python server.py启动。它不会打开网络端口而是静静地等待从标准输入接收 JSON-RPC 请求并将结果输出到标准输出。这正是 LangChain 等 Client 所期望的通信方式。至此一个功能完整、具备基础安全意识的本地文件 MCP Server 就完成了。你可以通过运行python server.py来测试它是否启动成功它会挂起等待输入。4. 接入 LangChain Agent 并实现智能问答Server 准备好了现在我们需要一个 LangChain Agent 来使用它。我们将创建一个agent.py文件。4.1 配置 LangChain 环境与 MCP 集成首先确保安装了 LangChain 和相关依赖。我们使用 OpenAI 的模型作为 Agent 的“大脑”。pip install langchain langchain-openai在agent.py中我们首先配置 MCP Client 并连接到我们的 Server。import asyncio from pathlib import Path from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from mcp import ClientSession from mcp.client.stdio import stdio_client async def create_mcp_tools(): 创建与本地 MCP Server 的连接并获取其暴露的工具。 # 指定我们刚才编写的 server.py 的路径 server_script_path Path(__file__).parent / server.py # 使用 stdio_client 启动 Server 进程并建立连接 async with stdio_client( command[python, str(server_script_path)], ) as (read_stream, write_stream): async with ClientSession(read_stream, write_stream) as session: # 初始化连接 await session.initialize() # 列出 Server 提供的所有工具 result await session.list_tools() tools [] # 将 MCP 工具转换为 LangChain 可用的工具格式 for tool_info in result.tools: # 我们需要创建一个适配器函数 async def tool_adapter(**kwargs): # 实际调用 MCP Server 的工具 response await session.call_tool(tool_info.name, argumentskwargs) # MCP 返回的结果在 content 字段中是一个列表 if response.content: # 通常我们取第一个文本内容块 for block in response.content: if block.type text: return block.text return 工具执行完成但未返回文本内容。 # 创建 LangChain Tool 对象 from langchain_core.tools import Tool langchain_tool Tool( nametool_info.name, descriptiontool_info.description or f调用 MCP 工具 {tool_info.name}, functool_adapter, # 注意这里需要处理异步稍后解决 args_schemaNone, # 可以基于 tool_info.inputSchema 创建这里简化 ) tools.append(langchain_tool) # 注意这里不能直接返回因为 session 离开 async with 后连接会关闭。 # 我们需要一种方式在 Agent 生命周期内保持连接。 # 更合理的架构是将 Session 管理封装到一个自定义的 Tool 类中。 # 为了示例清晰我们采用一个简化方案只获取一次工具列表并创建“一次性”调用函数。 # 但请注意这在实际生产环境中并不高效。 # 让我们重新设计这部分...第一个大坑会话管理与工具封装上面的代码揭示了一个关键问题ClientSession必须在async with块内使用。一旦离开这个块连接就关闭了。而 LangChain Agent 可能会在后续多次调用工具。我们不能在每次调用时都重新启动 Server 进程。解决方案创建一个自定义的LangChainTool类它在初始化时启动并保持 MCP 会话在 Agent 整个运行期间提供工具调用能力并在最后妥善关闭。4.2 重构创建可持续的 MCP 工具封装类我们调整思路创建一个专门管理 MCP 连接和工具调用的类。# 在 agent.py 中继续 import asyncio from contextlib import asynccontextmanager from typing import List, Any from langchain_core.tools import BaseTool from pydantic import BaseModel, Field from mcp import ClientSession, StdioServerParameters from mcp.client.stdio import stdio_client import subprocess import sys class MCPToolWrapper(BaseTool): 封装一个 MCP 工具使其能被 LangChain Agent 使用。 name: str description: str mcp_session: ClientSession # 持有会话引用 args_schema: type[BaseModel] | None None def _run(self, *args: Any, **kwargs: Any) - Any: 同步运行方法。LangChain 的 AgentExecutor 默认调用这个。 但我们的 MCP 调用是异步的所以这里需要特殊处理。 # 在同步环境中运行异步函数 return asyncio.run(self._arun(*args, **kwargs)) async def _arun(self, *args: Any, **kwargs: Any) - str: 异步运行方法。 try: response await self.mcp_session.call_tool(self.name, argumentskwargs) if response.content: for block in response.content: if block.type text: return block.text return 工具执行完成。 except Exception as e: return f调用工具 {self.name} 时出错: {str(e)} class MCPServerManager: 管理 MCP Server 进程和会话的生命周期。 def __init__(self, server_command: List[str]): self.server_command server_command self.process None self.session None async def __aenter__(self): 异步上下文管理器入口启动 Server 并创建会话。 # 启动子进程 self.process await asyncio.create_subprocess_exec( *self.server_command, stdinasyncio.subprocess.PIPE, stdoutasyncio.subprocess.PIPE, stderrasyncio.subprocess.PIPE, ) # 创建 stdio 流 # 注意asyncio.subprocess.PIPE 提供的是字节流需要包装 # mcp.client.stdio 提供了更便捷的方式但这里我们手动包装以说明原理 # 实际上使用 stdio_client 上下文管理器更安全它会处理这些。 # 让我们使用更稳健的方式 from mcp.client.stdio import stdio_client # stdio_client 内部会处理进程的启动和流的创建比手动管理更可靠。 # 因此我们重构一下不在这个类里管理进程而是让 stdio_client 全权负责。 # 这引出了第二个设计决策Manager 只管理会话不管理进程。 # 我们简化这个类... print(MCP Server 进程已启动。) return self async def __aexit__(self, exc_type, exc_val, exc_tb): 异步上下文管理器出口清理资源。 if self.process: try: self.process.terminate() await self.process.wait() except ProcessLookupError: pass print(MCP Server 进程已停止。) # 鉴于复杂度我们采用一个更直接、清晰的示例方案。 # 下面我们将编写一个“一次性”的完整 Agent 运行脚本它启动 Server创建工具运行 Agent然后退出。 # 这对于理解和测试已经足够。4.3 最终整合一个完整的可运行 Agent 示例为了避免陷入底层管理的复杂性我们展示一个端到端的、可运行的脚本。它会在一个事件循环中完成所有工作。# agent_demo.py import asyncio from pathlib import Path from langchain.agents import AgentExecutor, create_tool_calling_agent from langchain_core.prompts import ChatPromptTemplate from langchain_openai import ChatOpenAI from langchain_core.tools import Tool from mcp import ClientSession from mcp.client.stdio import stdio_client import sys async def main(): 主异步函数演示完整的 MCP Server 接入 LangChain Agent 流程。 # 1. 启动 MCP Server 并建立连接 server_script_path Path(__file__).parent / server.py print(f正在启动 MCP Server: {server_script_path}) async with stdio_client( command[sys.executable, str(server_script_path)], # 使用当前 Python 解释器 ) as streams: read_stream, write_stream streams async with ClientSession(read_stream, write_stream) as session: await session.initialize() # 2. 获取 Server 提供的工具列表 list_tools_result await session.list_tools() print(f从 Server 发现 {len(list_tools_result.tools)} 个工具。) # 3. 将 MCP 工具包装成 LangChain Tool langchain_tools [] for mcp_tool in list_tools_result.tools: # 动态创建工具调用函数 async def make_tool_func(tool_name): async def tool_func(**kwargs): try: response await session.call_tool(tool_name, argumentskwargs) for content in response.content: if content.type text: return content.text return 工具执行成功无文本返回。 except Exception as e: return f工具调用错误: {str(e)} return tool_func tool_func await make_tool_func(mcp_tool.name) # 创建 LangChain Tool # 注意这里简化了没有严格传递 inputSchema。生产环境需要处理。 langchain_tool Tool( namemcp_tool.name, descriptionmcp_tool.description or fMCP Tool: {mcp_tool.name}, functool_func, # LangChain 新版支持协程函数 ) langchain_tools.append(langchain_tool) print(f - 已加载工具: {mcp_tool.name}) # 4. 创建 LangChain Agent # 使用 OpenAI 模型 (请设置你的 OPENAI_API_KEY 环境变量) llm ChatOpenAI(modelgpt-4o-mini, temperature0) # 定义 Agent 的提示词模板 prompt ChatPromptTemplate.from_messages([ (system, 你是一个有帮助的助手可以访问本地文件系统。请根据用户问题使用合适的工具来获取信息然后给出回答。如果你没有合适的工具或无法获取信息请如实告知。), (placeholder, {chat_history}), (human, {input}), (placeholder, {agent_scratchpad}), ]) # 创建 Agent agent create_tool_calling_agent(llm, langchain_tools, prompt) # 创建 Agent 执行器 agent_executor AgentExecutor(agentagent, toolslangchain_tools, verboseTrue, handle_parsing_errorsTrue) # 5. 运行一个示例查询 print(\n--- Agent 测试开始 ---) test_queries [ 请列出当前项目根目录即 server.py 所在目录下有什么文件, 请读取 server.py 文件的前50行告诉我这个文件是做什么的, # 你可以尝试一个需要链式调用的查询例如 # “先列出根目录然后读取其中的 README.md 文件” ] for query in test_queries: print(f\n用户: {query}) try: # 注意agent_executor.invoke 在 LangChain 新版中是同步的但内部工具是异步的。 # 我们需要在异步环境中使用 ainvoke。 result await agent_executor.ainvoke({input: query}) print(f助手: {result[output]}) except Exception as e: print(f执行出错: {e}) print(\n--- Agent 测试结束 ---) # 会话和 Server 连接会在 async with 块结束时自动关闭 if __name__ __main__: asyncio.run(main())关键改进与说明一站式管理整个流程在一个async with栈中完成确保了 Server 进程和 Session 的正确生命周期管理。工具函数包装我们为每个 MCP 工具动态创建了一个异步函数tool_func它内部调用session.call_tool。这个函数被赋予 LangChain 的Tool对象。新版 LangChain 的Tool支持协程函数作为func。异步调用我们使用 AgentExecutor 的ainvoke方法进行异步调用这与我们异步的工具函数相匹配。错误处理在工具函数和 Agent 执行层面都添加了基本的错误处理。运行这个演示脚本前请确保已设置OPENAI_API_KEY环境变量。export OPENAI_API_KEYyour-api-key-here # 然后运行 python agent_demo.py如果一切顺利你将看到 Agent 启动加载工具并成功调用你的本地 MCP Server 来回答关于文件系统的问题。5. 进阶探讨与避坑指南通过上面的步骤你已经实现了一个可用的基础系统。但在实际生产或更复杂的场景中你会遇到更多挑战。以下是基于经验的进阶要点和常见问题排查。5.1 性能、安全与生产化考量Server 常驻与连接池上述示例每次运行 Agent 都重启 Server效率低下。生产环境应将 MCP Server 作为独立的守护进程运行并使用SSE (Server-Sent Events)或WebSocket传输层让多个 Client 可以连接。LangChain 社区正在推动对 SSE 传输的支持。更严格的安全沙箱我们的路径检查是基础。你必须实现一个绝对路径白名单机制。例如在 Server 初始化时传入一个root_dir所有工具操作都限制在此目录下。class SecureFileServer: def __init__(self, root_dir: Path): self.root_dir root_dir.resolve() def _secure_path(self, user_path: str) - Path: resolved (self.root_dir / user_path).resolve() # 关键检查确保解析后的路径仍在 root_dir 之下 if not str(resolved).startswith(str(self.root_dir)): raise McpError(code403, message访问路径越界) return resolved工具参数的 JSON Schema 精确传递我们的示例简化了args_schema。为了获得最佳的 LLM 调用效果应该将 MCP 工具定义中的inputSchema完整地转换为 Pydantic 模型并设置给 LangChain Tool。这能极大提升 Agent 调用工具时的参数填充准确率。资源Resources与提示词注入MCP 除了工具还有“资源”概念。资源如一个常驻的提示词模板、一个参考文档可以被“注入”到 LLM 的上下文窗口中而不需要显式调用工具。这对于为 Agent 提供静态背景知识非常有用。我们的示例未涉及但这是 MCP 的一个重要特性。5.2 常见问题排查实录在开发和集成过程中你可能会遇到以下问题问题1启动 Server 后Client 连接立即断开或报错 “Invalid JSON”。原因最可能是 Server 的 stdio 输出不符合 MCP 协议。Server 在初始化前打印了任何调试信息如print(Server starting...)都会破坏 JSON-RPC 消息。解决绝对禁止在 Server 主逻辑中使用print输出非协议内容。所有日志应重定向到标准错误sys.stderr。确保server.py中只有通过mcp_server发送的 JSON 数据会走到标准输出。问题2Agent 无法识别或错误调用工具。原因A工具描述description不够清晰。LLM 依赖描述来选择工具。确保描述简洁、准确说明工具的功能和输入参数的意义。原因B没有正确传递args_schema。LangChain 和底层 LLM 需要知道参数的类型字符串、数字、布尔值等。解决完善工具描述并实现args_schema的转换。可以写一个辅助函数将 MCP 的inputSchema(JSON Schema) 转换为 Pydantic 的BaseModel。问题3处理大型文件或复杂操作时超时。原因文件读取或目录遍历操作是同步且阻塞的如果操作非常耗时会卡住整个事件循环。解决将耗时的 I/O 操作放在线程池中执行避免阻塞异步主线程。可以使用asyncio.to_thread。mcp_server.tool() async def read_large_file(filepath: str) - str: def _sync_read(): # ... 同步的文件读取逻辑 pass content await asyncio.to_thread(_sync_read) return content问题4如何调试 MCP 协议通信方法一种有效的方式是使用mcpSDK 的日志功能或者编写一个简单的“中间人”调试脚本它位于 Client 和 Server 之间打印出所有经过的 JSON-RPC 消息。这能帮你清晰看到握手、列表、调用每个阶段的数据。手写一个 MCP Server 并接入 LangChain初看步骤不少但拆解后无非是“实现协议端点”、“封装工具”、“管理连接”三件事。这个过程最能加深你对 AI Agent 如何与外界交互的理解。当你需要让 Agent 访问公司内部的 CRM、独特的数据库或私有 API 时自定义 MCP Server 就成了那把唯一的钥匙。从这个小型的文件服务器开始你可以逐步扩展让它成为连接 LLM 与你的数字世界的强大桥梁。
