通過一個DEMO理解MCP（模型上下文協議）的生命周期

freeflydom

2025年4月18日 10:45 本文熱度 146

在LLM應用的快速發展中，一個核心挑戰始終存在：如何讓模型獲取最新、最準確的外部知識并有效利用工具？

背景其實很簡單：大模型（LLM）再強，也總有不知道的東西，怎么辦？讓它“查資料”“調工具”成了近兩年最熱的技術方向。從最早的 RAG（Retrieval-Augmented Generation），到 OpenAI 引領的 Function Call，再到現在 Anthropic 拋出的 MCP（Model Context Protocol），每一代方案都在試圖解答一個問題：模型如何以更自然的方式獲得外部世界的幫助？

MCP 主打的是統一標準和跨模型兼容性。雖然協議本身尚處于早期階段，設計也遠稱不上完美，但出現的時機十分巧妙。。就像當年 OpenAI 的 API，一旦形成事實標準，后面哪怕有點毛病，也可以很快改進，畢竟生態具有滾雪球效應，一旦用戶基數形成規模，自然而然就成為事實標準。

本篇文章將結合 MCP 官方 SDK，通過代碼和流程圖模擬一次帶 Tool 調用的完整交互過程，了解并看清 MCP 的全生命周期。

整體流程

一次MCP完整的調用流程如下：

圖1. 一次包含MCP調用的完整流程

圖1省略了第一步與第二步之間，list_tools()或resource()的步驟，也就是最開始MCP Host知道有哪些可用的工具與資源，我們在本 DEMO 中使用了硬編碼的方式將資源信息構建在提示詞中。

這里需要注意的是MCP Client與MCP Host（主機）并不是分離的部分，但為了時序圖清晰，這里將其邏輯上拆分為不同的部分，實際上MCP Host可以理解為我們需要嵌入AI的應用程序，例如 CRM 系統或 SaaS 服務，實際上Host中是包含MCP Client的代碼。實際的 MCP Host 與 Client 結構如下圖所示：

整體示例代碼

MCP Server

mcp server的代碼使用最簡單的方式啟動，并通過Python裝飾器注冊最簡單的兩個工具，為了DEMO簡單，hard code兩個工具（函數）返回值，代碼如下：

#mcp_server_demo.py
from mcp.server.fastmcp import FastMCP
import asyncio
mcp = FastMCP(name="weather-demo", host="0.0.0.0", port=1234)
@mcp.tool(name="get_weather", description="獲取指定城市的天氣信息")
async def get_weather(city: str) -> str:
    """
    獲取指定城市的天氣信息
    """
    weather_data = {
        "北京": "北京：晴，25°C",
        "上海": "上海：多云，27°C"
    }
    return weather_data.get(city, f"{city}：天氣信息未知")
@mcp.tool(name="suggest_activity", description="根據天氣描述推薦適合的活動")
async def suggest_activity(condition: str) -> str:
    """
    根據天氣描述推薦適合的活動
    """
    if "晴" in condition:
        return "天氣晴朗，推薦你去戶外散步或運動。"
    elif "多云" in condition:
        return "多云天氣適合逛公園或咖啡館。"
    elif "雨" in condition:
        return "下雨了，建議你在家閱讀或看電影。"
    else:
        return "建議進行室內活動。"
async def main():
    print("? 啟動 MCP Server: http://127.0.0.1:1234")
    await mcp.run_sse_async()
if __name__ == "__main__":
    asyncio.run(main())

大模型調用代碼

大模型調用選擇使用openrouter這個LLM的聚合網站，主要是因為該網站方便調用與測試不同的模型，同時網絡環境可以直接連接而不用其他手段。

代碼如下：

# llm_router.py
import json
import requests
# OpenRouter 配置
OPENROUTER_API_KEY = '這里寫入使用的Key'
OPENROUTER_API_URL = "https://openrouter.ai/api/v1/chat/completions"
OPENROUTER_HEADERS = {
    "Authorization": f"Bearer {OPENROUTER_API_KEY}",
    "Content-Type": "application/json",
    "HTTP-Referer": "http://localhost",
    "X-Title": "MCP Demo Server"
}
class OpenRouterLLM:
    """
    自定義 LLM 類，使用 OpenRouter API 來生成回復
    """
    def __init__(self, model: str = LLM_MODEL):
        self.model = model
    def generate(self, messages):
        """
        發送對話消息給 OpenRouter API 并返回 LLM 的回復文本
        參數:
            messages: 一個 list，每個元素都是形如 {'role': role, 'content': content} 的字典
        返回:
            LLM 返回的回復文本
        """
        request_body = {
            "model": self.model,
            "messages": messages
        }
        print(f"發送請求到 OpenRouter: {json.dumps(request_body, ensure_ascii=False)}")
        response = requests.post(
            OPENROUTER_API_URL,
            headers=OPENROUTER_HEADERS,
            json=request_body
        )
        if response.status_code != 200:
            print(f"OpenRouter API 錯誤: {response.status_code}")
            print(f"錯誤詳情: {response.text}")
            raise Exception(f"OpenRouter API 返回錯誤: {response.status_code}")
        response_json = response.json()
        print(f"OpenRouter API 響應: {json.dumps(response_json, ensure_ascii=False)}")
        # 提取 LLM 響應文本
        try:
            content = response_json['choices'][0]['message']['content']
            return content
        except KeyError:
            raise Exception("無法從 OpenRouter 響應中提取內容")
