MCP (Model Context Protocol) 技术理解 - 第五篇

引言

在前面的两篇文章中，我们已经讲了MCP服务端和客户端的核心原语，我相信大家已经对MCP已经有比较深入的了解了，那么我们今天就一起进入MCP server端的实战开发。

提示：为了大家都能看懂，这里的实战开发仅仅作为示例，开发比较简单的MCP server，同时鉴于MCP对Python SDK有比较好的支持，在下面的代码我们全部使用Python来写。

如果你对前面的内容感兴趣，可以点击这里跳转

MCP (Model Context Protocol) 技术理解 - 第一篇

MCP (Model Context Protocol) 技术理解 - 第二篇

MCP (Model Context Protocol) 技术理解 - 第三篇

MCP (Model Context Protocol) 技术理解 - 第四篇

MCP server 整体架构解析

我们来做一个能实时天气预报的MCP server，这个MCP server实现难度不大，非常适合教学。下面我们一起看看架构吧。

我们使用 FastMCP 框架实现。它通过标准输入/输出 (STDIO) 与客户端通信,提供天气预报和警报功能。

客户端 (Claude Desktop/IDE) 
↕ (STDIO) 
MCP Server (Python) 
↕ (HTTPS) 
NWS API

项目的依赖管理

核心依赖:

• mcp[cli]: MCP 协议框架
• httpx: 异步 HTTP 客户端
• 数据源: 美国国家气象局 (NWS) API

[project]
name = "weather-server"
version = "0.1.0"
description = "MCP Weather Server"
requires-python = ">=3.10"
dependencies = [
    "mcp[cli]>=1.2.0",
    "httpx>=0.27.0",
]

[project.scripts]
weather-server = "weather.server:main"

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

server.py

以下是MCP server核心代码解析

服务器初始化

我们当然先要创建MCP实例出来，然后为其命名为weather,设置数据格式为json

• 创建 FastMCP 实例,服务器名称为 "weather"

• json_response=True 表示返回 JSON 格式响应

mcp = FastMCP("weather", json_response=True)

辅助函数

make_nws_request(url) - HTTP 请求封装

• 封装对 NWS API 的异步请求
• 设置必需的 User-Agent 和 Accept 头
• 统一错误处理

async def make_nws_request(url: str) -> dict[str, Any] | None:
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json",
    }
    async with httpx.AsyncClient() as client:
        response = await client.get(url, headers=headers, timeout=30.0)
        return response.json()

format_alert(feature) - 警报格式化,可以将 API 返回的原始警报数据格式化为易读文本

def format_alert(feature: dict) -> str:
    """格式化警报信息"""
    props = feature["properties"]
    return f"""
Event: {props.get("event", "Unknown")}
Area: {props.get("areaDesc", "Unknown")}
Severity: {props.get("severity", "Unknown")}
Description: {props.get("description", "No description available")}
Instructions: {props.get("instruction", "No specific instructions provided")}
"""

MCP核心功能

Tool

Tool 1: get_alerts(state)，这里的功能是: 根据州代码(如 CA, NY)获取当前活跃的天气警报

调用流程:

1. 请求 /alerts/active/area/{state}
2. 检查返回的 features 列表
3. 格式化所有警报并用 --- 分隔返回

@mcp.tool()
async def get_alerts(state: str) -> str:
    """
    获取美国州的天气警报
    
    Args:
        state: 两位州代码 (例如: CA, NY)
    """
    url = f"{NWS_API_BASE}/alerts/active/area/{state}"
    data = await make_nws_request(url)
    
    if not data or "features" not in data:
        return "Unable to fetch alerts or no alerts found."
    
    if not data["features"]:
        return "No active alerts for this state."
    
    alerts = [format_alert(feature) for feature in data["features"]]
    return "n---n".join(alerts)

Tool 2: get_forecast(latitude, longitude),这里的功能是: 根据经纬度获取天气预报

两步调用:

• 第一步: 请求 /points/{lat},{lon} 获取该位置的预报网格端点

• 第二步: 使用返回的 forecast URL 获取详细预报

• 只返回前 5 个时段的预报

