Python + FastAPI 实战：5分钟搞定MCP服务搭建（含Redis配置避坑指南）

Python FastAPI 实战5分钟搞定MCP服务搭建含Redis配置避坑指南在构建智能对话系统时上下文管理往往是最容易被忽视却又最关键的一环。想象一下这样的场景当用户连续询问北京天气如何和那上海呢系统需要准确记住每个问题的上下文关系。这就是MCPModel Context Protocol要解决的核心问题——它像一位专业的会议记录员为每次对话建立结构化档案。1. 环境准备与基础搭建1.1 项目初始化首先创建项目目录结构这就像搭建乐高积木前先分类零件mkdir mcp_service cd mcp_service python -m venv venv source venv/bin/activate # Windows使用 venv\Scripts\activate1.2 依赖安装在requirements.txt中写入fastapi0.95.2 uvicorn0.22.0 pydantic2.0 redis4.5.5 python-dotenv1.0.0执行安装命令时建议添加--no-cache-dir避免缓存问题pip install --no-cache-dir -r requirements.txt2. MCP核心模型设计2.1 上下文帧定义在models/mcp.py中构建对话的基本单元from enum import Enum from pydantic import BaseModel, Field from typing import List, Dict, Any, Optional from datetime import datetime class Role(str, Enum): USER user AGENT agent TOOL tool SYSTEM system class ToolCall(BaseModel): name: str Field(..., min_length1) args: Dict[str, Any] Field(default_factorydict) call_id: str Field(..., regexr^[a-z0-9_]$) class ContextFrame(BaseModel): session_id: str Field(..., min_length3) turn_id: int Field(..., gt0) role: Role content: str tools_used: List[ToolCall] Field(default_factorylist) state: str Field(responding, patternr^\w$) metadata: Dict[str, Any] Field(default_factorydict) timestamp: datetime Field(default_factorydatetime.now)注意Pydantic的Field验证能有效拦截80%的非法输入特别对session_id这类关键字段建议设置min_length约束3. Redis集成实战3.1 连接配置技巧在项目根目录创建.env文件REDIS_HOST127.0.0.1 REDIS_PORT6379 REDIS_DB0 REDIS_PASSWORD REDIS_SSLFalse通过环境变量读取配置更安全# services/redis_manager.py import redis from pydantic import BaseSettings class RedisConfig(BaseSettings): host: str localhost port: int 6379 db: int 0 password: str None ssl: bool False class Config: env_file .env def get_redis(): config RedisConfig() return redis.Redis( hostconfig.host, portconfig.port, dbconfig.db, passwordconfig.password, sslconfig.ssl, decode_responsesFalse # 保持二进制格式避免编码问题 )3.2 常见坑点解决方案连接池耗尽添加连接池配置pool redis.ConnectionPool(max_connections50) redis_client redis.Redis(connection_poolpool)序列化问题推荐使用orjson替代jsonimport orjson def serialize(frame): return orjson.dumps(frame.dict()) def deserialize(data): return ContextFrame(**orjson.loads(data))键名冲突采用命名空间策略def get_redis_key(session_id: str) - str: return fmcp:session:{session_id}4. FastAPI服务封装4.1 路由设计在api/v1/endpoints.py中实现核心APIfrom fastapi import APIRouter, Depends from services.context_manager import get_manager from models.mcp import ContextFrame router APIRouter(prefix/mcp, tags[MCP]) router.post(/frames) async def add_frame( frame: ContextFrame, managerDepends(get_manager) ): try: manager.add_frame(frame) return {status: success} except redis.RedisError as e: raise HTTPException( status_code502, detailfRedis operation failed: {str(e)} )4.2 性能优化技巧启用Gzip压缩from fastapi.middleware.gzip import GZipMiddleware app.add_middleware(GZipMiddleware, minimum_size1024)异步Redis客户端from redis.asyncio import Redis async def get_async_redis(): return await Redis.from_url(redis://localhost)5. 完整部署流程5.1 生产级DockerfileFROM python:3.10-slim WORKDIR /app COPY . . RUN pip install --no-cache-dir -r requirements.txt \ groupadd -r appuser \ useradd -r -g appuser appuser \ chown -R appuser:appuser /app USER appuser ENV PYTHONPATH/app \ PYTHONUNBUFFERED1 \ REDIS_TIMEOUT5 EXPOSE 8000 CMD [uvicorn, main:app, --host, 0.0.0.0, --port, 8000]5.2 健康检查配置在main.py中添加from fastapi import status app.get(/health, status_codestatus.HTTP_204_NO_CONTENT) async def health_check(): if not redis_client.ping(): raise HTTPException( status_code503, detailRedis unavailable )启动服务时推荐使用这些参数uvicorn main:app \ --workers 4 \ --limit-concurrency 100 \ --timeout-keep-alive 30在实际项目中我发现Redis的maxmemory-policy配置对性能影响很大推荐设置为allkeys-lru并预留20%的内存余量。对于高频访问的session数据可以添加本地缓存层减少Redis压力。