# 如果需要獨立測試該模塊，可以在此進行簡單的測試
if __name__ == "__main__":
    # 示例系統提示和用戶輸入
    messages = [
        {"role": "system", "content": "你是一個智能助手，可以幫助查詢天氣信息。"},
        {"role": "user", "content": "請告訴我北京今天的天氣情況。"}
    ]
    llm = OpenRouterLLM()
    try:
        result = llm.generate(messages)
        print("LLM 返回結果:")
        print(result)
    except Exception as e:
        print(f"調用 OpenRouter 時發生異常: {e}")

MCP Client

這里的MCP Client，使用Server-Side Event（SSE）方式進行連接（題外話，MCP協議使用SSE協議作為默認遠程協議稍微有點奇怪，聽說后續迭代會考慮HTTP Streaming以及JSONRPC over HTTP2的方式）。

這里我們在main測試代碼中，嘗試列出所有可用的Tool與Resource，并嘗試調用Tool，結果如圖，可以看到能夠展示出MCP Server中定義的Tool。

# mcp_client_demo.py
import asyncio
from mcp.client.session import ClientSession
from mcp.client.sse import sse_client
class WeatherMCPClient:
    def __init__(self, server_url="http://127.0.0.1:1234/sse"):
        self.server_url = server_url
        self._sse_context = None
        self._session = None
    async def __aenter__(self):
        # 創建 SSE 通道
        self._sse_context = sse_client(self.server_url)
        self.read, self.write = await self._sse_context.__aenter__()
        # 創建 MCP 會話
        self._session = ClientSession(self.read, self.write)
        await self._session.__aenter__()
        await self._session.initialize()
        return self
    async def __aexit__(self, exc_type, exc_val, exc_tb):
        if self._session:
            await self._session.__aexit__(exc_type, exc_val, exc_tb)
        if self._sse_context:
            await self._sse_context.__aexit__(exc_type, exc_val, exc_tb)
    async def list_tools(self):
        return await self._session.list_tools()
    async def list_resources(self):
        return await self._session.list_resources()
    async def call_tool(self, name, arguments):
        return await self._session.call_tool(name, arguments)
async def main():
    async with WeatherMCPClient() as client:
        print("? 成功連接 MCP Server")
        tools = await client.list_tools()
        print("\n?? 可用工具:")
        print(tools)
        resources = await client.list_resources()
        print("\n?? 可用資源:")
        print(resources)
        print("\n?? 調用 WeatherTool 工具(city=北京)...")
        result = await client.call_tool("get_weather", {"city": "北京"})
        print("\n?? 工具返回:")
        for item in result.content:
            print(" -", item.text)
if __name__ == "__main__":
    asyncio.run(main())

MCP Host

MCP host的角色也就是我們需要嵌入AI的應用，可以是一個程序，可以是一個CRM系統，可以是一個OA，MCP Host包含MCP Client，用于集成LLM與Tool，MCP Host之外+Tool+大模型，共同構成了一套基于AI的系統，現在流行的說法是AI Agent（中文翻譯：AI智能體？）

MCP Host代碼中步驟注釋，與圖1中的整體MCP流程對齊。

import asyncio
import json
import re
from llm_router import OpenRouterLLM
from mcp_client_demo import WeatherMCPClient
def extract_json_from_reply(reply: str):
    """
    提取 LLM 返回的 JSON 內容，自動處理 markdown 包裹、多余引號、嵌套等。
    支持 string 或 dict 格式。
    如果無法解出 dict，則返回原始 string。
    """
    # 如果已經是 dict，直接返回
    if isinstance(reply, dict):
        return reply
    # 清除 markdown ```json ``` 包裹
    if isinstance(reply, str):
        reply = re.sub(r"^```(?:json)?|```$", "", reply.strip(), flags=re.IGNORECASE).strip()
    # 最多嘗試 3 層 json.loads 解碼
    for _ in range(3):
        try:
            parsed = json.loads(reply)
            if isinstance(parsed, dict):
                return parsed
            else:
                reply = parsed  # 如果解出來還是 str，繼續下一層
        except Exception:
            break
    # 如果最終不是 dict，返回原始字符串（表示是普通答復）
    return reply