@mcp.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取位置的天气预报"""

Resources

提供服务器的元信息和能力说明，Resources使用 URI 方案 weather://config 标识资源

@mcp.resource("weather://config")
def get_config() -> str:
    """获取服务配置信息"""
    return """
Weather Server Configuration:
- API: National Weather Service (NWS)
- Coverage: United States only
- Update Frequency: Real-time
- Supported Operations:
  * Weather Alerts by State
  * Weather Forecast by Location
"""

Prompts

Prompts为 LLM 生成结构化的提示词模板,指导 LLM 如何组织天气信息,帮助 LLM 构造请求

@mcp.prompt()
def weather_brief(location: str) -> str:
    """
    生成天气简报提示
    
    Args:
        location: 位置描述
    """
    return f"""
Please provide a weather brief for {location}. Include:
1. Current weather conditions
2. Today's forecast
3. Any active weather alerts
4. 3-day outlook

Format the information in a clear, concise manner suitable for quick reading.
"""

运行入口

这里就是我们MCP server的启动入口了，我们使用stdio标准通信

def main(): 
    mcp.run(transport="stdio") # 通过标准输入/输出通信 
    
if __name__ == "__main__": 
    main()

场景演示

这里我们来一个场景演示一下如何运行这个MCP服务器吧

当 Claude Desktop 连接到这个服务器后:

用户: "加州有什么天气警报吗?" → Claude 调用 get_alerts("CA")

用户: "旧金山的天气预报" → Claude 先获取旧金山坐标,再调用 get_forecast(37.77, -122.41)

用户: "给我写个西雅图的天气简报" → Claude 使用 weather_brief("Seattle") 提示模板

写MCP server时可能会踩到的坑

STDIO 传输的陷阱

MCP 通过 stdin/stdout 进行 JSON-RPC 通信,任何额外输出都会破坏协议格式

# 错误:在 STDIO 模式下使用 print 调试
print("Debug info")  # 会污染 STDIO 通道,导致协议解析失败

# 正确:使用 stderr 输出日志
import sys
print("Debug info", file=sys.stderr)

异步问题

如果我们在异步函数里面调用同步库，就有可能迟迟没有结果导致最终返回超时

# 错误:在异步函数中使用同步库
@mcp.tool()
async def get_data():
    response = requests.get(url)  # 阻塞整个事件循环!
    
# 正确:使用异步 HTTP 客户端
@mcp.tool()
async def get_data():
    async with httpx.AsyncClient() as client:
        response = await client.get(url)

如果未正确处理异步上下文，可能会导致资源泄漏

# 错误:客户端未关闭
client = httpx.AsyncClient()
response = await client.get(url)  # 资源泄漏

# 正确:使用上下文管理器
async with httpx.AsyncClient() as client:
    response = await client.get(url)

Tool问题

我们在定义Tool的时候，如果没有明确参数类型标注缺失，就有可能导致返回的数据类型不一致

# 错误:缺少类型提示
@mcp.tool()
async def get_alerts(state):  # LLM 不知道参数类型
    pass

# 正确:明确类型和文档
@mcp.tool()
async def get_alerts(state: str) -> str:
    """
    获取天气警报
    
    Args:
        state: 两位州代码 (例如: CA, NY)
    """
    pass

返回值格式不一致

# 错误:有时返回 dict,有时返回 str
@mcp.tool()
async def get_data(query: str):
    if condition:
        return {"result": "success"}
    return "Error occurred"

# 正确:统一返回字符串
@mcp.tool()
async def get_data(query: str) -> str:
    if condition:
        return json.dumps({"result": "success"})
    return "Error occurred"

Resources

在定义Resources的时候，如果URL格式不正确，也会导致出现报错

# 错误:使用文件路径风格
@mcp.resource("/config.json")  # 不符合 URI 规范

# 正确:使用 scheme://path 格式
@mcp.resource("weather://config")
@mcp.resource("weather://alerts/active")

还有很多的坑，但在这里就不一一列举了，需要大家自己去实践才能切身体会到。

总结

在这个简单的MCP server示例中，我们展示了 MCP 的三大核心能力:

1. Tools: 让 LLM 执行动态操作
2. Resources: 提供静态上下文
3. Prompts: 引导 LLM 生成更好的请求

在下一篇，我们可以讲一下MCP客户端的配置和MCP的一些高级特性