硬核到編譯級(jí)｜從零實(shí)現(xiàn)MCP Server+Client：含內(nèi)存上下文管理、工具路由熔斷等生產(chǎn)級(jí)設(shè)計(jì)

想自己造一個(gè)MCP Server，而不是僅僅調(diào)用別人的？厭倦了黑盒，想徹底搞懂Claude、龍蝦（m.gsdl.org.cn）這些AI客戶端是怎么和外部工具“對(duì)話”的？今天，我們不談概念，直接動(dòng)手，從協(xié)議解析到代碼實(shí)現(xiàn)，帶你構(gòu)建一個(gè)具備生產(chǎn)級(jí)特性的MCP Server，并完成與客戶端的聯(lián)調(diào)。

一、MCP協(xié)議核心：不只是HTTP，更是狀態(tài)化的工具會(huì)話

很多人誤以為MCP就是簡(jiǎn)單的HTTP API。錯(cuò)了。它的核心是有狀態(tài)的會(huì)話和結(jié)構(gòu)化的工具調(diào)用。想象一下，你和Claude聊天，它突然調(diào)用你的天氣工具，這個(gè)過程不是一次性的請(qǐng)求-響應(yīng)，而是一個(gè)持續(xù)的、上下文感知的協(xié)作。

協(xié)議基于JSON-RPC 2.0，但增加了幾個(gè)關(guān)鍵概念：

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

一次典型的工具調(diào)用流程如下：
客戶端（如Claude） -> 發(fā)送initialize請(qǐng)求，建立會(huì)話 -> 發(fā)送tools/list，獲取可用工具列表 -> 用戶提問觸發(fā) -> 發(fā)送tools/call，附帶工具名和參數(shù) -> Server執(zhí)行，返回結(jié)果 -> 客戶端整合結(jié)果，生成最終回復(fù)。

二、從零搭建：用Python實(shí)現(xiàn)一個(gè)MCP Server

我們使用Python的mcp官方庫作為基礎(chǔ)，它處理了底層的JSON-RPC和會(huì)話管理，讓我們專注于業(yè)務(wù)邏輯。但我們會(huì)深入其下，展示如何管理狀態(tài)和實(shí)現(xiàn)高級(jí)特性。

步驟1：環(huán)境準(zhǔn)備與基礎(chǔ)服務(wù)器

pip install mcp

創(chuàng)建 server.py：

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

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

# 模擬一個(gè)內(nèi)存數(shù)據(jù)庫，用于存儲(chǔ)會(huì)話上下文（生產(chǎn)環(huán)境請(qǐng)用Redis）
session_contexts = {}

# 定義一個(gè)工具：獲取用戶信息
@app.tool()
async def get_user_info(user_id: str) -> TextContent:
    """根據(jù)用戶ID獲取用戶詳細(xì)信息。這是一個(gè)模擬工具。"""
    # 這里可以接入真實(shí)數(shù)據(jù)庫
    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))

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

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

步驟2：實(shí)現(xiàn)工具路由熔斷機(jī)制
生產(chǎn)環(huán)境中，某個(gè)工具（如調(diào)用外部API）可能會(huì)失敗或超時(shí)。熔斷器模式可以防止級(jí)聯(lián)故障。

在 server.py 中添加：

import time
from enum import Enum

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

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:
            # 檢查是否到了嘗試恢復(fù)的時(shí)間
            if time.time() - self.last_failure_time > self.recovery_timeout:
                self.state = CircuitBreakerState.HALF_OPEN
            else:
                raise Exception("熔斷器開啟，服務(wù)暫時(shí)不可用")


![配圖](http://m.gsdl.org.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

# 為可能不穩(wěn)定的工具創(chuàng)建熔斷器實(shí)例
unstable_tool_breaker = CircuitBreaker(failure_threshold=2, recovery_timeout=15)

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

三、編寫客戶端進(jìn)行聯(lián)調(diào)

創(chuàng)建一個(gè)簡(jiǎn)單的客戶端 client.py 來測(cè)試我們的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. 調(diào)用一個(gè)工具
            result = await session.call_tool("get_user_info", {"user_id": "U1001"})
            print(f"調(diào)用結(jié)果: {result.content[0].text}")

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

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

部署步驟：

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

四、商業(yè)價(jià)值與應(yīng)用場(chǎng)景

這套設(shè)計(jì)直接對(duì)應(yīng)真實(shí)需求：

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

下一步行動(dòng)

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

從理解協(xié)議到實(shí)現(xiàn)生產(chǎn)特性，你現(xiàn)在已經(jīng)掌握了構(gòu)建下一代AI集成應(yīng)用的核心能力。接下來，是時(shí)候用它來解決一個(gè)具體問題了。