llm = OpenRouterLLM()
async def main():
    # === 初始化 MCP 客戶端 ===
    client = WeatherMCPClient()
    await client.__aenter__()
    tools = await client.list_tools()
    resources = await client.list_resources()
    tool_names = [t.name for t in tools.tools]
    tool_descriptions = "\n".join(f"- {t.name}: {t.description}" for t in tools.tools)
    resource_descriptions = "\n".join(f"- {r.uri}" for r in resources.resources)
    while True:
        # === Step 1. 用戶 → MCP主機：提出問題 ===
        user_input = input("\n請輸入你的問題（輸入 exit 退出）：\n> ")
        if user_input.lower() in ("exit", "退出"):
            break
        # 構造系統提示 + 工具說明
        system_prompt = (
            "你是一個智能助手，擁有以下工具和資源可以調用：\n\n"
            f"?? 工具列表：\n{tool_descriptions or '（無）'}\n\n"
            f"?? 資源列表：\n{resource_descriptions or '（無）'}\n\n"
            "請優先調用可用的Tool或Resource，而不是llm內部生成。僅根據上下文調用工具，不傳入不需要的參數進行調用\n"
            "如果需要，請以 JSON 返回 tool_calls，格式如下：\n"
            '{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}\n'
            "如無需調用工具，返回：{\"tool_calls\": null}"
        )
        # === 構造 LLM 上下文消息 ===
        messages = [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": user_input}
        ]
        final_reply = ""
        # === 循環處理 tool_calls，直到 LLM 給出最終 content 為止 ===
        while True:
            # === Step 2. MCP主機 → LLM：轉發上下文 ===
            reply = llm.generate(messages)
            print(f"\n?? LLM 回復：\n{reply}")
            # === Step 3. 解析 JSON 格式回復（或普通字符串） ===
            parsed = extract_json_from_reply(reply)
            # === 如果是普通自然語言字符串，說明 LLM 已直接答復用戶 ===
            if isinstance(parsed, str):
                final_reply = parsed
                break
            # === 如果是字典，判斷是否包含工具調用 ===
            tool_calls = parsed.get("tool_calls")
            if not tool_calls:
                # LLM 給出普通答復結構（帶 content 字段）
                final_reply = parsed.get("content", "")
                break
            # === 遍歷 LLM 請求的工具調用列表 ===
            for tool_call in tool_calls:
                # === Step 4. LLM → MCP客戶端：請求使用工具 ===
                tool_name = tool_call["name"]
                arguments = tool_call["arguments"]
                if tool_name not in tool_names:
                    raise ValueError(f"? 工具 {tool_name} 未注冊")
                # === Step 5. MCP客戶端 → MCP服務器：調用工具 ===
                print(f"?? 調用工具 {tool_name} 參數: {arguments}")
                result = await client.call_tool(tool_name, arguments)
                # === Step 8. MCP服務器 → MCP客戶端：返回結果 ===
                tool_output = result.content[0].text
                print(f"?? 工具 {tool_name} 返回：{tool_output}")
                # === Step 9. MCP客戶端 → LLM：提供工具結果 ===
                messages.append({
                    "role": "tool",
                    "name": tool_name,
                    "content": tool_output
                })
            # Step 10: 再次調用 LLM，進入下一輪（可能再次產生 tool_calls）
        # === Step 11. MCP主機 → 用戶：最終結果答復 ===
        print(f"\n?? 最終答復：{final_reply}")
    await client.__aexit__(None, None, None)
if __name__ == "__main__":
    asyncio.run(main())

用戶提問

DEMO的交互方式是一個簡單的Chatbox。假設用戶在聊天界面的輸入框里敲下：“上海的天氣如何” 。此時，用戶的問題通過 MCP 主機（MCP Host）被發送給大模型。

MCP Host 可以是一個瀏覽器前端、桌面應用，也可以只是后端的一段代碼。在這個場景里，它主要負責收集用戶輸入并與LLM通信。

對應流程圖1中的步驟1：提出問題與步驟2：轉發問題。

LLM 推理：是否需要外部Tool配合

