實戰：用 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)