保姆级教程:用Python和uv从零搭建你的第一个MCP服务器(附天气查询实战)

张开发
2026/6/6 8:48:00 15 分钟阅读
保姆级教程:用Python和uv从零搭建你的第一个MCP服务器(附天气查询实战)
从零构建Python驱动的MCP天气服务现代AI开发实战指南1. 环境配置与工具链搭建在开始构建MCPModel 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(transportstdio)关键组件说明FastMCP快速构建MCP服务的装饰器app.tool()声明可调用工具方法transportstdio使用标准输入输出通信3.2 客户端实现创建client.pyimport 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(timeout30.0) as client: # 复用连接结果缓存from cachetools import TTLCache weather_cache TTLCache(maxsize100, ttl3600)异步批处理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(nameX-API-KEY) async def verify_key(key: str Depends(api_key_header)): if key ! os.getenv(API_KEY): raise HTTPException(status_code403)输入验证from pydantic import BaseModel, constr class WeatherRequest(BaseModel): city: constr(min_length2, max_length50)限流保护from slowapi import Limiter from slowapi.util import get_remote_address limiter Limiter(key_funcget_remote_address) app.state.limiter limiter app.tool() limiter.limit(10/minute) async def get_weather(city: str): # ...在实际项目中我发现最容易被忽视的是连接超时设置。特别是在调用第三方天气API时合理的超时配置建议15-30秒能显著提高系统稳定性。另一个实用技巧是使用UV的--resolutionlowest-direct参数可以避免依赖冲突问题。

更多文章