用 FastAPI 零配置构建 SSE MCP 服务器

现在可以用支持 SSE 的 MCP 客户端连接 http://localhost:8001/mcp，然后就能看到 /hello 自动变成了有一个 name 参数的工具。

实际场景：天气查询工具

上面的例子太简单，我们做一个正经的：通过美国国家气象局公开 API 查天气预报，并把结果格式化返回。

from fastapi import FastAPI
from fastapi_mcp import add_mcp_server
from typing import Any
import httpx

NWS_API_BASE = "https://api.weather.gov"
USER_AGENT = "weather-app/1.0"

app = FastAPI()
mcp_server = add_mcp_server(
    app,
    mount_path="/mcp",
    name="Weather MCP Server",
    describe_all_responses=True,
    describe_full_response_schema=True
)

async def make_nws_request(url: str) -> dict[str, Any] | None:
    """向 NWS API 发起请求，并进行错误处理。"""
    headers = {
        "User-Agent": USER_AGENT,
        "Accept": "application/geo+json"
    }
    async with httpx.AsyncClient() as client:
        try:
            response = await client.get(url, headers=headers, timeout=30.0)
            response.raise_for_status()
            return response.json()
        except Exception:
            return None

@mcp_server.tool()
async def get_forecast(latitude: float, longitude: float) -> str:
    """获取地点的天气预报。
    参数:
        latitude: 地点的纬度
        longitude: 地点的经度
    """
    points_url = f"{NWS_API_BASE}/points/{latitude},{longitude}"
    points_data = await make_nws_request(points_url)
    if not points_data:
        return "Unable to fetch forecast data for this location."
    
    forecast_url = points_data["properties"]["forecast"]
    forecast_data = await make_nws_request(forecast_url)
    if not forecast_data:
        return "Unable to fetch detailed forecast."
    
    periods = forecast_data["properties"]["periods"]
    forecasts = []
    for period in periods[:5]:
        forecast = f"""
        {period['name']}: Temperature: {period['temperature']}°{period['temperatureUnit']} Wind: {period['windSpeed']} {period['windDirection']} Forecast: {period['detailedForecast']} 
        """
        forecasts.append(forecast)
    return "\n---\n".join(forecasts)

这个代码几乎就是原生的 FastAPI + 异步请求，唯一 MCP 相关的只有两处：add_mcp_server 和 @mcp_server.tool() 装饰器。后者可以让你显式声明哪些函数作为 MCP 工具，参数和文档也会自动暴露给客户端。

启动服务后，可以先用 curl 访问一下 /docs，你会发现 OpenAPI 文档里仍然正常，不影响原来的使用。MCP 端点挂在 /mcp 下面，互不干扰。

调试与连接

启动后，最方便的是用官方的 inspector 工具进行调试：

CLIENT_PORT=8081 SERVER_PORT=8082 npx -y @modelcontextprotocol/inspector

在浏览器打开 http://localhost:8081，选择 SSE 传输，输入你的服务器地址（比如 http://localhost:8001/mcp），就能看到自动生成的工具并测试。

关于 SSE：SSE 是单向的，服务器只能向客户端推送事件。MCP 协议用它来实现服务端到客户端的消息推送，但客户端发请求还是走普通的 HTTP POST。所以严格来说，这是 POST + SSE 混合模式，并不是 WebSocket 那样的双向通道。不用把 SSE 想得太复杂，对于 AI 调用工具这种请求-响应为主的场景，够用了。

如果你的 MCP 客户端只支持 stdio 模式（比如 Claude Desktop），可以用 mcp-proxy 做个转发：

安装：

uv tool install mcp-proxy

然后在 Claude Desktop 的配置文件中加上：

{
  "mcpServers": {
    "weather-api-mcp-proxy": {
      "command": "mcp-proxy",
      "args": ["http://127.0.0.1:8001/mcp"]
    }
  }
}

这样 Claude Desktop 就能通过 stdio 跟你的 SSE MCP 服务器通信了。

一些观察

fastapi-mcp 让 MCP 的接入成本变得很低，对于已有 FastAPI 项目的团队来说，几乎可以无缝集成。它自动生成的工具 schema 基于类型注解和文档字符串，质量还不错。但缺点也是有的：目前不支持流式响应、参数校验全靠 FastAPI 自身、缺乏对复杂依赖注入的支持。如果是生产环境，可能还需要自己在外面包一层控制逻辑。

另外，这种自动暴露所有端点的做法，安全上要小心——最好只暴露指定的路由，或者用 @mcp_server.tool() 手工标记需要的函数，像示例那样。

对于原型验证或内部工具集成，fastapi-mcp 足够轻快；要上生产，还得再打磨。