保姆级教程：用Python和uv从零搭建你的第一个MCP服务器（附天气查询实战）
从零构建Python驱动的MCP天气服务：现代AI开发实战指南
1. 环境配置与工具链搭建
在开始构建MCP（Model Context Protocol）服务之前，我们需要配置高效的开发环境。Python生态中新兴的uv工具链将成为我们的得力助手。
1.1 UV工具链详解
UV是新一代Python依赖管理工具，相比传统pip具有显著优势：
# 安装uv（两种方式任选） curl -LsSf https://astral.sh/uv/install.sh | sh # 或 pip install uv关键特性对比：
| 特性 | UV | 传统pip |
|---|---|---|
| 安装速度 | 快5-10倍 | 较慢 |
| 依赖解析 | 增量更新 | 全量解析 |
| 虚拟环境管理 | 内置轻量方案 | 需venv辅助 |
| 跨平台支持 | 全平台一致体验 | Windows兼容性问题 |
1.2 项目初始化实战
创建项目目录并初始化虚拟环境：
# 创建项目目录 mkdir mcp-weather && cd mcp-weather # 初始化UV虚拟环境 uv venv .venv # 激活环境（Linux/macOS） source .venv/bin/activate # Windows使用 .venv\Scripts\activate提示：UV会自动识别项目目录结构，无需手动指定venv路径
2. MCP核心架构解析
2.1 MCP三层通信模型
现代AI服务架构通常包含以下组件：
- 客户端(Client)：处理用户交互
- 协议层(Protocol)：标准化通信
- 服务端(Server)：提供具体能力
graph LR A[用户设备] -->|请求| B(Client) B -->|MCP协议| C[Server] C -->|OpenWeather API| D[第三方服务]2.2 协议设计要点
MCP通信需要关注三个核心维度：
- 传输层：HTTP/SSE/WebSocket选择
- 数据格式：JSON-RPC 2.0规范
- 错误处理：标准化错误码体系
典型请求示例：
{ "jsonrpc": "2.0", "method": "get_weather", "params": {"city": "Beijing"}, "id": 1 }3. 天气服务实现
3.1 服务端核心代码
创建weather_server.py文件：
import httpx from mcp.server import FastMCP app = FastMCP("WeatherService") @app.tool() async def get_weather(city: str) -> dict: """获取指定城市天气数据""" async with httpx.AsyncClient() as client: resp = await client.get( "https://api.openweathermap.org/data/2.5/weather", params={ "q": city, "appid": "YOUR_API_KEY", "units": "metric", "lang": "zh_cn" } ) return resp.json() if __name__ == "__main__": app.run(transport="stdio")关键组件说明：
FastMCP：快速构建MCP服务的装饰器@app.tool()：声明可调用工具方法transport="stdio"：使用标准输入输出通信
3.2 客户端实现
创建client.py：
import asyncio from mcp.client import ClientSession async def main(): async with ClientSession() as session: # 调用天气服务 result = await session.call_tool( "get_weather", {"city": "Beijing"} ) print(f"当前天气：{result}") asyncio.run(main())4. 高级功能扩展
4.1 多模型集成方案
通过MCP可以轻松集成不同AI模型：
# 模型路由配置 MODEL_MAP = { "gpt-4": OpenAIClient, "claude": AnthropicClient, "local": LocalLLMClient } async def route_request(model_type, query): client = MODEL_MAP[model_type]() return await client.process(query)4.2 性能优化技巧
- 连接池管理：
from httpx import AsyncClient async with AsyncClient(timeout=30.0) as client: # 复用连接- 结果缓存：
from cachetools import TTLCache weather_cache = TTLCache(maxsize=100, ttl=3600)- 异步批处理：
async def batch_requests(cities): tasks = [get_weather(city) for city in cities] return await asyncio.gather(*tasks)5. 部署与监控
5.1 生产环境部署
推荐使用Docker容器化部署：
FROM python:3.10-slim WORKDIR /app COPY . . RUN pip install uv && \ uv pip install -r requirements.txt CMD ["uvicorn", "weather_server:app", "--host", "0.0.0.0"]5.2 监控指标配置
Prometheus监控示例：
from prometheus_client import start_http_server, Counter REQUEST_COUNT = Counter( 'weather_requests_total', 'Total weather API requests' ) @app.tool() async def get_weather(city: str): REQUEST_COUNT.inc() # ...典型监控维度：
- 请求成功率
- 响应时间P99
- 并发连接数
- 缓存命中率
6. 安全最佳实践
- 认证鉴权：
from fastapi.security import APIKeyHeader api_key_header = APIKeyHeader(name="X-API-KEY") async def verify_key(key: str = Depends(api_key_header)): if key != os.getenv("API_KEY"): raise HTTPException(status_code=403)- 输入验证：
from pydantic import BaseModel, constr class WeatherRequest(BaseModel): city: constr(min_length=2, max_length=50)- 限流保护：
from slowapi import Limiter from slowapi.util import get_remote_address limiter = Limiter(key_func=get_remote_address) app.state.limiter = limiter @app.tool() @limiter.limit("10/minute") async def get_weather(city: str): # ...在实际项目中，我发现最容易被忽视的是连接超时设置。特别是在调用第三方天气API时，合理的超时配置（建议15-30秒）能显著提高系统稳定性。另一个实用技巧是使用UV的--resolution=lowest-direct参数，可以避免依赖冲突问题。