引言

当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支持两种传输机制：

传输方式	适用场景	优点	限制
stdio	本地进程间通信，服务器作为子进程启动	简单、无需网络、安全性高	仅限本机调用
HTTP with SSE	远程服务器、多客户端共享	支持远程访问、可扩展	需处理网络、鉴权

1.4 协议生命周期

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

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

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

环境准备：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应用创新的钥匙。

（全文完）