实战：用 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 定义了上下文数据的结构化表示，核心包含：

1. Context Frame（上下文帧）：一次交互的完整上下文单元，包含：

   • session_id：会话标识
   • turn_id：轮次编号
   • role：发送者角色（user / system / tool / agent）
   • content：内容（支持文本、结构化数据、工具调用等）
   • metadata：元信息（时间戳、来源、置信度等）
   • tools_called：本轮调用的工具列表
   • state：当前会话状态（如 waiting_tool, thinking, responding）

2. MCP Service：负责接收、路由、转换、持久化上下文帧的服务层。

3. 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 datetime

class RoleEnum(str, Enum):
    USER = "user"
    SYSTEM = "system"
    TOOL = "tool"
    AGENT = "agent"

class ToolCall(BaseModel):
    tool_name: str
    arguments: Dict[str, Any]
    call_id: str

class ContextFrame(BaseModel):
    session_id: str = Field(..., description="会话唯一ID")
    turn_id: int = Field(..., description="对话轮次")
    role: RoleEnum
    content: str
    tools_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 json

class ContextManager:
    def __init__(self, use_redis=False):
        self.sessions: Dict[str, List[ContextFrame]] = {}
        self.use_redis = use_redis
        if use_redis:
            self.redis_client = redis.Redis(host='localhost', port=6379, db=0)

    def add_frame(self, frame: ContextFrame):
        sid = frame.session_id
        if 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 ContextManager

router = 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 router

app = 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 uvicorn
    uvicorn.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 datetime

url = "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_url

    def 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_reply

    def _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)

✅ 五、MCP 的进阶价值

1. 可观测性：所有上下文帧可被记录、查询、分析，便于调试复杂 Agent 行为。
2. 状态恢复：通过 session_id 可恢复中断的对话状态。
3. 多Agent协作：不同 Agent 可订阅同一 session，实现上下文共享。
4. 审计与合规：完整记录对话轨迹，满足企业合规要求。
5. 插件化扩展：可轻松接入 Memory、Planner、Guardrail 等模块。

🔚 结语

MCP 不是银弹，但它为混乱的 LLM 上下文管理提供了一种标准化、可扩展、可观测的解决方案。通过本文的实战，你已掌握：

• MCP 协议的核心结构
• 用 Python + FastAPI 快速搭建 MCP 服务
• 在 Agent 中集成 MCP 实现上下文流转

下一步，你可以：

• 接入 Redis / PostgreSQL 实现持久化
• 增加 WebSocket 支持实现实时推送
• 开发 MCP 浏览器插件可视化上下文流
• 与 LangChain / LlamaIndex 集成

🌟 让 MCP 成为你 AI 工程化架构的“中枢神经系统”，告别上下文混乱，拥抱结构化智能！

—

💬 评论区开放：你在项目中是如何管理 LLM 上下文的？是否遇到过上下文丢失或混乱的问题？欢迎分享！

#MCP #AI架构 #Python实战 #LLMOps #智能体开发