收到用戶提問后，MCP 主機（Host）負責將用戶提問解析并附加上下文后轉發給大模型。主要取決于系統設計的智能程度、工具豐富度，以及 LLM 的能力邊界。通常可以是一段靜態的提示詞，或者從上下文中獲取動態的提示詞，也可以是通過一些外部API獲取數據生成提示詞，這并不是本文的重點，本文通過簡單的靜態提示詞進行。

本DEMO的靜態提示詞如下：

        # 構造系統提示 + 工具說明
        system_prompt = (
            "你是一個智能助手，擁有以下工具和資源可以調用：\n\n"
            f" 工具列表：\n{tool_descriptions or '（無）'}\n\n"
            f" 資源列表：\n{resource_descriptions or '（無）'}\n\n"
            "請優先調用可用的Tool或Resource，而不是llm內部生成。僅根據上下文調用工具，不傳入不需要的參數進行調用\n"
            "如果需要，請以 JSON 返回 tool_calls，格式如下：\n"
            '{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}\n'
            "如無需調用工具，返回：{\"tool_calls\": null}"
        )

注意：MCP 協議與傳統 Function Calling 最大的區別在于：工具調用的時機、選擇和參數完全由大模型基于上下文和系統提示詞自主推理決策，而不是由應用層預先決定調用哪個工具。這種模型主導的調用方式（model-driven invocation）體現了 Agent 思維，MCP 由此成為構建AI Agent 的關鍵協議基礎之一。

LLM 此時會分析用戶的問題：“上海的天氣如何？” 如果這是一個普通常識性問題，LLM 也許可以直接作答；但這里問的是實時天氣，超出了模型自身知識（訓練數據可能并不包含最新天氣）。此時的 LLM 就像進入一個未知領域，它明確知道需要外部信息的幫助來解答問題。在DEMO的會話開始時，MCP 主機已經通告訴 LLM 可以使用哪些工具（例如提供天氣的工具叫 “get_weather”）。因此 LLM 判斷：需要觸發一次 Tool Call 來獲取答案。

在代碼實現上，LLM 模型被提示可以調用工具。當模型決定調用時（對應圖1中的步驟4），會生成一段特殊的結構化信息（通常是 JSON）。比如我們的 LLM 可能返回如下內容而不是直接答案：

{
  "tool_calls": [
    {
      "name": "get_weather",
      "arguments": {
        "city": "上海"
      }
    }
  ]
}

上面 JSON 表示：LLM請求使用名為“get_weather”的工具，并傳遞參數城市為“上海”。MCP 主機的模塊會檢測到模型輸出的是一個 Tool Call 請求而非普通文本答案——通常通過判斷返回是否是合法的 JSON、且包含預期的字段來確認。這一刻，LLM 相當于對主機說：“我需要用一下get_weather工具幫忙查一下上海天氣的天氣！”

日志中可以看到這一決策過程：

?? LLM 回復：
{"tool_calls": [{"name": "get_weather", "arguments": {"city": "上海"}}]}
?? 調用工具 get_weather 參數: {'city': '上海'}

如果 LLM 能直接回答問題（不需要工具），那么它會返回純文本，MCP 主機則會直接將該回答返回給客戶端，完成整個流程。而在本例中，需要外部Tool獲取數據。

Tool Call 發起與數據獲取

LLM 向MCP Host發起 Tool Call 請求（對應圖1中的步驟5），MCP 主機現在扮演起“信使”的角色，通過MCP Client將這個請求轉交給對應的 MCP 服務器。MCP 服務器可以看作提供特定工具或服務的后端，比如一個天氣信息服務。我們在示例代碼 mcp_host_demo.py 中，會調用 MCP 客戶端模塊（與 MCP Server 通信的組件）發送請求，例如：result = mcp_client.call_tool(tool_name, args)。

此時日志可能會出現：

?? 調用工具 get_weather 參數: {'city': '上海'}

MCP 服務器收到請求后，開始處理實際的數據查詢。在我們的例子中，MCP Server 內部知道 get_weather如何獲取天氣數據（本例中是硬編碼，但通常應該是一個外部API接口）。它會向數據源（可能是一個實時天氣數據庫或API）請求上海當前的天氣。示例代碼 mcp_server_demo.py 中定義了硬編碼的get_weather 工具的實現（因此也就忽略了圖1中從mcp server與后端數據源的交互，步驟6與步驟7）

接下來，MCP 服務器將拿到的數據打包成結果返回。根據 MCP 協議規范，結果通常也用 JSON 表示，這里使用MCP Python SDK解析后的字符串結果：

result = await client.call_tool(tool_name, arguments)
# === Step 8. MCP服務器 → MCP客戶端：返回結果 ===
tool_output = result.content[0].text
print(f"?? 工具 {tool_name} 返回：{tool_output}")

