尧图网站建设 
1. 项目概述为什么我们需要一个“万能接口”最近在折腾AI Agent开发的朋友估计都绕不开一个词MCP。这玩意儿现在火得不行感觉一夜之间就成了Agent生态里的“当红炸子鸡”。我最早接触它是因为在尝试让我的Agent去调用一个内部的数据查询工具时遇到了大麻烦——每个工具都得写一堆胶水代码适配不同的输入输出格式调试起来简直是一场噩梦。直到发现了MCP我才意识到我们缺的从来不是强大的模型或者复杂的逻辑而是一个能让它们和外部世界顺畅对话的“普通话”。简单来说MCPModel Context Protocol你可以把它理解为AI Agent领域的“USB-C接口”标准。在USB-C统一之前你的手机、电脑、充电宝各有各的接口出门得带一堆线。MCP干的就是同样的事它定义了一套统一的协议让任何AI模型比如Claude、GPT都能以一种标准化的方式去发现、调用和管理外部工具、数据源或服务我们称之为“资源”。开发者不再需要为每一个工具单独编写适配器只需要按照MCP标准把工具“包装”成一个MCP Server任何兼容MCP的AI客户端比如Claude Desktop、Cursor就能即插即用。这解决了AI Agent开发中的一个核心痛点上下文构建的标准化与工具使用的碎片化。以前给Agent“扩展能力”是个高度定制化的脏活累活现在有了MCP就像是给Agent世界建立了“应用商店”和统一的安装规范。接下来我们就深入这个协议的内部看看它究竟是如何运作以及我们如何利用它来构建更强大、更易集成的AI智能体。2. MCP协议核心架构与设计哲学拆解要理解MCP不能只看它定义了几个API更要理解它背后的设计哲学。它不是一个重量级的框架而是一个轻量级的、基于JSON-RPC的通信协议。这种选择本身就体现了其核心思想简单、通用、与语言无关。2.1 核心组件Server, Client 与 TransportMCP的架构非常清晰主要包含三个角色MCP Server资源提供方这是能力的封装者。它可以是一个简单的脚本提供天气查询也可以是一个复杂的后端服务提供数据库操作或内部API。它的职责是按照MCP协议向客户端宣告自己提供了哪些“资源”工具或数据源并在客户端请求时执行具体操作。MCP Client资源消费方通常是AI应用本身比如Claude Desktop、Codeium或你自行开发的Agent框架。它的职责是发现可用的Server列出其提供的资源并在需要时发起调用请求。Transport传输层连接Server和Client的通道。MCP设计上支持多种传输方式目前最常见的是stdio标准输入输出和SSEServer-Sent Events。Stdio模式简单直接适合本地集成SSE则更适合网络环境。协议本身不关心传输细节这保证了其灵活性。这种架构的优势在于解耦。作为Server开发者你只需要关心如何实现工具功能并按照协议返回标准化的JSON。作为Client开发者你只需要实现协议解析就能接入无数个Server。这种“插座-插头”模型极大地降低了生态建设的门槛。2.2 协议核心资源Resources与工具ToolsMCP将外部能力抽象为两种核心类型这是理解其能力边界的关键资源Resources可以理解为静态或动态的“数据源”。它有一个唯一的URI如file:///path/to/doc或dynamic://stock/price/AAPL以及对应的文本内容。Client可以“读取”资源。例如一个Server可以提供一个“今日待办事项列表”资源AI在回答相关问题时会先读取这个资源来获取上下文。资源支持内容类型text, image, pdf等未来扩展性很强。工具Tools这就是我们通常理解的“可执行函数”。每个工具需要定义清晰的输入参数JSON Schema格式。当AI决定使用某个工具时Client会调用它并传入参数。Server执行后返回结果。比如“发送邮件”、“查询数据库”、“生成图表”都可以是工具。一个关键设计亮点MCP协议要求Server在初始化时就向Client宣告自己提供的所有资源和工具的“列表”和“描述”。这允许AI在生成回答的初期就充分知晓自己“手头有哪些牌可以打”从而做出更合理的规划和决策而不是盲目地生成文本后再去尝试调用可能不存在的功能。2.3 会话管理与上下文生命周期MCP协议是围绕“会话Session”建立的。一个Client和Server的连接就是一个会话。会话内有明确的生命周期初始化Initialize握手交换能力信息。资源/工具列表ListClient获取Server的能力清单。调用CallClient请求调用工具或读取资源。通知NotificationsServer可以主动推送资源变更如“文件已更新”帮助AI保持上下文新鲜度。结束Shutdown优雅关闭连接。这个生命周期管理使得连接状态清晰可控避免了资源泄漏或状态混乱的问题特别适合需要长期运行、动态感知环境变化的Agent场景。3. 手把手实战从零构建你的第一个MCP Server理论说得再多不如动手做一遍。我们以构建一个最简单的“时间查询服务器”为例展示如何从零实现一个MCP Server。我将使用Python因为它生态友好有现成的SDK。3.1 环境准备与SDK选择首先你需要一个Python环境3.8。官方和社区提供了多种SDK来抽象协议细节让我们专注于业务逻辑。这里我们使用mcp这个Python库它目前是社区最活跃的选择之一。# 创建项目目录并进入 mkdir my-first-mcp-server cd my-first-mcp-server # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装MCP SDK pip install mcp3.2 编写Server核心逻辑创建一个名为time_server.py的文件。我们的目标是提供两个能力1. 一个名为get_current_time的工具返回当前时间2. 一个名为time_guide的资源提供如何使用本服务器的说明。# time_server.py import asyncio from datetime import datetime from mcp import Server, StdioServerParameters from mcp.types import Tool, TextContent, ResourceTemplate # 1. 创建Server实例 server Server(time-server) # 2. 定义一个工具Tool server.list_tools() async def handle_list_tools(): # 返回本Server提供的所有工具描述 return [ Tool( nameget_current_time, description获取当前的系统日期和时间。, inputSchema{ type: object, properties: { format: { type: string, description: 时间格式可选。默认为%Y-%m-%d %H:%M:%S。例如%Y年%m月%d日。, default: %Y-%m-%d %H:%M:%S } } } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): # 根据工具名分发处理逻辑 if name get_current_time: fmt arguments.get(format, %Y-%m-%d %H:%M:%S) current_time datetime.now().strftime(fmt) return [ TextContent( typetext, textf当前时间是{current_time} ) ] else: raise ValueError(f未知工具: {name}) # 3. 定义一个资源Resource server.list_resources() async def handle_list_resources(): # 返回本Server提供的所有资源描述 return [ ResourceTemplate( uritime://guide, name时间服务器使用指南, description关于如何使用本时间查询服务器的说明文档。, mimeTypetext/plain ) ] server.read_resource() async def handle_read_resource(uri: str): # 根据资源URI返回内容 if uri time://guide: guide_text 时间服务器使用指南 本服务器提供以下功能 1. 工具 get_current_time: 获取当前时间。 - 可选参数 format: 指定时间格式字符串例如 %Y年%m月%d日。 2. 资源 time://guide: 本说明文档。 示例请求获取当前年月日。 return TextContent(typetext, textguide_text) else: raise ValueError(f未知资源: {uri}) # 4. 主函数使用Stdio传输启动服务器 async def main(): async with server.run_stdio(StdioServerParameters()): # 保持服务器运行直到被终止 await asyncio.Future() if __name__ __main__: asyncio.run(main())代码解读与注意事项装饰器模式server.list_tools()和server.call_tool()是SDK提供的装饰器用于注册处理函数。这种模式清晰地将“能力声明”和“能力执行”分开。输入模式Schema定义工具时inputSchema至关重要。它用JSON Schema精确描述了AI调用该工具时需要或可以传递的参数。写得好AI调用起来就准确写得模糊就容易出错。建议为每个参数提供详细的description和合理的default值。资源URI资源URI是一个字符串标识符建议使用自定义的Scheme如time://来避免冲突。它不是真实的网络地址只是一个协议内的唯一ID。异步编程MCP SDK基于 asyncio所有处理函数都是async的。确保你的业务逻辑也是非阻塞的如果是耗时操作如网络请求要使用异步库。3.3 运行与测试你的Server保存文件后你可以直接运行它。但由于MCP Client通常通过stdin/stdout与其通信直接运行看不到输出。我们需要一个测试Client。最简单的方法是使用官方提供的mcpCLI 工具需单独安装mcp-cli或使用一个已集成MCP Client的应用。这里我们用一个更直观的方法写一个极简的测试脚本。# test_client.py (仅用于基础测试非标准MCP Client) import subprocess import json import sys # 启动Server进程 proc subprocess.Popen( [sys.executable, time_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 模拟发送一个初始化请求简化版 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 0.1.0, capabilities: {}, clientInfo: {name: test-client} } } proc.stdin.write(json.dumps(init_request) \n) proc.stdin.flush() # 读取一行响应 response proc.stdout.readline() print(Server响应:, response) # 结束进程 proc.terminate()更实际的测试是将其配置到真正的MCP Client中如 Claude Desktop。在Claude Desktop中配置MCP Server打开Claude Desktop设置。找到“开发者设置”或“MCP服务器”部分。点击“添加服务器”。配置如下名称My Time Server命令python或你的Python解释器完整路径参数/path/to/your/time_server.py的完整路径保存并重启Claude Desktop。重启后当你与Claude对话时它就能自动识别并使用你编写的get_current_time工具了。你可以尝试说“请告诉我现在的时间用中文格式。”4. 高级应用连接真实世界与复杂工具集成构建一个返回时间的Server只是入门。MCP真正的威力在于连接各种异构系统。下面我们探讨几个更复杂的场景。4.1 集成外部API构建天气查询Server假设我们要集成一个天气API。这里以假想的weatherapi.com为例。# weather_server.py (部分关键代码) import aiohttp # ... 其他导入 ... server Server(weather-server) WEATHER_API_KEY your_api_key_here # 切记不要硬编码在代码中应使用环境变量 server.list_tools() async def handle_list_tools(): return [ Tool( nameget_weather, description根据城市名称查询实时天气情况。, inputSchema{ type: object, properties: { city: { type: string, description: 城市名称例如Beijing, Shanghai, New York }, days: { type: integer, description: 预报天数1-3默认为1, default: 1 } }, required: [city] # 指定必填参数 } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_weather: city arguments[city] days arguments.get(days, 1) # 使用异步HTTP客户端调用外部API async with aiohttp.ClientSession() as session: url fhttp://api.weatherapi.com/v1/forecast.json params { key: WEATHER_API_KEY, q: city, days: days, aqi: no, alerts: no } async with session.get(url, paramsparams) as resp: if resp.status 200: data await resp.json() current data[current] forecast data[forecast][forecastday][0][day] text f{city}当前天气{current[condition][text]}温度{current[temp_c]}°C体感{current[feelslike_c]}°C湿度{current[humidity]}%。 text f\n今日预报最高{forecast[maxtemp_c]}°C最低{forecast[mintemp_c]}°C。 return [TextContent(typetext, texttext)] else: error_text await resp.text() return [TextContent(typetext, textf查询天气失败{error_text})] # ... 错误处理 ...关键点异步HTTP请求使用aiohttp等异步库避免阻塞Server主线程。错误处理外部API调用可能失败必须将清晰的错误信息返回给AI而不是抛出未处理的异常导致整个会话中断。安全API密钥等敏感信息务必通过环境变量 (os.getenv) 或配置文件读取绝对不要提交到版本库。4.2 包装命令行工具赋予AI系统操作能力让AI安全地执行系统命令是一个强大但危险的能力。MCP可以提供一个受控的接口。# sys_tool_server.py (部分关键代码) import subprocess import shlex from typing import List # 定义一个允许执行的命令白名单 ALLOWED_COMMANDS { list_directory: {cmd: [ls, -la], desc: 列出当前目录的详细内容。}, disk_usage: {cmd: [df, -h], desc: 查看磁盘使用情况。}, process_list: {cmd: [ps, aux], desc: 显示所有运行中的进程。}, } server.list_tools() async def handle_list_tools(): tools [] for name, info in ALLOWED_COMMANDS.items(): tools.append(Tool( namename, descriptioninfo[desc], inputSchema{type: object, properties: {}} # 此例中无参数 )) return tools server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name in ALLOWED_COMMANDS: cmd_list ALLOWED_COMMANDS[name][cmd] try: # 安全地执行白名单内的命令 result subprocess.run( cmd_list, capture_outputTrue, textTrue, timeout10 # 设置超时防止挂起 ) output f命令执行完成。\n返回码{result.returncode}\n输出\n{result.stdout} if result.stderr: output f\n错误输出\n{result.stderr} return [TextContent(typetext, textoutput)] except subprocess.TimeoutExpired: return [TextContent(typetext, text命令执行超时。)] except Exception as e: return [TextContent(typetext, textf执行命令时发生异常{e})] else: raise ValueError(f命令不在白名单内: {name})重要警告提供命令行工具接口风险极高。必须严格遵守以下原则最小权限原则像上面一样严格使用命令白名单绝不接受用户动态输入的命令字符串。沙盒环境考虑在Docker容器或具有严格权限限制的用户环境中运行此类Server。审计日志记录所有工具调用和结果便于追踪和审计。超时控制必须为子进程设置超时防止恶意或错误命令无限运行。4.3 动态资源与通知实现实时数据同步MCP支持Server主动向Client发送通知notifications/resources/updated这对于需要实时更新上下文的场景非常有用。例如一个监控日志文件的Server。# log_monitor_server.py (概念性代码展示通知机制) import asyncio import watchfiles # 需要安装 watchfiles 库 server.list_resources() async def handle_list_resources(): return [ ResourceTemplate( urifile:///var/log/app/current.log, name应用实时日志, description应用程序的最新日志内容。, mimeTypetext/plain ) ] async def watch_log_file(server_instance): log_path /var/log/app/current.log # 使用 watchfiles 监控文件变化 async for changes in watchfiles.awatch(log_path): for change in changes: if change[0] watchfiles.Change.modified: # 文件被修改向所有连接的Client发送资源更新通知 # 注意这需要Server实例能管理多个Client会话标准SDK可能需要扩展或使用特定模式 # 此处为概念展示 # await server_instance.send_notification(resources/updated, { # uri: file:///var/log/app/current.log # }) print(f[通知] 日志文件已更新: {log_path}) # 在实际实现中你需要一个方法来广播通知给相关会话。 # 在main函数中启动监控任务 async def main(): # ... 初始化server ... # 创建后台任务监控文件 asyncio.create_task(watch_log_file(server)) # ... 运行server ...这个例子展示了MCP协议如何支持“主动推送”模式使得AI的上下文可以随着外部数据的变化而自动更新为实现真正动态、感知环境的Agent奠定了基础。5. 生态现状、最佳实践与避坑指南MCP协议虽然理念先进但作为一个新兴标准其生态和实践仍在快速演进中。根据我这段时间的实践分享一些观察和经验。5.1 当前生态与工具链主流Client支持Claude Desktop目前对MCP支持最友好、最成熟的客户端。配置简单体验流畅是学习和测试MCP Server的首选环境。Cursor IDE作为AI原生编辑器也集成了MCP Client可以直接在编码环境中调用MCP工具场景非常契合。其他框架如LangChain、LlamaIndex等Agent框架正在逐步增加对MCP的原生支持或社区插件。Server SDK与工具Python (mcp)社区最活跃文档和示例相对丰富是快速上手的最佳选择。TypeScript/Node.js (modelcontextprotocol/sdk)官方SDK由Anthropic维护类型定义完善适合前端或Node.js生态的开发者。CLI工具 (mcp-cli)用于调试、测试和发现MCP Server的实用命令行工具。社区Server仓库GitHub上已经出现了不少优秀的开源MCP Server项目例如github-search-mcp搜索GitHub仓库。sqlite-mcp操作SQLite数据库。filesystem-mcp安全的文件系统操作。多去这些项目看看是学习如何编写高质量Server的绝佳途径。5.2 开发与部署最佳实践清晰的工具命名与描述工具和资源的name、description是AI理解其功能的唯一途径。务必使用清晰、无歧义的自然语言描述。好的描述应包含目的、输入参数的含义、输出结果的格式。严谨的输入模式Schema定义充分利用JSON Schema的type,enum,pattern,minimum/maximum等属性来约束输入。这能极大减少AI调用时的参数错误。为可选参数设置合理的default值。错误处理的友好性Server端必须做好错误处理。当工具执行失败时返回结构化的错误信息帮助AI理解问题所在例如“数据库连接失败请检查网络”比一个空的错误堆栈更有用。性能与超时工具执行应设置超时。对于可能耗时的操作如网络请求要确保是异步的避免阻塞整个Server影响其他请求。配置化与安全Server的配置如API端点、密钥必须通过环境变量或配置文件管理。在代码中硬编码敏感信息是严重的安全隐患。日志与监控为Server添加适当的日志记录记录工具的调用、参数和结果注意脱敏这对于调试和运维至关重要。5.3 常见问题与排查技巧在实际开发和集成中你肯定会遇到各种问题。下面是一个快速排查清单问题现象可能原因排查步骤Claude Desktop 无法识别 Server1. 配置命令或路径错误。2. Server启动失败或立即崩溃。3. 协议版本不兼容。1. 在终端手动运行配置的命令看Server能否正常启动并等待输入。2. 查看Claude Desktop的日志通常可在设置中找到日志文件路径。3. 确保Server实现了正确的初始化握手。AI 不调用工具1. 工具描述不够清晰AI不理解何时使用。2. 工具列表未正确返回。3. AI模型本身策略限制。1. 优化工具的名称和描述使其意图更明确。2. 使用mcp-cli或写测试脚本连接Server手动发送tools/list请求检查返回是否正确。3. 在对话中明确提示AI可以使用某个工具。工具调用返回错误1. Server端逻辑错误或异常未捕获。2. AI传递的参数不符合Schema。3. 网络或外部服务问题。1. 查看Server的运行日志或标准错误输出stderr。2. 在call_tool处理函数开头打印收到的arguments检查其格式。3. 模拟AI的请求用curl或脚本直接测试Server的接口。连接不稳定或断开1. Server进程崩溃。2. 传输层如stdio缓冲区问题。3. 心跳或保活机制缺失。1. 加强Server的异常处理避免未处理的异常导致进程退出。2. 确保读写stdin/stdout时遵循行分隔的JSON-RPC格式。3. 对于长连接考虑实现更健壮的传输层或使用SSE。一个关键的调试工具mcp-cli。安装后pip install mcp-cli你可以用它来直接与你的Server交互模拟Client的行为这是验证Server是否按协议工作的最直接方法。# 使用 mcp-cli 测试你的 Server mcp run python time_server.py # 连接后可以尝试输入 /list 查看工具和资源/call tool_name 调用工具。MCP协议的出现标志着AI Agent从“单机智能”走向“生态智能”的关键一步。它通过一个简洁的接口标准解决了工具集成的混乱问题。虽然目前生态还在早期但方向已经非常明确。对于开发者而言现在开始学习并构建自己的MCP Server相当于提前布局了未来AI应用生态的“基础设施层”。从简单的工具封装开始逐步尝试连接更复杂的系统你会发现让AI真正成为你的得力助手门槛正在变得越来越低。 MCP协议详解：构建AI Agent统一工具接口的实践指南
1. 项目概述为什么我们需要一个“万能接口”最近在折腾AI Agent开发的朋友估计都绕不开一个词MCP。这玩意儿现在火得不行感觉一夜之间就成了Agent生态里的“当红炸子鸡”。我最早接触它是因为在尝试让我的Agent去调用一个内部的数据查询工具时遇到了大麻烦——每个工具都得写一堆胶水代码适配不同的输入输出格式调试起来简直是一场噩梦。直到发现了MCP我才意识到我们缺的从来不是强大的模型或者复杂的逻辑而是一个能让它们和外部世界顺畅对话的“普通话”。简单来说MCPModel Context Protocol你可以把它理解为AI Agent领域的“USB-C接口”标准。在USB-C统一之前你的手机、电脑、充电宝各有各的接口出门得带一堆线。MCP干的就是同样的事它定义了一套统一的协议让任何AI模型比如Claude、GPT都能以一种标准化的方式去发现、调用和管理外部工具、数据源或服务我们称之为“资源”。开发者不再需要为每一个工具单独编写适配器只需要按照MCP标准把工具“包装”成一个MCP Server任何兼容MCP的AI客户端比如Claude Desktop、Cursor就能即插即用。这解决了AI Agent开发中的一个核心痛点上下文构建的标准化与工具使用的碎片化。以前给Agent“扩展能力”是个高度定制化的脏活累活现在有了MCP就像是给Agent世界建立了“应用商店”和统一的安装规范。接下来我们就深入这个协议的内部看看它究竟是如何运作以及我们如何利用它来构建更强大、更易集成的AI智能体。2. MCP协议核心架构与设计哲学拆解要理解MCP不能只看它定义了几个API更要理解它背后的设计哲学。它不是一个重量级的框架而是一个轻量级的、基于JSON-RPC的通信协议。这种选择本身就体现了其核心思想简单、通用、与语言无关。2.1 核心组件Server, Client 与 TransportMCP的架构非常清晰主要包含三个角色MCP Server资源提供方这是能力的封装者。它可以是一个简单的脚本提供天气查询也可以是一个复杂的后端服务提供数据库操作或内部API。它的职责是按照MCP协议向客户端宣告自己提供了哪些“资源”工具或数据源并在客户端请求时执行具体操作。MCP Client资源消费方通常是AI应用本身比如Claude Desktop、Codeium或你自行开发的Agent框架。它的职责是发现可用的Server列出其提供的资源并在需要时发起调用请求。Transport传输层连接Server和Client的通道。MCP设计上支持多种传输方式目前最常见的是stdio标准输入输出和SSEServer-Sent Events。Stdio模式简单直接适合本地集成SSE则更适合网络环境。协议本身不关心传输细节这保证了其灵活性。这种架构的优势在于解耦。作为Server开发者你只需要关心如何实现工具功能并按照协议返回标准化的JSON。作为Client开发者你只需要实现协议解析就能接入无数个Server。这种“插座-插头”模型极大地降低了生态建设的门槛。2.2 协议核心资源Resources与工具ToolsMCP将外部能力抽象为两种核心类型这是理解其能力边界的关键资源Resources可以理解为静态或动态的“数据源”。它有一个唯一的URI如file:///path/to/doc或dynamic://stock/price/AAPL以及对应的文本内容。Client可以“读取”资源。例如一个Server可以提供一个“今日待办事项列表”资源AI在回答相关问题时会先读取这个资源来获取上下文。资源支持内容类型text, image, pdf等未来扩展性很强。工具Tools这就是我们通常理解的“可执行函数”。每个工具需要定义清晰的输入参数JSON Schema格式。当AI决定使用某个工具时Client会调用它并传入参数。Server执行后返回结果。比如“发送邮件”、“查询数据库”、“生成图表”都可以是工具。一个关键设计亮点MCP协议要求Server在初始化时就向Client宣告自己提供的所有资源和工具的“列表”和“描述”。这允许AI在生成回答的初期就充分知晓自己“手头有哪些牌可以打”从而做出更合理的规划和决策而不是盲目地生成文本后再去尝试调用可能不存在的功能。2.3 会话管理与上下文生命周期MCP协议是围绕“会话Session”建立的。一个Client和Server的连接就是一个会话。会话内有明确的生命周期初始化Initialize握手交换能力信息。资源/工具列表ListClient获取Server的能力清单。调用CallClient请求调用工具或读取资源。通知NotificationsServer可以主动推送资源变更如“文件已更新”帮助AI保持上下文新鲜度。结束Shutdown优雅关闭连接。这个生命周期管理使得连接状态清晰可控避免了资源泄漏或状态混乱的问题特别适合需要长期运行、动态感知环境变化的Agent场景。3. 手把手实战从零构建你的第一个MCP Server理论说得再多不如动手做一遍。我们以构建一个最简单的“时间查询服务器”为例展示如何从零实现一个MCP Server。我将使用Python因为它生态友好有现成的SDK。3.1 环境准备与SDK选择首先你需要一个Python环境3.8。官方和社区提供了多种SDK来抽象协议细节让我们专注于业务逻辑。这里我们使用mcp这个Python库它目前是社区最活跃的选择之一。# 创建项目目录并进入 mkdir my-first-mcp-server cd my-first-mcp-server # 创建虚拟环境推荐 python -m venv venv # 激活虚拟环境 # Windows: venv\Scripts\activate # Mac/Linux: source venv/bin/activate # 安装MCP SDK pip install mcp3.2 编写Server核心逻辑创建一个名为time_server.py的文件。我们的目标是提供两个能力1. 一个名为get_current_time的工具返回当前时间2. 一个名为time_guide的资源提供如何使用本服务器的说明。# time_server.py import asyncio from datetime import datetime from mcp import Server, StdioServerParameters from mcp.types import Tool, TextContent, ResourceTemplate # 1. 创建Server实例 server Server(time-server) # 2. 定义一个工具Tool server.list_tools() async def handle_list_tools(): # 返回本Server提供的所有工具描述 return [ Tool( nameget_current_time, description获取当前的系统日期和时间。, inputSchema{ type: object, properties: { format: { type: string, description: 时间格式可选。默认为%Y-%m-%d %H:%M:%S。例如%Y年%m月%d日。, default: %Y-%m-%d %H:%M:%S } } } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): # 根据工具名分发处理逻辑 if name get_current_time: fmt arguments.get(format, %Y-%m-%d %H:%M:%S) current_time datetime.now().strftime(fmt) return [ TextContent( typetext, textf当前时间是{current_time} ) ] else: raise ValueError(f未知工具: {name}) # 3. 定义一个资源Resource server.list_resources() async def handle_list_resources(): # 返回本Server提供的所有资源描述 return [ ResourceTemplate( uritime://guide, name时间服务器使用指南, description关于如何使用本时间查询服务器的说明文档。, mimeTypetext/plain ) ] server.read_resource() async def handle_read_resource(uri: str): # 根据资源URI返回内容 if uri time://guide: guide_text 时间服务器使用指南 本服务器提供以下功能 1. 工具 get_current_time: 获取当前时间。 - 可选参数 format: 指定时间格式字符串例如 %Y年%m月%d日。 2. 资源 time://guide: 本说明文档。 示例请求获取当前年月日。 return TextContent(typetext, textguide_text) else: raise ValueError(f未知资源: {uri}) # 4. 主函数使用Stdio传输启动服务器 async def main(): async with server.run_stdio(StdioServerParameters()): # 保持服务器运行直到被终止 await asyncio.Future() if __name__ __main__: asyncio.run(main())代码解读与注意事项装饰器模式server.list_tools()和server.call_tool()是SDK提供的装饰器用于注册处理函数。这种模式清晰地将“能力声明”和“能力执行”分开。输入模式Schema定义工具时inputSchema至关重要。它用JSON Schema精确描述了AI调用该工具时需要或可以传递的参数。写得好AI调用起来就准确写得模糊就容易出错。建议为每个参数提供详细的description和合理的default值。资源URI资源URI是一个字符串标识符建议使用自定义的Scheme如time://来避免冲突。它不是真实的网络地址只是一个协议内的唯一ID。异步编程MCP SDK基于 asyncio所有处理函数都是async的。确保你的业务逻辑也是非阻塞的如果是耗时操作如网络请求要使用异步库。3.3 运行与测试你的Server保存文件后你可以直接运行它。但由于MCP Client通常通过stdin/stdout与其通信直接运行看不到输出。我们需要一个测试Client。最简单的方法是使用官方提供的mcpCLI 工具需单独安装mcp-cli或使用一个已集成MCP Client的应用。这里我们用一个更直观的方法写一个极简的测试脚本。# test_client.py (仅用于基础测试非标准MCP Client) import subprocess import json import sys # 启动Server进程 proc subprocess.Popen( [sys.executable, time_server.py], stdinsubprocess.PIPE, stdoutsubprocess.PIPE, stderrsubprocess.PIPE, textTrue ) # 模拟发送一个初始化请求简化版 init_request { jsonrpc: 2.0, id: 1, method: initialize, params: { protocolVersion: 0.1.0, capabilities: {}, clientInfo: {name: test-client} } } proc.stdin.write(json.dumps(init_request) \n) proc.stdin.flush() # 读取一行响应 response proc.stdout.readline() print(Server响应:, response) # 结束进程 proc.terminate()更实际的测试是将其配置到真正的MCP Client中如 Claude Desktop。在Claude Desktop中配置MCP Server打开Claude Desktop设置。找到“开发者设置”或“MCP服务器”部分。点击“添加服务器”。配置如下名称My Time Server命令python或你的Python解释器完整路径参数/path/to/your/time_server.py的完整路径保存并重启Claude Desktop。重启后当你与Claude对话时它就能自动识别并使用你编写的get_current_time工具了。你可以尝试说“请告诉我现在的时间用中文格式。”4. 高级应用连接真实世界与复杂工具集成构建一个返回时间的Server只是入门。MCP真正的威力在于连接各种异构系统。下面我们探讨几个更复杂的场景。4.1 集成外部API构建天气查询Server假设我们要集成一个天气API。这里以假想的weatherapi.com为例。# weather_server.py (部分关键代码) import aiohttp # ... 其他导入 ... server Server(weather-server) WEATHER_API_KEY your_api_key_here # 切记不要硬编码在代码中应使用环境变量 server.list_tools() async def handle_list_tools(): return [ Tool( nameget_weather, description根据城市名称查询实时天气情况。, inputSchema{ type: object, properties: { city: { type: string, description: 城市名称例如Beijing, Shanghai, New York }, days: { type: integer, description: 预报天数1-3默认为1, default: 1 } }, required: [city] # 指定必填参数 } ) ] server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name get_weather: city arguments[city] days arguments.get(days, 1) # 使用异步HTTP客户端调用外部API async with aiohttp.ClientSession() as session: url fhttp://api.weatherapi.com/v1/forecast.json params { key: WEATHER_API_KEY, q: city, days: days, aqi: no, alerts: no } async with session.get(url, paramsparams) as resp: if resp.status 200: data await resp.json() current data[current] forecast data[forecast][forecastday][0][day] text f{city}当前天气{current[condition][text]}温度{current[temp_c]}°C体感{current[feelslike_c]}°C湿度{current[humidity]}%。 text f\n今日预报最高{forecast[maxtemp_c]}°C最低{forecast[mintemp_c]}°C。 return [TextContent(typetext, texttext)] else: error_text await resp.text() return [TextContent(typetext, textf查询天气失败{error_text})] # ... 错误处理 ...关键点异步HTTP请求使用aiohttp等异步库避免阻塞Server主线程。错误处理外部API调用可能失败必须将清晰的错误信息返回给AI而不是抛出未处理的异常导致整个会话中断。安全API密钥等敏感信息务必通过环境变量 (os.getenv) 或配置文件读取绝对不要提交到版本库。4.2 包装命令行工具赋予AI系统操作能力让AI安全地执行系统命令是一个强大但危险的能力。MCP可以提供一个受控的接口。# sys_tool_server.py (部分关键代码) import subprocess import shlex from typing import List # 定义一个允许执行的命令白名单 ALLOWED_COMMANDS { list_directory: {cmd: [ls, -la], desc: 列出当前目录的详细内容。}, disk_usage: {cmd: [df, -h], desc: 查看磁盘使用情况。}, process_list: {cmd: [ps, aux], desc: 显示所有运行中的进程。}, } server.list_tools() async def handle_list_tools(): tools [] for name, info in ALLOWED_COMMANDS.items(): tools.append(Tool( namename, descriptioninfo[desc], inputSchema{type: object, properties: {}} # 此例中无参数 )) return tools server.call_tool() async def handle_call_tool(name: str, arguments: dict): if name in ALLOWED_COMMANDS: cmd_list ALLOWED_COMMANDS[name][cmd] try: # 安全地执行白名单内的命令 result subprocess.run( cmd_list, capture_outputTrue, textTrue, timeout10 # 设置超时防止挂起 ) output f命令执行完成。\n返回码{result.returncode}\n输出\n{result.stdout} if result.stderr: output f\n错误输出\n{result.stderr} return [TextContent(typetext, textoutput)] except subprocess.TimeoutExpired: return [TextContent(typetext, text命令执行超时。)] except Exception as e: return [TextContent(typetext, textf执行命令时发生异常{e})] else: raise ValueError(f命令不在白名单内: {name})重要警告提供命令行工具接口风险极高。必须严格遵守以下原则最小权限原则像上面一样严格使用命令白名单绝不接受用户动态输入的命令字符串。沙盒环境考虑在Docker容器或具有严格权限限制的用户环境中运行此类Server。审计日志记录所有工具调用和结果便于追踪和审计。超时控制必须为子进程设置超时防止恶意或错误命令无限运行。4.3 动态资源与通知实现实时数据同步MCP支持Server主动向Client发送通知notifications/resources/updated这对于需要实时更新上下文的场景非常有用。例如一个监控日志文件的Server。# log_monitor_server.py (概念性代码展示通知机制) import asyncio import watchfiles # 需要安装 watchfiles 库 server.list_resources() async def handle_list_resources(): return [ ResourceTemplate( urifile:///var/log/app/current.log, name应用实时日志, description应用程序的最新日志内容。, mimeTypetext/plain ) ] async def watch_log_file(server_instance): log_path /var/log/app/current.log # 使用 watchfiles 监控文件变化 async for changes in watchfiles.awatch(log_path): for change in changes: if change[0] watchfiles.Change.modified: # 文件被修改向所有连接的Client发送资源更新通知 # 注意这需要Server实例能管理多个Client会话标准SDK可能需要扩展或使用特定模式 # 此处为概念展示 # await server_instance.send_notification(resources/updated, { # uri: file:///var/log/app/current.log # }) print(f[通知] 日志文件已更新: {log_path}) # 在实际实现中你需要一个方法来广播通知给相关会话。 # 在main函数中启动监控任务 async def main(): # ... 初始化server ... # 创建后台任务监控文件 asyncio.create_task(watch_log_file(server)) # ... 运行server ...这个例子展示了MCP协议如何支持“主动推送”模式使得AI的上下文可以随着外部数据的变化而自动更新为实现真正动态、感知环境的Agent奠定了基础。5. 生态现状、最佳实践与避坑指南MCP协议虽然理念先进但作为一个新兴标准其生态和实践仍在快速演进中。根据我这段时间的实践分享一些观察和经验。5.1 当前生态与工具链主流Client支持Claude Desktop目前对MCP支持最友好、最成熟的客户端。配置简单体验流畅是学习和测试MCP Server的首选环境。Cursor IDE作为AI原生编辑器也集成了MCP Client可以直接在编码环境中调用MCP工具场景非常契合。其他框架如LangChain、LlamaIndex等Agent框架正在逐步增加对MCP的原生支持或社区插件。Server SDK与工具Python (mcp)社区最活跃文档和示例相对丰富是快速上手的最佳选择。TypeScript/Node.js (modelcontextprotocol/sdk)官方SDK由Anthropic维护类型定义完善适合前端或Node.js生态的开发者。CLI工具 (mcp-cli)用于调试、测试和发现MCP Server的实用命令行工具。社区Server仓库GitHub上已经出现了不少优秀的开源MCP Server项目例如github-search-mcp搜索GitHub仓库。sqlite-mcp操作SQLite数据库。filesystem-mcp安全的文件系统操作。多去这些项目看看是学习如何编写高质量Server的绝佳途径。5.2 开发与部署最佳实践清晰的工具命名与描述工具和资源的name、description是AI理解其功能的唯一途径。务必使用清晰、无歧义的自然语言描述。好的描述应包含目的、输入参数的含义、输出结果的格式。严谨的输入模式Schema定义充分利用JSON Schema的type,enum,pattern,minimum/maximum等属性来约束输入。这能极大减少AI调用时的参数错误。为可选参数设置合理的default值。错误处理的友好性Server端必须做好错误处理。当工具执行失败时返回结构化的错误信息帮助AI理解问题所在例如“数据库连接失败请检查网络”比一个空的错误堆栈更有用。性能与超时工具执行应设置超时。对于可能耗时的操作如网络请求要确保是异步的避免阻塞整个Server影响其他请求。配置化与安全Server的配置如API端点、密钥必须通过环境变量或配置文件管理。在代码中硬编码敏感信息是严重的安全隐患。日志与监控为Server添加适当的日志记录记录工具的调用、参数和结果注意脱敏这对于调试和运维至关重要。5.3 常见问题与排查技巧在实际开发和集成中你肯定会遇到各种问题。下面是一个快速排查清单问题现象可能原因排查步骤Claude Desktop 无法识别 Server1. 配置命令或路径错误。2. Server启动失败或立即崩溃。3. 协议版本不兼容。1. 在终端手动运行配置的命令看Server能否正常启动并等待输入。2. 查看Claude Desktop的日志通常可在设置中找到日志文件路径。3. 确保Server实现了正确的初始化握手。AI 不调用工具1. 工具描述不够清晰AI不理解何时使用。2. 工具列表未正确返回。3. AI模型本身策略限制。1. 优化工具的名称和描述使其意图更明确。2. 使用mcp-cli或写测试脚本连接Server手动发送tools/list请求检查返回是否正确。3. 在对话中明确提示AI可以使用某个工具。工具调用返回错误1. Server端逻辑错误或异常未捕获。2. AI传递的参数不符合Schema。3. 网络或外部服务问题。1. 查看Server的运行日志或标准错误输出stderr。2. 在call_tool处理函数开头打印收到的arguments检查其格式。3. 模拟AI的请求用curl或脚本直接测试Server的接口。连接不稳定或断开1. Server进程崩溃。2. 传输层如stdio缓冲区问题。3. 心跳或保活机制缺失。1. 加强Server的异常处理避免未处理的异常导致进程退出。2. 确保读写stdin/stdout时遵循行分隔的JSON-RPC格式。3. 对于长连接考虑实现更健壮的传输层或使用SSE。一个关键的调试工具mcp-cli。安装后pip install mcp-cli你可以用它来直接与你的Server交互模拟Client的行为这是验证Server是否按协议工作的最直接方法。# 使用 mcp-cli 测试你的 Server mcp run python time_server.py # 连接后可以尝试输入 /list 查看工具和资源/call tool_name 调用工具。MCP协议的出现标志着AI Agent从“单机智能”走向“生态智能”的关键一步。它通过一个简洁的接口标准解决了工具集成的混乱问题。虽然目前生态还在早期但方向已经非常明确。对于开发者而言现在开始学习并构建自己的MCP Server相当于提前布局了未来AI应用生态的“基础设施层”。从简单的工具封装开始逐步尝试连接更复杂的系统你会发现让AI真正成为你的得力助手门槛正在变得越来越低。