实战：用 Python 搭建 MCP 服务 —— 模型上下文协议（Model Context Protocol）应用指南

📌 实战：用 Python 搭建 MCP 服务 —— 模型上下文协议（Model Context Protocol）应用指南

标签：#MCP #AI工程化 #Python #LLM上下文管理 #Agent架构
在这里插入图片描述

🎯 引言：为什么需要 MCP？

在构建大语言模型（LLM）驱动的智能体（Agent）系统时，一个核心挑战是：如何在多轮对话、工具调用、多Agent协作中，高效、结构化地传递和管理上下文？

传统方式（如字符串拼接、JSON随意传递）往往导致：

上下文结构混乱，难以调试
Agent 间协作信息丢失
工具调用参数与模型输入耦合严重
多轮对话状态难以追踪

为解决这些问题，业界提出了 MCP（Model Context Protocol）—— 模型上下文协议。它是一种标准化的上下文数据交换格式与通信协议，旨在统一 Agent、Tool、Memory、Planner 等模块之间的“语言”。

本文将手把手教你用 Python 实现一个简易但完整的 MCP 服务，让你的 LLM 应用从此告别“上下文泥潭”。

🧩 一、MCP 协议核心概念

MCP 定义了上下文数据的结构化表示，核心包含：

Context Frame（上下文帧）：一次交互的完整上下文单元，包含：
- session_id：会话标识
- turn_id：轮次编号
- role：发送者角色（user / system / tool / agent）
- content：内容（支持文本、结构化数据、工具调用等）
- metadata：元信息（时间戳、来源、置信度等）
- tools_called：本轮调用的工具列表
- state：当前会话状态（如 waiting_tool, thinking, responding）
MCP Service：负责接收、路由、转换、持久化上下文帧的服务层。
MCP Client：Agent 或 Tool 通过 Client 发送/接收上下文帧。

🚀 二、Python 实战：搭建 MCP 服务

我们使用 FastAPI + Pydantic + Redis（可选）构建一个轻量级 MCP 服务。

📁 项目结构：

mcp_service/
├── models/
│   └── mcp.py          # MCP 数据模型
├── services/
│   └── context_manager.py  # 上下文管理逻辑
├── api/
│   └── routes.py       # API 路由
├── main.py             # 启动入口
└── requirements.txt

🔧 Step 1：定义 MCP 数据模型（models/mcp.py）

# models/mcp.py
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
from enum import Enum
from datetime import datetimeclass RoleEnum(str, Enum):USER = "user"SYSTEM = "system"TOOL = "tool"AGENT = "agent"class ToolCall(BaseModel):tool_name: strarguments: Dict[str, Any]call_id: strclass ContextFrame(BaseModel):session_id: str = Field(..., description="会话唯一ID")turn_id: int = Field(..., description="对话轮次")role: RoleEnumcontent: strtools_called: List[ToolCall] = Field(default_factory=list)state: str = Field(default="responding")metadata: Dict[str, Any] = Field(default_factory=dict)timestamp: datetime = Field(default_factory=datetime.utcnow)class Config:json_encoders = {datetime: lambda v: v.isoformat()}

🧠 Step 2：实现上下文管理器（services/context_manager.py）

# services/context_manager.py
from typing import List, Dict
from .models.mcp import ContextFrame
import redis
import jsonclass ContextManager:def __init__(self, use_redis=False):self.sessions: Dict[str, List[ContextFrame]] = {}self.use_redis = use_redisif use_redis:self.redis_client = redis.Redis(host='localhost', port=6379, db=0)def add_frame(self, frame: ContextFrame):sid = frame.session_idif sid not in self.sessions:self.sessions[sid] = []self.sessions[sid].append(frame)if self.use_redis:key = f"mcp:session:{sid}"self.redis_client.rpush(key, frame.json())self.redis_client.expire(key, 3600)  # 1小时过期def get_session_frames(self, session_id: str) -> List[ContextFrame]:if self.use_redis:key = f"mcp:session:{session_id}"frames_data = self.redis_client.lrange(key, 0, -1)return [ContextFrame.parse_raw(f) for f in frames_data]else:return self.sessions.get(session_id, [])def get_latest_frame(self, session_id: str) -> Optional[ContextFrame]:frames = self.get_session_frames(session_id)return frames[-1] if frames else None