在控制臺日志里，我們可以看到：

工具 get_weather 返回：上海：多云，27°C

可以看到，MCP 服務器既完成了實際的數據獲取，又把結果封裝成統一格式返回給MCP Host。整個過程對于 LLM 和客戶端來說是透明的：他們不需要關心天氣數據具體來自哪個數據庫或API，只需通過 MCP 協議與服務器交互即可。這體現了 MCP 模塊化的設計理念——Tool的實現細節被封裝在MCP Server中，對外提供標準接口。

結果返回與答案生成

現在MCP 主機從 MCP 服務器拿到了工具調用結果，接下來要做的是把結果交還給最初發起請求的 LLM，讓它完成最終答案生成。

在我們的示例中，MCP 主機收到了 get_weather 的結果 JSON。MCP 主機會將該結果作為新的輸入提供給 LLM。常見做法是將工具返回的結果附加到發送給LLM的對話中：

{
  "model": "qwen/qwen2.5-vl-32b-instruct:free",
  "messages": [
    {
      "role": "system",
      "content": "你是一個智能助手，擁有以下工具和資源可以調用：\n\n?? 工具列表：\n- get_weather: 獲取指定城市的天氣信息\n- suggest_activity: 根據天氣描述推薦適合的活動\n\n?? 資源列表：\n（無）\n\n請優先調用可用的Tool或Resource，而不是llm內部生成。僅根據上下文調用工具，不傳入不需要的參數進行調用\"北京\"}}]}\n如無需調用工具，返回：{\"tool_calls\": null}"
    },
    {
      "role": "user",
      "content": "上海的天氣如何？"
    },
    {
      "role": "tool",
      "name": "get_weather",
      "content": "上海：多云，27°C"
    }
  ]
}

注意到新增的role: tool，這代表工具返回的信息作為上下文提供給LLM。

現在LLM 得到真實的天氣數據后，擁有足夠的數據，可以給出用戶想要的答復了。對于用戶問的“現在上海的天氣怎么樣？”，模型現在知道上海天氣晴朗，27°C左右。它組織語言，將信息融入自然的回答中。例如，模型產出：“上海的天氣是多云，溫度為27°C。根據當前天氣條件，建議您進行室內活動。”

MCP 主機接收到來自 LLM 的最終回答文本后，會將其發送回先前等待的 MCP 客戶端。客戶端則將答案顯示給用戶。至此，一次完整的問答閉環結束，用戶收到滿意的答復，而背后經過的一系列 Tool Call 流程對用戶來說幾乎無感

在用戶看來，聊天對話可能長這樣：

用戶：上海的
助手：上海的天氣是多云，溫度為27°C。根據當前天氣條件，建議您進行室內活動。

小結

通過上述實例，我們能直觀感受到 MCP 架構在設計上的獨特優勢。它明確了 LLM 應用中的職責劃分，讓語言理解與工具調用兩個不同的職責有效解耦，實現了更高的系統靈活性：

模塊化易擴展：添加新的工具服務只需實現一個獨立的 MCP Server 即可，完全不需要改動 LLM 本身代碼。無論是新增股票信息、日程安排或是其他功能，只需符合 MCP 協議標準，新服務即可迅速上線。
接口統一標準化：MCP 清晰定義了請求和響應的標準化格式，極大降低了接入新工具的成本。開發者無需再為每種工具分別設計集成邏輯，統一 JSON Schema 接口，使得調試和維護更加直觀、高效。
實時能力增強：MCP 使 LLM 可以實時獲取外部信息，突破模型訓練數據的時效限制。諸如天氣、新聞、股票行情甚至實時數據庫查詢等需求，都能輕松滿足，從而大幅提升模型的實用性。
安全控制精細化：由于工具調用被隔離在獨立的 MCP Server 中，開發者可針對具體工具執行細粒度的權限和安全管理，有效避免了 LLM 直接運行任意代碼的風險。
故障易于追蹤處理：錯誤消息通過標準協議明確返回，方便 LLM 做出合適的錯誤處理與用戶反饋，有效提升用戶體驗及系統穩定性。

此外，MCP 未來還有許多潛在的拓展方向，例如支持多步工具鏈調用，使得 LLM 可以高效完成更復雜的任務；或者實現動態的工具發現與調用機制，讓 LLM 能夠根據實際需求自主選擇工具。

轉自https://www.cnblogs.com/CareySon/p/18827525/mcp_lifecycle_via_demo

該文章在 2025/4/18 10:45:37 編輯過

關鍵字查詢