MCP协议深度解析：从原理到实战，打造你的第一个AI工具集成

引言

当ChatGPT等大模型惊艳世界之后，开发者面临的下一个难题是：如何让AI真正触及企业数据、调用外部工具、记住会话上下文？传统的做法是各自编写复杂的插件系统或自定义HTTP API，导致兼容性差、重复劳动严重。2024年11月，Anthropic发布了模型上下文协议（Model Context Protocol，MCP），旨在为AI应用与外部数据源、工具之间建立统一的“万能插口”。

MCP将AI模型与外部世界的关系标准化为客户端-服务器架构：模型所在的宿主应用（如Claude Desktop、自研IDE）作为MCP客户端，而提供数据或功能的进程作为MCP服务器。通过这种协议，你的模型可以安全、可控地访问文件系统、数据库、API甚至其他AI服务。

本文将从原理出发，带你亲手构建一个可运行的MCP天气查询服务器，并通过客户端完成工具调用，让你在最短时间内掌握MCP的实战精髓。

一、MCP核心概念

1.1 架构三要素

• MCP Host：真正运行AI模型的应用，如Claude Desktop、VS Code插件、自研ChatBot。
• MCP Client：嵌入在Host中的协议客户端，负责与服务器建立连接、发送请求、接收响应。
• MCP Server：轻量级服务程序，通过标准输入输出（stdio）或HTTP+SSE暴露能力，例如查询数据库、读取文件、调用外部API。

┌───────────┐ ┌───────────┐ ┌───────────┐ │ 主机应用 │ ←→ │ MCP客户端 │ ←→ │ MCP服务器 │ │ (AI模型) │ │ (协议实现) │ │ (数据/工具) │ └───────────┘ └───────────┘ └───────────┘

1.2 三大原语：Resources、Tools、Prompts

MCP服务器可以向客户端声明自己提供的三种能力：

• Resources（资源）：类似REST的GET端点，暴露只读数据，如文件内容、数据库记录。客户端可以订阅资源的更新通知。
• Tools（工具）：可执行的操作，类似RPC。AI模型可以决定调用哪个工具并传入参数，工具执行后返回结果，模型再基于结果继续生成回答。这是实现函数调用（Function Calling）的标准化方式。
• Prompts（提示模板）：提供预定义的对话模板，方便用户快速发起特定类型的任务。

1.3 传输层选择

MCP支持两种传输机制：

1.4 协议生命周期

一次典型的MCP交互分为三个阶段：
1.初始化：客户端发送initialize请求，双方协商协议版本和能力。
2.就绪：客户端发送initialized通知，双方开始正常交互。
3.关闭：连接关闭，释放资源。

二、实战：构建一个天气查询MCP服务器

我们将用Python实现一个MCP服务器，提供一个get_weather工具，返回指定城市的实时天气。客户端可以是任何支持MCP的应用，这里我们会用官方的mcpPython包快速搭建。

环境准备：Python ≥ 3.10，安装依赖：

pip install "mcp[cli]>1.0.0"

2.1 编写MCP服务器

创建weather_server.py：

#!/usr/bin/env python3 """ 天气查询 MCP 服务器 提供 get_weather 工具，返回模拟的天气数据 """ import json import asyncio from mcp.server import Server, NotificationOptions from mcp.server.models import InitializationCapabilities from mcp.server.stdio import stdio_server # 创建一个MCP服务器实例 server = Server("weather-server") # 模拟天气数据库 WEATHER_DATA = { "北京": {"temperature": 22, "humidity": 45, "condition": "晴"}, "上海": {"temperature": 28, "humidity": 70, "condition": "多云"}, "广州": {"temperature": 31, "humidity": 80, "condition": "阵雨"}, } @server.list_tools() async def list_tools() -> list: """声明服务器提供的工具列表""" return [ { "name": "get_weather", "description": "获取指定城市的天气信息", "inputSchema": { "type": "object", "properties": { "city": { "type": "string", "description": "城市名称，如：北京、上海、广州" } }, "required": ["city"] } } ] @server.call_tool() async def call_tool(name: str, arguments: dict) -> list: """处理工具调用请求""" if name != "get_weather": raise ValueError(f"未知工具: {name}") city = arguments.get("city", "") weather = WEATHER_DATA.get(city) if weather is None: return [{"type": "text", "text": f"抱歉，暂无 {city} 的天气数据"}] # 构造返回结果，使用MCP内容块 result_text = ( f"城市: {city}\n" f"温度: {weather['temperature']}°C\n" f"湿度: {weather['humidity']}%\n" f"天气: {weather['condition']}" ) return [{"type": "text", "text": result_text}] async def main(): """通过标准输入输出启动服务器""" # 配置服务器能力 capabilities = InitializationCapabilities( sampling=None, # 本例不支持采样 ) # 使用stdio传输运行服务器 async with stdio_server() as (read_stream, write_stream): await server.run( read_stream, write_stream, server.create_initialization_options(capabilities), ) if __name__ == "__main__": asyncio.run(main())

