在LLM應(yīng)用的快速發(fā)展中,一個(gè)核心挑戰(zhàn)始終存在:如何讓模型獲取最新、最準(zhǔn)確的外部知識并有效利用工具?
背景其實(shí)很簡單:大模型(LLM)再強(qiáng),也總有不知道的東西,怎么辦?讓它“查資料”“調(diào)工具”成了近兩年最熱的技術(shù)方向。從最早的 RAG(Retrieval-Augmented Generation),到 OpenAI 引領(lǐng)的 Function Call,再到現(xiàn)在 Anthropic 拋出的 MCP(Model Context Protocol),每一代方案都在試圖解答一個(gè)問題:模型如何以更自然的方式獲得外部世界的幫助?
MCP 主打的是統(tǒng)一標(biāo)準(zhǔn)和跨模型兼容性。雖然協(xié)議本身尚處于早期階段,設(shè)計(jì)也遠(yuǎn)稱不上完美,但出現(xiàn)的時(shí)機(jī)十分巧妙。。就像當(dāng)年 OpenAI 的 API,一旦形成事實(shí)標(biāo)準(zhǔn),后面哪怕有點(diǎn)毛病,也可以很快改進(jìn),畢竟生態(tài)具有滾雪球效應(yīng),一旦用戶基數(shù)形成規(guī)模,自然而然就成為事實(shí)標(biāo)準(zhǔn)。
本篇文章將結(jié)合 MCP 官方 SDK,通過代碼和流程圖模擬一次帶 Tool 調(diào)用的完整交互過程,了解并看清 MCP 的全生命周期。
整體流程
一次MCP完整的調(diào)用流程如下:

圖1. 一次包含MCP調(diào)用的完整流程
圖1省略了第一步與第二步之間,list_tools()或resource()的步驟,也就是最開始MCP Host知道有哪些可用的工具與資源,我們在本 DEMO 中使用了硬編碼的方式將資源信息構(gòu)建在提示詞中。
這里需要注意的是MCP Client與MCP Host(主機(jī))并不是分離的部分,但為了時(shí)序圖清晰,這里將其邏輯上拆分為不同的部分,實(shí)際上MCP Host可以理解為我們需要嵌入AI的應(yīng)用程序,例如 CRM 系統(tǒng)或 SaaS 服務(wù),實(shí)際上Host中是包含MCP Client的代碼。實(shí)際的 MCP Host 與 Client 結(jié)構(gòu)如下圖所示:

整體示例代碼
MCP Server
mcp server的代碼使用最簡單的方式啟動(dòng),并通過Python裝飾器注冊最簡單的兩個(gè)工具,為了DEMO簡單,hard code兩個(gè)工具(函數(shù))返回值,代碼如下:
#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="根據(jù)天氣描述推薦適合的活動(dòng)")
async def suggest_activity(condition: str) -> str:
"""
根據(jù)天氣描述推薦適合的活動(dòng)
"""
if "晴" in condition:
return "天氣晴朗,推薦你去戶外散步或運(yùn)動(dòng)。"
elif "多云" in condition:
return "多云天氣適合逛公園或咖啡館。"
elif "雨" in condition:
return "下雨了,建議你在家閱讀或看電影。"
else:
return "建議進(jìn)行室內(nèi)活動(dòng)。"
async def main():
print("? 啟動(dòng) MCP Server: http://127.0.0.1:1234")
await mcp.run_sse_async()
if __name__ == "__main__":
asyncio.run(main())

