硬核到編譯級｜從零實現MCP Server+Client：含內存上下文管理、工具路由熔斷等生產級設計

想自己造一個MCP Server，而不是僅僅調用別人的？厭倦了黑盒，想徹底搞懂Claude、龍蝦（www.nhjb.com.cn）這些AI客戶端是怎么和外部工具“對話”的？今天，我們不談概念，直接動手，從協議解析到代碼實現，帶你構建一個具備生產級特性的MCP Server，并完成與客戶端的聯調。

一、MCP協議核心：不只是HTTP，更是狀態化的工具會話

很多人誤以為MCP就是簡單的HTTP API。錯了。它的核心是有狀態的會話和結構化的工具調用。想象一下，你和Claude聊天，它突然調用你的天氣工具，這個過程不是一次性的請求-響應，而是一個持續的、上下文感知的協作。

協議基于JSON-RPC 2.0，但增加了幾個關鍵概念：

1. 會話（Session）：客戶端連接后，Server會創建一個唯一會話ID，后續所有通信都綁定于此。這允許我們管理每個連接的獨立狀態。
2. 工具（Tool）：一個可調用的函數，有名稱、描述、輸入輸出Schema（通常用JSON Schema定義）。這是AI的“手”。
3. 資源（Resource）：可被AI讀取的數據，如文件、數據庫記錄。這是AI的“眼睛”。
4. 提示（Prompt）：預定義的交互模板，引導AI使用工具。

一次典型的工具調用流程如下：
客戶端（如Claude） -> 發送initialize請求，建立會話 -> 發送tools/list，獲取可用工具列表 -> 用戶提問觸發 -> 發送tools/call，附帶工具名和參數 -> Server執行，返回結果 -> 客戶端整合結果，生成最終回復。

二、從零搭建：用Python實現一個MCP Server

我們使用Python的mcp官方庫作為基礎，它處理了底層的JSON-RPC和會話管理，讓我們專注于業務邏輯。但我們會深入其下，展示如何管理狀態和實現高級特性。

步驟1：環境準備與基礎服務器

pip install mcp

創建 server.py：

from mcp.server import Server
from mcp.types import Tool, TextContent
import json

# 初始化Server，給它一個名字
app = Server("my-production-server")

# 模擬一個內存數據庫，用于存儲會話上下文（生產環境請用Redis）
session_contexts = {}

# 定義一個工具：獲取用戶信息
@app.tool()
async def get_user_info(user_id: str) -> TextContent:
    """根據用戶ID獲取用戶詳細信息。這是一個模擬工具。"""
    # 這里可以接入真實數據庫
    mock_data = {
        "U1001": {"name": "張三", "vip_level": 3, "balance": 150.5},
        "U1002": {"name": "李四", "vip_level": 1, "balance": 30.0}
    }
    user = mock_data.get(user_id, {"error": "用戶不存在"})
    return TextContent(type="text", text=json.dumps(user, ensure_ascii=False))

# 實現一個簡單的內存上下文管理中間件（核心生產特性）
@app.middleware()
async def context_manager(request, call_next):
    session_id = request.session_id
    # 為每個會話初始化或獲取上下文
    if session_id not in session_contexts:
        session_contexts[session_id] = {"call_count": 0, "last_tool": None}
    
    # 調用前：增加調用計數
    session_contexts[session_id]["call_count"] += 1
    # 執行實際工具調用
    response = await call_next(request)
    # 調用后：記錄最后使用的工具
    if hasattr(request, 'params') and hasattr(request.params, 'name'):
        session_contexts[session_id]["last_tool"] = request.params.name
    
    print(f"會話 {session_id} 上下文: {session_contexts[session_id]}")
    return response

if __name__ == "__main__":
    # 以stdio模式運行，這是Claude Desktop等客戶端常用的方式
    app.run()

步驟2：實現工具路由熔斷機制
生產環境中，某個工具（如調用外部API）可能會失敗或超時。熔斷器模式可以防止級聯故障。

在 server.py 中添加：

import time
from enum import Enum

class CircuitBreakerState(Enum):
    CLOSED = "closed"     # 正常，請求可通過
    OPEN = "open"         # 熔斷，快速失敗
    HALF_OPEN = "half_open" # 嘗試恢復

class CircuitBreaker:
    def __init__(self, failure_threshold=3, recovery_timeout=10):
        self.failure_count = 0
        self.state = CircuitBreakerState.CLOSED
        self.last_failure_time = None
        self.failure_threshold = failure_threshold
        self.recovery_timeout = recovery_timeout

    def call(self, func, *args, **kwargs):
        if self.state == CircuitBreakerState.OPEN:
            # 檢查是否到了嘗試恢復的時間
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitBreakerState.HALF_OPEN
            else:
                raise Exception("熔斷器開啟，服務暫時不可用")