🌐 Step 3：构建 FastAPI 路由（api/routes.py）

# api/routes.py
from fastapi import APIRouter, HTTPException
from ..models.mcp import ContextFrame
from ..services.context_manager import ContextManagerrouter = APIRouter()
context_manager = ContextManager(use_redis=False)  # 可配置为 True@router.post("/mcp/push", summary="推送上下文帧")
async def push_context_frame(frame: ContextFrame):try:context_manager.add_frame(frame)return {"status": "success", "message": "Frame added"}except Exception as e:raise HTTPException(status_code=500, detail=str(e))@router.get("/mcp/session/{session_id}", summary="获取会话所有帧")
async def get_session_frames(session_id: str):frames = context_manager.get_session_frames(session_id)return {"session_id": session_id, "frames": frames}@router.get("/mcp/session/{session_id}/latest", summary="获取最新帧")
async def get_latest_frame(session_id: str):frame = context_manager.get_latest_frame(session_id)if not frame:raise HTTPException(status_code=404, detail="Session or frame not found")return frame

⚡ Step 4：启动服务（main.py）

# main.py
from fastapi import FastAPI
from api.routes import routerapp = FastAPI(title="MCP Service", version="0.1.0")app.include_router(router, prefix="/api/v1")@app.get("/")
def read_root():return {"message": "Welcome to MCP Service - Model Context Protocol"}if __name__ == "__main__":import uvicornuvicorn.run(app, host="0.0.0.0", port=8000)

📦 Step 5：安装依赖（requirements.txt）

fastapi
uvicorn
pydantic
redis  # 可选
python-dotenv  # 可选

启动服务：

pip install -r requirements.txt
python main.py

访问 http://localhost:8000/docs 查看自动生成的 API 文档！

🧪 三、客户端调用示例

你可以用任何 HTTP 客户端调用 MCP 服务。以下是 Python 示例：

# client_example.py
import requests
import json
from datetime import datetimeurl = "http://localhost:8000/api/v1/mcp/push"frame_data = {"session_id": "chat_123","turn_id": 1,"role": "user","content": "今天北京天气如何？","tools_called": [],"state": "waiting_tool","metadata": {"source": "web", "user_id": "u_001"}
}response = requests.post(url, json=frame_data)
print(response.json())# 获取最新帧
resp = requests.get("http://localhost:8000/api/v1/mcp/session/chat_123/latest")
print(resp.json())

🧠 四、在 Agent 系统中集成 MCP

假设你有一个天气查询 Agent，它可以这样使用 MCP：

class WeatherAgent:def __init__(self, mcp_url="http://localhost:8000/api/v1"):self.mcp_url = mcp_urldef process(self, user_input: str, session_id: str, turn_id: int):# 1. 接收用户输入 → 推送 MCP 帧self._push_frame(session_id, turn_id, "user", user_input)# 2. 决定调用工具tool_call = {"tool_name": "get_weather","arguments": {"location": "北京"},"call_id": "tc_001"}# 3. 推送 Agent 思考帧self._push_frame(session_id, turn_id + 1, "agent","正在调用天气API...",tools_called=[tool_call],state="calling_tool")# 4. 模拟工具返回weather_result = "北京，晴，25°C"self._push_frame(session_id, turn_id + 2, "tool",weather_result,metadata={"tool_name": "get_weather", "call_id": "tc_001"})# 5. Agent 生成最终回复final_reply = f"为您查询到：{weather_result}"self._push_frame(session_id, turn_id + 3, "agent",final_reply,state="responding")return final_replydef _push_frame(self, session_id, turn_id, role, content, **kwargs):data = {"session_id": session_id,"turn_id": turn_id,"role": role,"content": content,**kwargs}requests.post(f"{self.mcp_url}/mcp/push", json=data)