代码解读：
-Server("weather-server")初始化MCP服务器实例。
-list_tools()装饰器注册工具列表，返回包含工具名、描述、输入参数的JSON Schema。
-call_tool()装饰器实现具体的工具逻辑，接收name和arguments，返回文本内容块。
-main()函数通过stdio_server()创建标准输入输出传输通道，并启动服务器运行循环。

2.2 测试服务器（MCP Inspector）

MCP官方提供了一个图形化的调试工具——MCP Inspector（需要Node.js环境）。先全局安装：

npx @modelcontextprotocol/inspector

然后在Inspector界面中，选择Transport Type为“Standard Input/Output”，Command为python weather_server.py，点击Connect。连接成功后，你会在Tools标签页看到get_weather工具，输入{"city":"北京"}即可看到返回结果。

2.3 编写MCP客户端调用示例

如果你希望在自研的应用中集成MCP客户端，可以这样编写client.py：

import asyncio from mcp.client.stdio import stdio_client from mcp.client.session import ClientSession async def main(): # 启动天气服务器作为子进程 command = ["python", "weather_server.py"] async with stdio_client(command) as (read, write): async with ClientSession(read, write) as session: # 初始化连接 await session.initialize() # 列出所有可用工具 tools_result = await session.list_tools() print("可用工具:", [tool.name for tool in tools_result.tools]) # 调用 get_weather 工具 result = await session.call_tool( "get_weather", arguments={"city": "北京"} ) # 提取文本内容 if result.content and len(result.content) > 0: print("查询结果:\n", result.content[0].text) if __name__ == "__main__": asyncio.run(main())

运行效果：

可用工具: ['get_weather'] 查询结果: 城市: 北京 温度: 22°C 湿度: 45% 天气: 晴

2.4 扩展：提供Resource和Prompt

为了让服务器更丰满，可以增加一个资源，暴露服务器状态，以及一个提示模板。

在weather_server.py中追加：

@server.list_resources() async def list_resources() -> list: return [ { "uri": "weather://status", "name": "服务器状态", "description": "返回天气服务器的运行状态", "mimeType": "text/plain" } ] @server.read_resource() async def read_resource(uri: str) -> str: if uri == "weather://status": return "天气服务器运行正常，支持城市: 北京、上海、广州" else: raise ValueError(f"未知资源: {uri}") @server.list_prompts() async def list_prompts() -> list: return [ { "name": "weather_prompt", "description": "生成天气查询的对话提示", "arguments": [ {"name": "city", "description": "要查询的城市", "required": True} ] } ] @server.get_prompt() async def get_prompt(name: str, arguments: dict) -> list: if name == "weather_prompt": city = arguments.get("city", "未知城市") return [ { "role": "user", "content": f"请查询{city}的天气，并给出穿衣建议。" } ]

客户端可以相应调用list_resources()、read_resource("weather://status")和get_prompt()等接口来使用这些能力。

三、常见问题与注意事项

3.1 传输层选择建议

• 本地单用户：使用stdio，无需网络配置，安全简单。
• 多人共享/远程服务：使用HTTP+SSE，但需考虑鉴权（可结合OAuth2）、TLS加密。

3.2 安全性

• stdio模式下，服务器以子进程运行，权限范围由父进程控制，较为安全。
• 对于HTTP模式，务必启用认证，避免任意客户端调用敏感工具。
• 工具实现中应做好参数验证，防止注入攻击。

3.3 性能与并发

MCP服务器默认处理请求是串行的。如果工具涉及I/O密集操作（如网络请求），建议使用异步编程模型，避免阻塞事件循环。在高并发远程场景下，可以部署多个服务器实例配合负载均衡。

3.4 版本兼容性

MCP协议仍在快速演进，编写代码时请固定mcp包的版本，并查阅官方规范更新。生产环境应做好版本协商。

3.5 调试技巧

• 使用MCP Inspector进行可视化调试。
• 设置环境变量MCP_LOG_LEVEL=debug可以输出详细的通信日志。
• 可使用ngrok等工具将本地HTTP服务器暴露到公网进行远程测试。

四、总结

通过本文，我们从MCP的设计理念出发，深入理解了客户端-服务器架构、三大原语以及传输机制。而后通过一个完整的天气查询MCP服务器实例，展示了如何定义工具、处理调用，并使用客户端完成端到端测试。我们还演示了如何添加资源和提示模板，让你的服务器更具实用价值。

MCP的出现，标志着AI工具生态向标准化迈出了关键一步。不管是为Claude Desktop编写插件，还是在自己的应用中集成外部数据，MCP都能大幅降低集成成本，让AI“手脚”更加强大。建议你从今天开始，将现有的API或工具用MCP包装起来，享受一键接入AI的乐趣。

未来，随着MCP协议愈发成熟，我们可能会看到更多的工具市场、服务编排模式诞生。掌握MCP，就是掌握了下一波AI应用创新的钥匙。

（全文完）