前言
在第一篇MCP文章中我們簡單介紹了Sampling：
採樣是工具與LLM交互以生成文本的機制。通過採樣，工具可以請求LLM生成文本內容，例如生成詩歌、文章或其他文本內容。採樣允許工具利用LLM的能力來創建內容，而不僅限於執行預定義的操作。
為什麼我們要在MCP Server通過Sampling方式調用Client的LLM，而不是MCP Server直接調用LLM呢？這背後其實有一套巧妙的設計哲學：
- MCP 服務端更像是一個"指揮家"，它統籌整合各種資源和工具，將它們編排成一個完整的服務提供給客户端。當服務端在實現其功能時需要藉助 LLM 的"智慧"，由服務端發起請求（服務端 -> 客户端 -> LLM）是最合理的安排。
- 根據 MCP 的設計理念，服務端專注於提供工具和資源服務，而不是直接與 LLM 交互。這就像是一個專業的中介，負責協調而不是親自下場。因此，服務端會將請求發給客户端，由客户端這個"橋樑"再將請求轉發到 LLM（服務端 -> 客户端 -> LLM）。
本文基於FastMCP演示下MCP Server和MCP Client如何實現Sampling，讓你徹底搞懂這個有趣的機制。
MCP Server
在MCP Server端，我們實現了一個情感分析工具，它會通過Sampling機制請求LLM幫助分析文本情感：
from fastmcp import Context, FastMCP
from mcp.types import SamplingMessage, TextContent
from pkg.log import logger
mcp = FastMCP("custom")
@mcp.tool()
async def analyze_sentiment(text: str, ctx: Context) -> dict:
"""Analyze the sentiment of a given text.
Args:
text (str): The text to analyze.
Returns:
dict: A dictionary containing the sentiment analysis result.
"""
prompt = f"""Analyze the sentiment of the following text as positive, negative, or neutral.
Just output a single word - 'positive', 'negative', or 'neutral'.
"""
logger.info(f"Analyzing sentiment for text: {text}, prompt: {prompt}")
response = await ctx.sample(
messages=[SamplingMessage(role="user", content=TextContent(type="text",text=text))],
system_prompt=prompt
)
logger.info(f"response: {response}")
sentiment = response.text.strip().lower()
# Map to standard sentiment values
if "positive" in sentiment:
sentiment = "positive"
elif "negative" in sentiment:
sentiment = "negative"
else:
sentiment = "neutral"
return {"text": text, "sentiment": sentiment}
if __name__ == "__main__":
mcp.run(transport="streamable-http", host="localhost", port=8001, show_banner=False)
這段代碼的核心在於ctx.sample()調用。當工具需要LLM的"智慧"時，它不會直接調用LLM API，而是通過上下文中的sample方法發起一個採樣請求。這就像你問朋友一個問題，朋友會去請教更專業的人，然後把答案告訴你。
MCP Client
MCP Client端的實現更加有趣，它需要同時扮演"翻譯官"和"調度員"的角色：
import asyncio
import json
import readline # For enhanced input editing
import traceback
from typing import cast
from fastmcp import Client
from fastmcp.client.sampling import SamplingMessage, SamplingParams
from mcp.shared.context import RequestContext
from openai import AsyncOpenAI
from openai.types.chat import ChatCompletionMessageFunctionToolCall
from pkg.config import cfg
from pkg.log import logger
class MCPHost:
"""MCP主機類，用於管理與MCP服務器的連接和交互"""
def __init__(self, server_uri: str):
"""
初始化MCP客户端
Args:
server_uri (str): MCP服務器的URI地址
"""
# 初始化MCP客户端連接
self.mcp_client: Client = Client(server_uri, sampling_handler=self.sampling_handler)
# 初始化異步OpenAI客户端用於與LLM交互
self.llm = AsyncOpenAI(
base_url=cfg.llm_base_url,
api_key=cfg.llm_api_key,
)
# 存儲對話歷史消息
self.messages = []
async def close(self):
"""關閉MCP客户端連接"""
if self.mcp_client:
await self.mcp_client.close()
async def sampling_handler(self, messages: list[SamplingMessage], params: SamplingParams, ctx: RequestContext) -> str:
"""處理採樣消息的回調函數"""
conversation = []
# Use the system prompt if provided
system_prompt = params.systemPrompt or "You are a helpful assistant."
conversation.append({"role": "system", "content": system_prompt})
for message in messages:
content = message.content.text if hasattr(message.content, 'text') else str(message.content)
conversation.append({"role": message.role, "content": content})
resp = await self.llm.chat.completions.create(
model=cfg.llm_model,
messages=conversation,
temperature=0.3,
)
message = resp.choices[0].message
return message.content if hasattr(message, "content") else ""
async def process_query(self, query: str) -> str:
"""Process a user query by interacting with the MCP server and LLM.
Args:
query (str): The user query to process.
Returns:
str: The response from the MCP server.
"""
# 將用户查詢添加到消息歷史中
self.messages.append({
"role": "user",
"content": query,
})
# 使用異步上下文管理器確保MCP客户端連接正確建立和關閉
async with self.mcp_client:
# 從MCP服務器獲取可用工具列表
tools = await self.mcp_client.list_tools()
# 構造LLM可以理解的工具格式
available_tools = []
# 將MCP工具轉換為OpenAI格式
for tool in tools:
available_tools.append({
"type": "function",
"function": {
"name": tool.name,
"description": tool.description,
"parameters": tool.inputSchema,
}
})
logger.info(f"Available tools: {[tool['function']['name'] for tool in available_tools]}")
# 調用LLM，傳入對話歷史和可用工具
resp = await self.llm.chat.completions.create(
model=cfg.llm_model,
messages=self.messages,
tools=available_tools,
temperature=0.3,
)
# 存儲最終響應文本
final_text = []
# 獲取LLM的首個響應消息
message = resp.choices[0].message
# 如果響應包含直接內容，則添加到結果中
if hasattr(message, "content") and message.content:
final_text.append(message.content)
# 循環處理工具調用，直到沒有更多工具調用為止
while message.tool_calls:
# 遍歷所有工具調用
for tool_call in message.tool_calls:
# 確保工具調用有函數信息
if not hasattr(tool_call, "function"):
continue
# 類型轉換以獲取函數調用詳情
function_call = cast(ChatCompletionMessageFunctionToolCall, tool_call)
function = function_call.function
tool_name = function.name
# 解析函數參數
tool_args = json.loads(function.arguments)
# 檢查MCP客户端是否已連接
if not self.mcp_client.is_connected():
raise RuntimeError("Session not initialized. Cannot call tool.")
# 調用MCP服務器上的指定工具
result = await self.mcp_client.call_tool(tool_name, tool_args)
# 將助手的工具調用添加到消息歷史中
self.messages.append({
"role": "assistant",
"tool_calls": [
{
"id": tool_call.id,
"type": "function",
"function": {
"name": function.name,
"arguments": function.arguments
}
}
]
})
# 將工具調用結果添加到消息歷史中
self.messages.append({
"role": "tool",
"tool_call_id":tool_call.id,
"content": str(result.content) if result.content else ""
})
# 基於工具調用結果再次調用LLM
final_resp = await self.llm.chat.completions.create(
model=cfg.llm_model,
messages=self.messages,
tools=available_tools,
temperature=0.3,
)
# 更新消息為最新的LLM響應
message = final_resp.choices[0].message
# 如果響應包含內容，則添加到最終結果中
if message.content:
final_text.append(message.content)
# 返回連接後的完整響應
return "\n".join(final_text)
async def chat_loop(self):
"""主聊天循環，處理用户輸入並顯示響應"""
print("Welcome to the MCP chat! Type 'quit' to exit.")
# 持續處理用户輸入直到用户退出
while True:
try:
# 獲取用户輸入
query = input("You: ").strip()
# 檢查退出命令
if query.lower() == "quit":
print("Exiting chat. Goodbye!")
break
# 跳過空輸入
if not query:
continue
# 處理用户查詢並獲取響應
resp = await self.process_query(query)
print(f"Assistant: {resp}")
# 捕獲並記錄聊天循環中的任何異常
except Exception as e:
logger.error(f"Error in chat loop: {str(e)}")
logger.error(traceback.format_exc())
async def main():
"""主函數，程序入口點"""
# 創建MCP主機實例
client = MCPHost(server_uri="http://localhost:8001/mcp")
try:
# 啓動聊天循環
await client.chat_loop()
except Exception as e:
# 記錄主程序中的任何異常
logger.error(f"Error in main: {str(e)}")
logger.error(traceback.format_exc())
finally:
# 確保客户端連接被正確關閉
await client.close()
if __name__ == "__main__":
# 運行主程序
asyncio.run(main())
Client的關鍵在於sampling_handler函數，當Server端發起採樣請求時，Client會通過這個函數接收請求，並實際調用LLM完成文本生成。這就像一個稱職的助理，當老闆（Server）需要某些信息時，助理（Client）會去查詢資料（LLM）並把結果彙報給老闆。
Client運行輸出
Welcome to the MCP chat! Type 'quit' to exit.
You: 分析下這句的情感傾向：問君能有幾多愁，恰似一江春水向東流
Assistant: 這句詩"問君能有幾多愁，恰似一江春水向東流"表達的情感傾向是**負面**的。它通過比喻的方式，將憂愁比作一江春水向東流，暗示了憂愁的綿長和無法排解，帶有濃厚的哀愁與感傷情緒。
You: what can you do?
Assistant: 我可以幫助你進行情感分析，例如分析詩句、句子的情感傾向。如果你有其他需求，也可以告訴我，我會盡力提供幫助！
You: quit
Exiting chat. Goodbye!
從運行結果可以看出，當用户請求分析詩句情感時，整個流程是這樣的：
- 用户輸入需要分析的詩句
- Client調用Server端的analyze_sentiment工具
- 工具通過Sampling機制請求LLM分析情感
- Client接收LLM的響應並返回給用户
小結
通過這個示例，我們可以看到MCP中Sampling機制的巧妙之處：
- 職責分離：MCP Server專注於業務邏輯和工具編排，不直接與LLM交互，保持了架構的清晰性。
- 靈活性：Client端可以自由選擇不同的LLM提供商和模型，Server端無需關心具體實現細節。
- 可擴展性：可以輕鬆添加更多需要LLM能力的工具，而無需修改Client端的LLM調用邏輯。
- 統一接口：通過標準化的Sampling接口，不同組件之間可以無縫協作。
這種設計讓MCP系統既保持了良好的模塊化結構，又充分發揮了LLM的能力，真正做到了"各司其職，協同工作"。
參考
- yuan - 一文讀懂 MCP 的 Sampling（採樣），賦予 MCP 服務端智能能力！
