从零构建MCP Server:Model Context Protocol开发完全指南
引言
随着大语言模型(LLM)能力的飞速发展,如何让模型安全、高效地连接外部世界——比如数据库、API、文件系统——成为 AI 应用落地的关键瓶颈。Anthropic 推出的Model Context Protocol (MCP)正是为了解决这一难题而生的开放标准。它将 LLM 与外部工具、数据源的集成规范化,让开发者可以用一种统一的方式“教会”模型使用各种外部能力。
本文将带你从零开始,掌握 MCP Server 的开发流程:从核心概念梳理,到编写第一个完整可运行的服务器,再到配置客户端进行真实交互,最后分享生产环境中的常见问题与最佳实践。读完这篇文章,你将具备独立开发并部署 MCP Server 的能力。
MCP 核心概念
在动手编写代码之前,我们需要理解 MCP 的架构与关键原语。
1. 客户端-服务器架构
MCP 采用经典的 C/S 模型:
- MCP Host:发起连接的一方,通常是 Claude Desktop、IDE 插件或你自己的 AI 应用;
- MCP Client:运行在 Host 内部,负责与 Server 维持一对一连接;
- MCP Server:你将要开发的部分,提供具体的工具、资源和提示等能力。
一个典型的交互流程如下:
- Host 启动内嵌的 Client,Client 与 Server 建立连接(如 stdio 或 HTTP);
- Client 请求 Server 的能力列表(工具、资源……);
- 用户通过 Host 与模型对话,模型根据需求调用 Server 提供的工具;
- Server 执行工具并返回结果,模型汇总后回应用户。
2. 三大核心原语
MCP 定义了三种服务器可暴露的能力:
- 工具(Tools):类似 API 端点,模型可以调用它们来执行操作,例如查询天气、发送邮件、读写文件。每个工具由名称、描述和输入模式(JSON Schema)构成;
- 资源(Resources):暴露只读数据,如文件内容或数据库记录。资源用 URI 标识,支持订阅和内容变更通知;
- 提示(Prompts):预定义的对话模板,可帮助用户快速开启特定任务。
本教程将重点放在工具的构建上,因为它是绝大多数开发场景的核心。
3. 传输机制
MCP 支持两种主流传输方式:
- stdio:通过标准输入输出通信,简单直接,特别适合本地进程间调用(如 Claude Desktop 集成);
- Server-Sent Events (SSE):基于 HTTP 的流式传输,允许远程连接,适用于将 Server 部署为服务供多个客户端使用。
我们将首先使用stdio进行演示,它无需任何网络配置,极易上手。
实战示例:构建一个天气查询 MCP Server
下面我们使用 Python 和最新的mcp库,一步步实现一个能够查询天气的工具服务器,并配置到 Claude Desktop 中测试。
1. 环境准备
确保你已安装 Python 3.10+,然后创建项目目录并安装依赖:
mkdir mcp-weather-server && cd mcp-weather-server python -m venv venv source venv/bin/activate # Windows: venv\Scripts\activate pip install mcpmcp库由 Anthropic 官方维护,提供了编写 Server 和 Client 的全套基础设施。
2. 编写服务器代码
创建weather_server.py,填入以下完整代码(包含详细注释):
#!/usr/bin/env python3 """ 一个简单的 MCP Server:提供天气查询工具。 使用 stdio 传输,可直接接入 Claude Desktop。 """ import asyncio import json from typing import Any from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import Tool, TextContent # 1. 创建 Server 实例,名称用于标识 app = Server("weather-server") # 2. 声明工具列表 —— 告诉客户端我们能做什么 @app.list_tools() async def list_tools() -> list[Tool]: """返回当前服务器支持的所有工具定义""" return [ Tool( name="get_weather", description="获取指定城市的当前天气信息", inputSchema={ "type": "object", "properties": { "city": { "type": "string", "description": "城市名称,例如 Beijing" }, "unit": { "type": "string", "enum": ["celsius", "fahrenheit"], "description": "温度单位,默认为摄氏度" } }, "required": ["city"] } ) ] # 3. 处理工具调用 —— 真正执行逻辑的地方 @app.call_tool() async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]: """根据工具名称和参数执行操作,返回结果""" if name == "get_weather": city = arguments.get("city", "Unknown") unit = arguments.get("unit", "celsius") # 模拟天气数据(实际应用中可替换为真实 API 调用) weather = { "Beijing": {"temp": 22, "condition": "晴"}, "Shanghai": {"temp": 28, "condition": "多云"}, "Guangzhou": {"temp": 30, "condition": "雷阵雨"}, } info = weather.get(city, {"temp": 20, "condition": "未知"}) temp = info["temp"] condition = info["condition"] if unit == "fahrenheit": temp = temp * 9 / 5 + 32 result_text = f"{city}当前天气:{condition},温度{temp:.1f}{'℉' if unit == 'fahrenheit' else '℃'}" return [TextContent(type="text", text=result_text)] # 未知工具应抛出明确错误 raise ValueError(f"未知工具: {name}") # 4. 主函数,启动 stdio 传输并运行服务器 async def main(): """使用 stdio 协议启动服务器""" async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() ) if __name__ == "__main__": asyncio.run(main())3. 运行与测试
直接用 Python 运行即可启动一个 stdio MCP Server:
python weather_server.py此时服务器会阻塞等待标准输入,这符合 stdio 传输的特性。为了验证工具是否工作正常,我们可以使用 MCP 自带的开发者工具mcp dev(需额外安装mcp[cli]):
pip install 'mcp[cli]' mcp dev weather_server.py该命令会启动一个本地 web 界面,让你可视化地列出工具、发起调用并查看结果。推荐在开发阶段使用。
4. 接入 Claude Desktop
最实用的场景是将 MCP Server 配置到 Claude Desktop 中,让模型直接调用你的工具。
打开 Claude Desktop 的配置文件:
- macOS:~/Library/Application Support/Claude/claude_desktop_config.json
- Windows:%APPDATA%\Claude\claude_desktop_config.json编辑文件,添加
mcpServers段:
{ "mcpServers": { "weather": { "command": "python", "args": ["/你的绝对路径/mcp-weather-server/weather_server.py"] } } }- 保存并重启 Claude Desktop,在对话中输入:“帮我查一下北京今天天气怎么样?”。如果一切正常,Claude 会自动调用
get_weather工具并展示返回结果。
如果日志发现连接错误,请检查 Python 路径和文件权限。
进阶:添加资源与错误处理
暴露静态资源
除了工具,你也可以公开只读资源。例如,为上面的天气服务器添加一个“城市列表”资源:
from mcp.types import Resource, ResourceContents @app.list_resources() async def list_resources() -> list[Resource]: return [ Resource( uri="weather://cities.json", name="支持的城市列表", description="当前系统支持查询天气的城市", mimeType="application/json" ) ] @app.read_resource() async def read_resource(uri: str) -> ResourceContents: if uri == "weather://cities.json": cities = ["Beijing", "Shanghai", "Guangzhou"] return ResourceContents( uri=uri, mimeType="application/json", text=json.dumps(cities, ensure_ascii=False) ) raise ValueError("资源不存在")这样客户端就可以通过读取资源获取城市列表,用于下拉选择等 UI 组件。
健壮的错误处理
生产环境中,务必捕获异常并返回结构化错误,而非让服务器崩溃:
@app.call_tool() async def call_tool(name: str, arguments: dict) -> list[TextContent]: try: if name == "get_weather": city = arguments.get("city") if not city: raise ValueError("缺少参数 'city'") # ... 业务逻辑 except Exception as e: # 返回错误消息,模型会据此理解并做出相应反馈 return [TextContent(type="text", text=f"工具调用失败:{str(e)}")]模型接收到错误字符串后,通常会向用户解释问题或调整参数重试。
常见问题与注意事项
1. stdio vs SSE,如何选择?
- stdio:最简单,无需网络配置,适合本地工具和 Claude Desktop 集成。缺点是只能被一个客户端连接;
- SSE:支持远程访问和多个客户端,需要实现一个 HTTP 服务器并处理跨域。当你的服务需要被多个用户或远程 AI 应用调用时,使用 SSE 是更好的选择。
从 stdio 切换到 SSE,只需修改主函数:
import uvicorn from mcp.server.sse import sse_server if __name__ == "__main__": uvicorn.run(sse_server(app), host="0.0.0.0", port=8000)然后客户端连接http://your-server:8000/sse。
2. 安全考量
- 输入校验:永远不要信任模型传来的参数,必须进行类型检查和范围限制,防止注入攻击;
- 权限控制:文件系统工具应限制可读写的目录范围;网络请求应做域名白名单;
- 敏感信息:不要在工具描述或返回值中硬编码 API 密钥等敏感数据,应从环境变量读取。
3. 性能与资源管理
当工具涉及耗时操作(如调用外部 API)时,务必使用异步 I/O(async/await),避免阻塞事件循环。对于计算密集型任务,考虑放在线程池或子进程中执行。
4. 版本兼容与更新
MCP 协议仍在快速演进,你需要留意mcp库的更新日志。升级前在测试环境验证,避免破坏现有集成。
总结
本文从 MCP 的基本架构讲起,通过一个完整的天气查询服务器实例,演示了如何声明工具、处理调用,并将服务接入 Claude Desktop。我们进一步探讨了资源暴露、错误处理、传输选择以及安全实践。
MCP 的价值在于标准化——它让 LLM 与外部世界的交互不再是千篇一律的硬编码插件,而是可复用、可组合的模块。掌握了 MCP Server 的开发,你就拥有了一把钥匙,可以安全、优雅地为 AI 应用打开通往无限数据与功能的大门。
未来,随着生态的完善,MCP 有望成为 AI 工具集成的事实标准。现在开始动手实践,你将站在这股浪潮的最前沿。