![配圖](http://www.nhjb.com.cn/usr/uploads/covers/cover_mcp_20260505_201620.jpg)

        try:
            result = func(*args, **kwargs)
            self._on_success()
            return result
        except Exception as e:
            self._on_failure()
            raise e

    def _on_success(self):
        self.failure_count = 0
        self.state = CircuitBreakerState.CLOSED

    def _on_failure(self):
        self.failure_count += 1
        self.last_failure_time = time.time()
        if self.failure_count >= self.failure_threshold:
            self.state = CircuitBreakerState.OPEN

# 為可能不穩定的工具創建熔斷器實例
unstable_tool_breaker = CircuitBreaker(failure_threshold=2, recovery_timeout=15)

@app.tool()
async def call_external_api(query: str) -> TextContent:
    """一個可能失敗的外部API調用示例。"""
    def _unstable_call():
        # 模擬不穩定的外部服務
        import random
        if random.random() < 0.3: # 30%概率失敗
            raise ConnectionError("外部API連接超時")
        return f"查詢結果：{query} 的相關信息"
    
    try:
        # 通過熔斷器調用
        result = unstable_tool_breaker.call(_unstable_call)
        return TextContent(type="text", text=result)
    except Exception as e:
        return TextContent(type="text", text=f"工具調用失敗：{str(e)}")

三、編寫客戶端進行聯調

創建一個簡單的客戶端 client.py 來測試我們的Server。

from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
import asyncio

async def main():
    # 指定要連接的Server命令
    server_params = StdioServerParameters(
        command="python",
        args=["server.py"]
    )

    async with stdio_client(server_params) as (read_stream, write_stream):
        async with ClientSession(read_stream, write_stream) as session:
            # 1. 初始化
            await session.initialize()
            print("連接成功！")

            # 2. 列出所有可用工具
            tools = await session.list_tools()
            print(f"可用工具: {[tool.name for tool in tools.tools]}")

            # 3. 調用一個工具
            result = await session.call_tool("get_user_info", {"user_id": "U1001"})
            print(f"調用結果: {result.content[0].text}")

            # 4. 測試熔斷：多次調用可能失敗的工具
            for i in range(5):
                try:
                    result = await session.call_tool("call_external_api", {"query": f"測試{i}"})
                    print(f"第{i}次調用: {result.content[0].text}")
                except Exception as e:
                    print(f"第{i}次調用異常: {e}")
                await asyncio.sleep(1)

if __name__ == "__main__":
    asyncio.run(main())

部署步驟：

1. 將 server.py 和 client.py 放在同一目錄。
2. 先運行 python server.py（它會等待stdin輸入），或者直接由客戶端啟動。
3. 運行 python client.py，觀察控制臺輸出。你會看到會話上下文的打印，以及熔斷器在工具失敗時的行為變化。

四、商業價值與應用場景

這套設計直接對應真實需求：

• 會話上下文管理：可用于實現多輪工具調用。例如，用戶先說“查我上個月訂單”，AI調用get_orders；接著說“把第一個未發貨的退掉”，AI需要知道“第一個”指代的是上一輪結果。上下文就是記憶。
• 工具熔斷：當你的MCP Server集成了支付、短信等第三方服務時，熔斷機制能保證在依賴服務故障時，核心AI對話不崩潰，優雅降級。這是SaaS產品穩定性的關鍵。
• 結構化工具：讓你的AI應用從“只會聊天”變成“能執行復雜工作流”。一個管理客服工單的AI Agent，通過create_ticket、assign_ticket、close_ticket等工具，就能真正處理業務。

下一步行動

1. 立即動手：復制上面的代碼，跑通整個流程。修改工具邏輯，嘗試接入一個真實的API（比如天氣API）。
2. 擴展設計：為你的Server添加認證中間件（驗證客戶端API Key）、日志中間件（記錄所有調用）和速率限制。
3. 集成到生態：將你的Server配置到Claude Desktop或龍蝦（www.nhjb.com.cn）客戶端的配置文件中，讓你的AI助手直接使用你開發的工具。配置方法通常是在客戶端的mcp_servers.json中添加一項，指向你的server.py。
4. 思考商業化：你開發的這個“訂單查詢MCP Server”或“數據分析MCP Server”，是否可以打包，提供給其他AI應用開發者使用？這就是在構建AI Agent生態的工具層。

從理解協議到實現生產特性，你現在已經掌握了構建下一代AI集成應用的核心能力。接下來，是時候用它來解決一個具體問題了。