大模型調(diào)用代碼
大模型調(diào)用選擇使用openrouter這個(gè)LLM的聚合網(wǎng)站,主要是因?yàn)樵摼W(wǎng)站方便調(diào)用與測試不同的模型,同時(shí)網(wǎng)絡(luò)環(huán)境可以直接連接而不用其他手段。
代碼如下:
import json
import requests
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 來生成回復(fù)
"""
def __init__(self, model: str = LLM_MODEL):
self.model = model
def generate(self, messages):
"""
發(fā)送對話消息給 OpenRouter API 并返回 LLM 的回復(fù)文本
參數(shù):
messages: 一個(gè) list,每個(gè)元素都是形如 {'role': role, 'content': content} 的字典
返回:
LLM 返回的回復(fù)文本
"""
request_body = {
"model": self.model,
"messages": messages
}
print(f"發(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 錯(cuò)誤: {response.status_code}")
print(f"錯(cuò)誤詳情: {response.text}")
raise Exception(f"OpenRouter API 返回錯(cuò)誤: {response.status_code}")
response_json = response.json()
print(f"OpenRouter API 響應(yīng): {json.dumps(response_json, ensure_ascii=False)}")
try:
content = response_json['choices'][0]['message']['content']
return content
except KeyError:
raise Exception("無法從 OpenRouter 響應(yīng)中提取內(nèi)容")
if __name__ == "__main__":
messages = [
{"role": "system", "content": "你是一個(gè)智能助手,可以幫助查詢天氣信息。"},
{"role": "user", "content": "請告訴我北京今天的天氣情況。"}
]
llm = OpenRouterLLM()
try:
result = llm.generate(messages)
print("LLM 返回結(jié)果:")
print(result)
except Exception as e:
print(f"調(diào)用 OpenRouter 時(shí)發(fā)生異常: {e}")
MCP Client
這里的MCP Client,使用Server-Side Event(SSE)方式進(jìn)行連接(題外話,MCP協(xié)議使用SSE協(xié)議作為默認(rèn)遠(yuǎn)程協(xié)議稍微有點(diǎn)奇怪,聽說后續(xù)迭代會考慮HTTP Streaming以及JSONRPC over HTTP2的方式)。
這里我們在main測試代碼中,嘗試列出所有可用的Tool與Resource,并嘗試調(diào)用Tool,結(jié)果如圖,可以看到能夠展示出MCP Server中定義的Tool。
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):
self._sse_context = sse_client(self.server_url)
self.read, self.write = await self._sse_context.__aenter__()
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?? 調(diào)用 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的應(yīng)用,可以是一個(gè)程序,可以是一個(gè)CRM系統(tǒng),可以是一個(gè)OA,MCP Host包含MCP Client,用于集成LLM與Tool,MCP Host之外+Tool+大模型,共同構(gòu)成了一套基于AI的系統(tǒng),現(xiàn)在流行的說法是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 內(nèi)容,自動(dòng)處理 markdown 包裹、多余引號、嵌套等。
支持 string 或 dict 格式。
如果無法解出 dict,則返回原始 string。
"""
if isinstance(reply, dict):
return reply
if isinstance(reply, str):
reply = re.sub(r"^```(?:json)?|```$", "", reply.strip(), flags=re.IGNORECASE).strip()
for _ in range(3):
try:
parsed = json.loads(reply)
if isinstance(parsed, dict):
return parsed
else:
reply = parsed
except Exception:
break
return reply
llm = OpenRouterLLM()
async def main():
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:
user_input = input("\n請輸入你的問題(輸入 exit 退出):\n> ")
if user_input.lower() in ("exit", "退出"):
break
system_prompt = (
"你是一個(gè)智能助手,擁有以下工具和資源可以調(diào)用:\n\n"
f"?? 工具列表:\n{tool_descriptions or '(無)'}\n\n"
f"?? 資源列表:\n{resource_descriptions or '(無)'}\n\n"
"請優(yōu)先調(diào)用可用的Tool或Resource,而不是llm內(nèi)部生成。僅根據(jù)上下文調(diào)用工具,不傳入不需要的參數(shù)進(jìn)行調(diào)用\n"
"如果需要,請以 JSON 返回 tool_calls,格式如下:\n"
'{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}\n'
"如無需調(diào)用工具,返回:{\"tool_calls\": null}"
)
messages = [
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_input}
]
final_reply = ""
while True:
reply = llm.generate(messages)
print(f"\n?? LLM 回復(fù):\n{reply}")
parsed = extract_json_from_reply(reply)
if isinstance(parsed, str):
final_reply = parsed
break
tool_calls = parsed.get("tool_calls")
if not tool_calls:
final_reply = parsed.get("content", "")
break
for tool_call in tool_calls:
tool_name = tool_call["name"]
arguments = tool_call["arguments"]
if tool_name not in tool_names:
raise ValueError(f"? 工具 {tool_name} 未注冊")
print(f"?? 調(diào)用工具 {tool_name} 參數(shù): {arguments}")
result = await client.call_tool(tool_name, arguments)
tool_output = result.content[0].text
print(f"?? 工具 {tool_name} 返回:{tool_output}")
messages.append({
"role": "tool",
"name": tool_name,
"content": tool_output
})
print(f"\n?? 最終答復(fù):{final_reply}")
await client.__aexit__(None, None, None)
if __name__ == "__main__":
asyncio.run(main())
用戶提問
DEMO的交互方式是一個(gè)簡單的Chatbox。假設(shè)用戶在聊天界面的輸入框里敲下:“上海的天氣如何” 。此時(shí),用戶的問題通過 MCP 主機(jī)(MCP Host) 被發(fā)送給大模型。
MCP Host 可以是一個(gè)瀏覽器前端、桌面應(yīng)用,也可以只是后端的一段代碼。在這個(gè)場景里,它主要負(fù)責(zé)收集用戶輸入并與LLM通信。
對應(yīng)流程圖1中的步驟1:提出問題 與步驟2:轉(zhuǎn)發(fā)問題。
LLM 推理:是否需要外部Tool配合
收到用戶提問后,MCP 主機(jī)(Host)負(fù)責(zé)將用戶提問解析并附加上下文后轉(zhuǎn)發(fā)給大模型。主要取決于系統(tǒng)設(shè)計(jì)的智能程度、工具豐富度,以及 LLM 的能力邊界。通常可以是一段靜態(tài)的提示詞,或者從上下文中獲取動(dòng)態(tài)的提示詞,也可以是通過一些外部API獲取數(shù)據(jù)生成提示詞,這并不是本文的重點(diǎn),本文通過簡單的靜態(tài)提示詞進(jìn)行。
本DEMO的靜態(tài)提示詞如下:
# 構(gòu)造系統(tǒng)提示 + 工具說明
system_prompt = (
"你是一個(gè)智能助手,擁有以下工具和資源可以調(diào)用:\n\n"
f" 工具列表:\n{tool_descriptions or '(無)'}\n\n"
f" 資源列表:\n{resource_descriptions or '(無)'}\n\n"
"請優(yōu)先調(diào)用可用的Tool或Resource,而不是llm內(nèi)部生成。僅根據(jù)上下文調(diào)用工具,不傳入不需要的參數(shù)進(jìn)行調(diào)用\n"
"如果需要,請以 JSON 返回 tool_calls,格式如下:\n"
'{"tool_calls": [{"name": "get_weather", "arguments": {"city": "北京"}}]}\n'
"如無需調(diào)用工具,返回:{\"tool_calls\": null}"
)
注意:MCP 協(xié)議與傳統(tǒng) Function Calling 最大的區(qū)別在于:工具調(diào)用的時(shí)機(jī)、選擇和參數(shù)完全由大模型基于上下文和系統(tǒng)提示詞自主推理決策,而不是由應(yīng)用層預(yù)先決定調(diào)用哪個(gè)工具。這種模型主導(dǎo)的調(diào)用方式(model-driven invocation)體現(xiàn)了 Agent 思維,MCP 由此成為構(gòu)建AI Agent 的關(guān)鍵協(xié)議基礎(chǔ)之一。
LLM 此時(shí)會分析用戶的問題:“上海的天氣如何?” 如果這是一個(gè)普通常識性問題,LLM 也許可以直接作答;但這里問的是實(shí)時(shí)天氣,超出了模型自身知識(訓(xùn)練數(shù)據(jù)可能并不包含最新天氣)。此時(shí)的 LLM 就像進(jìn)入一個(gè)未知領(lǐng)域,它明確知道需要外部信息的幫助來解答問題。在DEMO的會話開始時(shí),MCP 主機(jī)已經(jīng)通告訴 LLM 可以使用哪些工具(例如提供天氣的工具叫 “get_weather”)。因此 LLM 判斷:需要觸發(fā)一次 Tool Call 來獲取答案。
在代碼實(shí)現(xiàn)上,LLM 模型被提示可以調(diào)用工具。當(dāng)模型決定調(diào)用時(shí)(對應(yīng)圖1中的步驟4),會生成一段特殊的結(jié)構(gòu)化信息(通常是 JSON)。比如我們的 LLM 可能返回如下內(nèi)容而不是直接答案:
{
"tool_calls": [
{
"name": "get_weather",
"arguments": {
"city": "上海"
}
}
]
}
上面 JSON 表示:LLM請求使用名為“get_weather”的工具,并傳遞參數(shù)城市為“上海”。MCP 主機(jī)的 模塊會檢測到模型輸出的是一個(gè) Tool Call 請求 而非普通文本答案——通常通過判斷返回是否是合法的 JSON、且包含預(yù)期的字段來確認(rèn)。這一刻,LLM 相當(dāng)于對主機(jī)說:“我需要用一下get_weather工具幫忙查一下上海天氣的天氣!”
日志中可以看到這一決策過程:
?? LLM 回復(fù):
{"tool_calls": [{"name": "get_weather", "arguments": {"city": "上海"}}]}
?? 調(diào)用工具 get_weather 參數(shù): {'city': '上海'}
如果 LLM 能直接回答問題(不需要工具),那么它會返回純文本,MCP 主機(jī)則會直接將該回答返回給客戶端,完成整個(gè)流程。而在本例中, 需要外部Tool獲取數(shù)據(jù)。
Tool Call 發(fā)起與數(shù)據(jù)獲取
LLM 向MCP Host發(fā)起 Tool Call 請求(對應(yīng)圖1中的步驟5),MCP 主機(jī)現(xiàn)在扮演起“信使”的角色,通過MCP Client將這個(gè)請求轉(zhuǎn)交給對應(yīng)的 MCP 服務(wù)器。MCP 服務(wù)器可以看作提供特定工具或服務(wù)的后端,比如一個(gè)天氣信息服務(wù)。我們在示例代碼 mcp_host_demo.py 中,會調(diào)用 MCP 客戶端模塊(與 MCP Server 通信的組件)發(fā)送請求,例如:result = mcp_client.call_tool(tool_name, args)。
此時(shí)日志可能會出現(xiàn):
?? 調(diào)用工具 get_weather 參數(shù): {'city': '上海'}
MCP 服務(wù)器收到請求后,開始處理實(shí)際的數(shù)據(jù)查詢。在我們的例子中,MCP Server 內(nèi)部知道 get_weather如何獲取天氣數(shù)據(jù)(本例中是硬編碼,但通常應(yīng)該是一個(gè)外部API接口)。它會向數(shù)據(jù)源(可能是一個(gè)實(shí)時(shí)天氣數(shù)據(jù)庫或API)請求上海當(dāng)前的天氣。示例代碼 mcp_server_demo.py 中定義了 硬編碼的get_weather 工具的實(shí)現(xiàn)(因此也就忽略了圖1中從mcp server與后端數(shù)據(jù)源的交互,步驟6與步驟7)
接下來,MCP 服務(wù)器將拿到的數(shù)據(jù)打包成結(jié)果返回。根據(jù) MCP 協(xié)議規(guī)范,結(jié)果通常也用 JSON 表示,這里使用MCP Python SDK解析后的字符串結(jié)果:
result = await client.call_tool(tool_name, arguments)
tool_output = result.content[0].text
print(f"?? 工具 {tool_name} 返回:{tool_output}")
在控制臺日志里,我們可以看到:
工具 get_weather 返回:上海:多云,27°C
可以看到,MCP 服務(wù)器既完成了實(shí)際的數(shù)據(jù)獲取,又把結(jié)果封裝成統(tǒng)一格式返回給MCP Host。整個(gè)過程對于 LLM 和客戶端來說是透明的:他們不需要關(guān)心天氣數(shù)據(jù)具體來自哪個(gè)數(shù)據(jù)庫或API,只需通過 MCP 協(xié)議與服務(wù)器交互即可。這體現(xiàn)了 MCP 模塊化的設(shè)計(jì)理念——Tool的實(shí)現(xiàn)細(xì)節(jié)被封裝在MCP Server中,對外提供標(biāo)準(zhǔn)接口。
結(jié)果返回與答案生成
現(xiàn)在MCP 主機(jī)從 MCP 服務(wù)器拿到了工具調(diào)用結(jié)果,接下來要做的是把結(jié)果交還給最初發(fā)起請求的 LLM,讓它完成最終答案生成。
在我們的示例中,MCP 主機(jī)收到了 get_weather 的結(jié)果 JSON。MCP 主機(jī)會將該結(jié)果作為新的輸入提供給 LLM。常見做法是將工具返回的結(jié)果附加到發(fā)送給LLM的對話中:
{
"model": "qwen/qwen2.5-vl-32b-instruct:free",
"messages": [
{
"role": "system",
"content": "你是一個(gè)智能助手,擁有以下工具和資源可以調(diào)用:\n\n?? 工具列表:\n- get_weather: 獲取指定城市的天氣信息\n- suggest_activity: 根據(jù)天氣描述推薦適合的活動(dòng)\n\n?? 資源列表:\n(無)\n\n請優(yōu)先調(diào)用可用的Tool或Resource,而不是llm內(nèi)部生成。僅根據(jù)上下文調(diào)用工具,不傳入不需要的參數(shù)進(jìn)行調(diào)用\"北京\"}}]}\n如無需調(diào)用工具,返回:{\"tool_calls\": null}"
},
{
"role": "user",
"content": "上海的天氣如何?"
},
{
"role": "tool",
"name": "get_weather",
"content": "上海:多云,27°C"
}
]
}
注意到新增的role: tool,這代表工具返回的信息作為上下文提供給LLM。
現(xiàn)在LLM 得到真實(shí)的天氣數(shù)據(jù)后,擁有足夠的數(shù)據(jù),可以給出用戶想要的答復(fù)了。對于用戶問的“現(xiàn)在上海的天氣怎么樣?”,模型現(xiàn)在知道上海天氣晴朗,27°C左右。它組織語言,將信息融入自然的回答中。例如,模型產(chǎn)出:“上海的天氣是多云,溫度為27°C。根據(jù)當(dāng)前天氣條件,建議您進(jìn)行室內(nèi)活動(dòng)。”
MCP 主機(jī)接收到來自 LLM 的最終回答文本后,會將其發(fā)送回先前等待的 MCP 客戶端。客戶端則將答案顯示給用戶。至此,一次完整的問答閉環(huán)結(jié)束,用戶收到滿意的答復(fù),而背后經(jīng)過的一系列 Tool Call 流程對用戶來說幾乎無感
在用戶看來,聊天對話可能長這樣:
用戶:上海的
助手:上海的天氣是多云,溫度為27°C。根據(jù)當(dāng)前天氣條件,建議您進(jìn)行室內(nèi)活動(dòng)。
小結(jié)
通過上述實(shí)例,我們能直觀感受到 MCP 架構(gòu)在設(shè)計(jì)上的獨(dú)特優(yōu)勢。它明確了 LLM 應(yīng)用中的職責(zé)劃分,讓語言理解與工具調(diào)用兩個(gè)不同的職責(zé)有效解耦,實(shí)現(xiàn)了更高的系統(tǒng)靈活性:
模塊化易擴(kuò)展:添加新的工具服務(wù)只需實(shí)現(xiàn)一個(gè)獨(dú)立的 MCP Server 即可,完全不需要改動(dòng) LLM 本身代碼。無論是新增股票信息、日程安排或是其他功能,只需符合 MCP 協(xié)議標(biāo)準(zhǔn),新服務(wù)即可迅速上線。
接口統(tǒng)一標(biāo)準(zhǔn)化:MCP 清晰定義了請求和響應(yīng)的標(biāo)準(zhǔn)化格式,極大降低了接入新工具的成本。開發(fā)者無需再為每種工具分別設(shè)計(jì)集成邏輯,統(tǒng)一 JSON Schema 接口,使得調(diào)試和維護(hù)更加直觀、高效。
實(shí)時(shí)能力增強(qiáng):MCP 使 LLM 可以實(shí)時(shí)獲取外部信息,突破模型訓(xùn)練數(shù)據(jù)的時(shí)效限制。諸如天氣、新聞、股票行情甚至實(shí)時(shí)數(shù)據(jù)庫查詢等需求,都能輕松滿足,從而大幅提升模型的實(shí)用性。
安全控制精細(xì)化:由于工具調(diào)用被隔離在獨(dú)立的 MCP Server 中,開發(fā)者可針對具體工具執(zhí)行細(xì)粒度的權(quán)限和安全管理,有效避免了 LLM 直接運(yùn)行任意代碼的風(fēng)險(xiǎn)。
故障易于追蹤處理:錯(cuò)誤消息通過標(biāo)準(zhǔn)協(xié)議明確返回,方便 LLM 做出合適的錯(cuò)誤處理與用戶反饋,有效提升用戶體驗(yàn)及系統(tǒng)穩(wěn)定性。
此外,MCP 未來還有許多潛在的拓展方向,例如支持多步工具鏈調(diào)用,使得 LLM 可以高效完成更復(fù)雜的任務(wù);或者實(shí)現(xiàn)動(dòng)態(tài)的工具發(fā)現(xiàn)與調(diào)用機(jī)制,讓 LLM 能夠根據(jù)實(shí)際需求自主選擇工具。
轉(zhuǎn)自https://www.cnblogs.com/CareySon/p/18827525/mcp_lifecycle_via_demo
該文章在 2025/4/18 10:45:37 編輯過