Agent 與 MCP
MCP 進階第一課:sampling、進度通知與 roots
MCP 進階第一課:sampling、進度通知與 roots
前面幾篇把 MCP 的基本盤打完了。這篇進入進階主題的第一部分,三個讓 server 更成熟的功能:sampling 讓 server 借用 client 的 Claude 連線來生成文字;logging 與 progress 通知讓長時間任務不再像當機;roots 幫檔案存取劃出安全邊界。三個都不難,但都是公開 server 上線前該想清楚的事。
你將學到什麼
sampling 是什麼
server 請 client 代為呼叫 Claude,費用歸 client。
兩端怎麼實作
server 發 create_message,client 註冊 callback 接。
進度與日誌通知
用 Context 把長任務的狀態即時回報給使用者。
roots 檔案邊界
授權資料夾讓 Claude 找得到檔案,也出不了圍欄。
sampling:請 client 幫你呼叫 Claude
想像你的 MCP server 有個研究工具,會抓一堆維基百科資料,最後需要把資料總結成報告。生成文字要用到 Claude,你有兩個選項。選項一:讓 server 自己接 Claude,那就得自備 API key、處理認證、吸收費用,整合的複雜度全上身。選項二就是 sampling:server 產好 prompt,問 client 一句「可以幫我呼叫 Claude 嗎?」,由本來就連著 Claude 的 client 代打,把生成結果傳回來。
這一搬,好處全出來了:
- server 不必整合語言模型,複雜度大降。
- token 費用由 client 承擔,不是 server。
- server 不需要任何 API key 與認證設定。
- 對公開 server 尤其關鍵:你不會想替所有陌生使用者買單生成費用。
什麼時候最值得用?公開 server。你的 server 一旦開放給陌生人連,任何會生成文字的功能都可能被灌爆;sampling 把模型呼叫整個留在 client 端,server 只出 prompt,功能照提供,帳單不經手。而 client 那邊本來就有 Claude 的連線與憑證,多接這一件事的成本很低。
兩端的實作
server 端在工具函式裡透過 Context 發出 create_message 請求,拿回生成結果:
@mcp.tool()
async def summarize(text_to_summarize: str, ctx: Context):
prompt = f"""
Please summarize the following text:
{text_to_summarize}
"""
result = await ctx.session.create_message(
messages=[
SamplingMessage(
role="user",
content=TextContent(type="text", text=prompt)
)
],
max_tokens=4000,
system_prompt="You are a helpful research assistant",
)
if result.content.type == "text":
return result.content.text
raise ValueError("Sampling failed")
create_message 的參數跟你平常呼叫 Claude API 很像:messages、max_tokens、system_prompt 一應俱全,只是收件人從 API 換成了 client。拿到結果先檢查內容型別是文字才回傳,失敗就丟例外。
client 端則註冊一個 sampling callback:收到請求就用 Anthropic SDK 呼叫 Claude,把結果包成 CreateMessageResult 回去,並在建立 session 時把 callback 傳入:
async def sampling_callback(
context: RequestContext, params: CreateMessageRequestParams
):
# call Claude with the Anthropic SDK
text = await chat(params.messages)
return CreateMessageResult(
role="assistant",
model=model,
content=TextContent(type="text", text=text),
)
async with ClientSession(
read, write, sampling_callback=sampling_callback
) as session:
await session.initialize()
流程再走一次:server 端的工具做完自己的活(例如抓完資料),組好 prompt 發出 create_message;client 端的 callback 收到後,用自己的 Claude 連線代打,把生成文字包回去;server 拿到文字,放進工具結果回覆。責任與費用的分界線,清清楚楚。
logging 與 progress:把過程說出來
Claude 呼叫一個要跑很久的工具時,使用者常常只能盯著沒有動靜的畫面,猜它到底是在跑還是掛了。logging 與 progress 通知就是解方,而且實作起來只是幾行程式,卻是使用體驗差距最大的地方。工具函式透過自動注入的 Context 引數,用 context.info() 送日誌、用 context.report_progress() 回報進度:
@mcp.tool(
name="research",
description="Research a given topic"
)
async def research(
topic: str = Field(description="Topic to research"),
*,
context: Context
):
await context.info("About to do research...")
await context.report_progress(20, 100)
sources = await do_research(topic)
await context.info("Writing report...")
await context.report_progress(70, 100)
return await generate_report(sources)
示範的 research 工具就是典型場景:查資料要一陣子,寫報告又要一陣子,中間沒有任何回饋的話,使用者早就開始懷疑是不是壞了。加上兩行 info 與兩次 report_progress,體感完全不同。
client 端則註冊 logging callback 與 progress callback 來接收:logging callback 在建立 session 時傳入,progress callback 在個別工具呼叫時傳入。要怎麼呈現隨應用型態自由選:
- CLI 應用:直接把訊息與進度印到終端機。
- 網頁應用:用 WebSocket、SSE 或輪詢把更新推到瀏覽器。
- 桌面應用:更新進度條與狀態顯示。
通知完全是選配,忽略也不影響功能,但對使用體驗的差別非常大:使用者看得到進度條在動、日誌在跑,就知道系統活著。
roots:檔案存取的邊界
假設 server 有個影片轉檔工具,使用者說「把 biking.mp4 轉成 mov」。Claude 只拿到檔名,它沒有辦法搜遍整台電腦找出檔案在哪;要求使用者每次都打完整路徑,又太不友善。
roots 的做法:由 client 授予 server 一組可存取的資料夾。Claude 先呼叫 list_roots 看有哪些地方可以找,再逐層讀目錄找到檔案,最後帶完整路徑呼叫轉檔工具。使用者照樣只說一句話,找檔案的事自動完成。
roots 同時是安全邊界:只授權桌面資料夾,server 就碰不到文件或下載資料夾,試圖存取邊界外的檔案會拿到錯誤。要注意的是,SDK 不會自動幫你擋,你得自己實作檢查。
常見做法是寫一個 is_path_allowed() 之類的輔助函式:拿到請求的路徑,對照已授權的 roots 清單,確認落在範圍內才放行,並在每個碰檔案的工具裡先驗過再動作。
整理 roots 帶來的四個好處:
- 使用者不必打完整路徑,一句話就能指定檔案。
- 搜尋範圍聚焦在授權資料夾,找檔案更快。
- 防止意外碰到邊界外的敏感檔案。
- 彈性:roots 可以透過工具提供,也可以直接注入 prompt。
