AG Day 28 MCP client:把研究助理接上 MCP 生態
執行需求:CPU 可跑。前三天我們把 MCP 的概念(AG Day 25(原文連結))、第一個 FastMCP server(AG Day 26(原文連結))、以及 resources 與 prompts 的進階能力(AG Day 27(原文連結))一步一步建立起來,但都停在「伺服器這一端」。今天要把視角換到客戶端:寫一個能跟任何遵循 MCP 規格的伺服器對話的客戶端介面,把它接進 research-agent 既有的 LangGraph 流程(AG Day 11 到 AG Day 18)。換句話說,今天要讓我們的代理不再只能呼叫專案內部的函式,而是能自動發現、動態串接、跨行程呼叫任何支援 MCP 的外部工具與資源。本篇完全在本機 CPU 上執行,不需要任何 API 金鑰,並沿用 AG Day 26 的 basic_server.py 作為可呼叫的目標。
引言
先想像一個情境:未來某天,MCP 生態成熟了,你會在社群看到「官方出品的『政府開放資料 MCP server』」「某團隊寫的『學術論文搜尋 MCP server』」「另一個獨立開發者的『SQL 資料庫查詢 MCP server』」。這些 server 各自獨立發布、各自維護,但都遵循同一份規格。今天的目標就是讓 research-agent 只要設定一次,就能自動跟這些 server 對話——不需要為每個 server 寫專屬的 wrapper。
這個願景聽起來理所當然,但實作上要解決三個問題。第一,連線的生命週期管理:每個 server 都是獨立行程,啟動它、監看它、結束時關掉它,都是客戶端的責任。第二,工具名稱的命名空間:當兩個 server 都提供 echo 工具時,客戶端要怎麼區分它們。第三,訊息的路由:客戶端同時對接多個 server 時,每一個 JSON-RPC 訊息都要送到正確的目的地,並且不能把來自 server A 的回應誤送到 server B 的請求上。
今天的程式碼會建立一個輕量的 client 介面,把這三個問題一起處理掉。我們也會把它跟 LangGraph 的 StateGraph 串起來,示範一個「ReAct 風格」的代理流程:模型決定要呼叫 MCP 提供的工具、客戶端執行工具、把結果寫回狀態、模型再根據結果決定下一步。這是 AG Day 14 學過的 ReAct 模式延伸到 MCP 生態的版本,也是後續多代理架構(AG Day 29 起)的基礎。
原理/觀念
Client 介面的責任
MCP 客戶端的核心責任有四個:握手(initialize 與 notifications/initialized)、能力發現(tools/list、resources/list、prompts/list)、訊息路由(把 request 送到正確的 server、把 response 配對回發起者)、資源清理(連線結束時關閉 stdin 並等待子行程退出)。前兩項昨天已經走過,今天要處理的是後兩項,特別是「同一個 host 對接多個 server」的情境。
實務上的設計是:為每個 server 建立一個獨立的 client 物件,各自管理自己的連線、心跳與訊息配對;host 內部再用一個「client 集合」來管理所有 client 的生命週期。這種「每連線一物件」的模式跟 AG Day 8 學過的「每工具一函式」是同樣的工程紀律:把責任切細,讓除錯與測試都更單純。
命名空間:避免工具名稱碰撞
當兩個 server 都提供 echo 工具時,客戶端必須能區分它們。最直覺的方法是用「server 名稱當前綴」:把來自 research-agent-basic 的 echo 重新命名為 basic__echo,把來自 sql-explorer 的 echo 重新命名為 sql__echo。模型看到的是兩個完全不同的工具,呼叫時不會混淆。
這個設計的代價是:模型看到的工具名稱會比 server 原本提供的長一截,且 server 的更新可能會讓前綴改變。為了降低這個代價,請把前綴策略寫在 client 的設定檔裡,而不是寫死在程式裡。常見的做法是用 server name 去掉常見前綴(例如 research-agent-basic 變成 basic),再以兩個底線 __ 串接工具名稱。
工具動態註冊到 LangGraph
LangGraph 的 StateGraph 允許我們在建立圖的時候動態加入節點與邊。當客戶端發現一個新的 server 時,可以:第一步,呼叫 client.list_tools() 取得工具清單;第二步,把每個工具包成一個 LangGraph 的 ToolNode 節點(AG Day 15 學過);第三步,把工具名稱加進模型的工具清單裡。當 server 離線時,反向操作即可。
這套「動態發現、動態註冊」的設計讓代理能即時適應 MCP 生態的變化。今天我們會把這套機制用一個簡化版實作出來:啟動時列出所有 server 提供的工具、把它們預先註冊進 LangGraph 流程;之後若要新增 server,只要重啟流程即可(更完整的熱插拔留到 AG Day 30 的多代理通訊主題)。
ReAct + MCP:標準代理迴圈
AG Day 14 我們用 LangGraph 的 create_react_agent 實作了 ReAct 代理。今天的版本與當時的差別是「工具來源從專案內函式換成 MCP 工具」。模型看到的工具描述、參數 schema、回傳格式都由 MCP server 提供;客戶端負責把模型輸出的 tool_call 翻譯成 tools/call 訊息並送給對應 server。
這條轉譯鏈不長,但每一段都要正確:工具名稱要對應到正確的 client、參數要通過 schema 驗證、回傳要拆解成模型看得懂的內容物件。我們會在實作段落示範這條鏈的每一環,特別強調錯誤處理——當 MCP 工具回傳錯誤時,模型看到的訊息應該是結構化的錯誤,而不是把例外原封不動丟回去。
完整實作
接下來在 research-agent/src/research_agent/ 底下新增 mcp_client.py,把 MCP 客戶端包成可被 LangGraph 使用的介面。我們也會擴充昨天的 basic_server.py,加上一個具備兩段相加功能的工具,讓今天的範例能展示「工具真的會被執行」。
第一步,先把我們需要的依賴安裝好。FastMCP 與官方 mcp 套件昨天已經裝過,這裡再確認一次相依版本:
cd research-agent
uv add mcp
uv lock
段落說明:mcp 套件是官方 SDK,客戶端的介面都在這個套件裡(ClientSession、StdioServerParameters、stdio_client 等)。FastMCP 是伺服端的高階介面,但客戶端通常直接用 mcp.client.* 的低階介面,以保留最大的控制權。兩個套件可以共存於同一個專案,不會衝突。
第二步,建立 MCP 客戶端的基礎介面。我們把「啟動子行程、握手、列出工具」這條鏈包成一個類別,讓之後串接 LangGraph 時只需要操作這個類別:
"""research-agent/src/research_agent/mcp_client.py
MCP 客戶端的基礎介面:啟動子行程、握手、列出工具與資源。
"""
from dataclasses import dataclass, field
from pathlib import Path
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client
@dataclass
class MCPServerSpec:
"""描述一個 MCP server 的啟動方式。"""
name: str
command: str
args: list[str] = field(default_factory=list)
env: dict[str, str] = field(default_factory=dict)
class MCPClient:
"""包裝單一 MCP server 的連線與工具清單。"""
def __init__(self, spec: MCPServerSpec) -> None:
self.spec = spec
self._tools: list[dict] = []
self._resources: list[dict] = []
self._prompts: list[dict] = []
@property
def tools(self) -> list[dict]:
return self._tools
@property
def resources(self) -> list[dict]:
return self._resources
@property
def prompts(self) -> list[dict]:
return self._prompts
async def connect(self) -> None:
"""啟動子行程、完成握手、列出所有能力。"""
params = StdioServerParameters(
command=self.spec.command,
args=self.spec.args,
env=self.spec.env or None,
)
self._stdio = stdio_client(params)
read, write = await self._stdio.__aenter__()
self._session = ClientSession(read, write)
await self._session.__aenter__()
await self._session.initialize()
tools = await self._session.list_tools()
resources = await self._session.list_resources()
prompts = await self._session.list_prompts()
self._tools = [
{"name": tool.name, "description": tool.description, "schema": tool.inputSchema}
for tool in tools.tools
]
self._resources = [
{"uri": res.uri, "name": res.name}
for res in resources.resources
]
self._prompts = [
{"name": p.name, "description": p.description, "arguments": p.arguments}
for p in prompts.prompts
]
async def call_tool(self, name: str, arguments: dict) -> str:
"""呼叫指定工具並回傳文字內容;不做例外轉換,呼叫端自行處理。"""
result = await self._session.call_tool(name, arguments)
parts: list[str] = []
for item in result.content:
# 簡化:只處理文字內容;圖片與其他型別依需求再擴充
if getattr(item, "type", None) == "text":
parts.append(item.text)
return "\n".join(parts)
async def read_resource(self, uri: str) -> str:
"""讀取資源並回傳文字內容。"""
result = await self._session.read_resource(uri)
return "\n".join(item.text for item in result.contents if hasattr(item, "text"))
async def close(self) -> None:
"""結束 session 與子行程。"""
await self._session.__aexit__(None, None, None)
await self._stdio.__aexit__(None, None, None)
def default_servers() -> list[MCPServerSpec]:
"""預設的 server 清單;之後可以用設定檔取代。"""
project = Path(__file__).resolve().parents[2]
return [
MCPServerSpec(
name="basic",
command="python",
args=[str(project / "src/research_agent/mcp_servers/basic_server.py")],
),
]
段落說明:connect() 用 async with 的 __aenter__/__aexit__ 來管理兩個 async context manager(stdio_client 與 ClientSession),這是官方 SDK 的慣例用法。default_servers() 故意只放昨天寫的 basic_server,方便今天測試;之後可以用 YAML 或 TOML 設定檔讀多個 server。請記得 connect() 與 close() 必須成對呼叫,否則子行程會留在背景佔用資源。
第三步,把 MCP 客戶端接進 LangGraph。我們沿用 AG Day 14 的 create_react_agent,差別在「工具是從 MCP 動態載入的」。先把客戶端啟動、把工具清單與名稱前綴處理好,再呼叫官方 API:
"""research-agent/src/research_agent/mcp_graph.py
把 MCP 客戶端整合進 LangGraph 的 ReAct 流程。
"""
from langgraph.prebuilt import create_react_agent
from langchain_core.tools import tool
from mcp_client import MCPClient, default_servers
def namespaced_name(server_name: str, tool_name: str) -> str:
"""把 server 名稱與工具名稱組合成唯一識別。"""
short = server_name.split("-")[-1] # research-agent-basic -> basic
return f"{short}__{tool_name}"
def build_langchain_tools(clients: dict[str, MCPClient]) -> list:
"""把 MCP 工具轉成 LangChain 的 StructuredTool。"""
tools = []
for server_name, client in clients.items():
for tool_def in client.tools:
captured_name = tool_def["name"]
captured_description = tool_def["description"]
captured_schema = tool_def["schema"]
@tool(captured_name, args_schema=captured_schema)
def _mcp_tool(**kwargs) -> str:
"""被 MCP 包裝後的 LangChain 工具;實際執行走客戶端。"""
return client.call_tool_sync(captured_name, kwargs)
tools.append(_mcp_tool)
return tools
async def build_agent():
clients = {spec.name: MCPClient(spec) for spec in default_servers()}
for client in clients.values():
await client.connect()
langchain_tools = build_langchain_tools(clients)
agent = create_react_agent(model="gpt-4o-mini", tools=langchain_tools)
return agent, clients
段落說明:build_langchain_tools() 用一個小技巧——把 captured_name、captured_description、captured_schema、client 在閉包裡複製一份給每個工具,避免 closure over loop variable 的常見錯誤。模型名稱在這裡先用一個示意字串,實務上請從環境變數 RESEARCH_AGENT_MODEL 讀取,與全系列保持一致。call_tool_sync() 是我們在 MCPClient 上加的同步版本(把 async 呼叫包成同步),讓 LangChain 工具介面能用。實務上你也可以改成全 async,但那需要把整個 agent 改成 async,本系列為求簡潔先用同步版本。
第四步,把同步呼叫的薄封裝補進 MCPClient。LangChain 工具介面預設是同步的,所以我們加一個同步入口:
"""research-agent/src/research_agent/mcp_client.py(節錄)
在 MCPClient 上新增同步介面,供 LangChain 工具使用。
"""
import asyncio
class MCPClient:
# ...(既有內容)
def call_tool_sync(self, name: str, arguments: dict) -> str:
"""同步呼叫工具;內部跑在一個新的 event loop。"""
return asyncio.run(self._call_tool_async(name, arguments))
async def _call_tool_async(self, name: str, arguments: dict) -> str:
return await self.call_tool(name, arguments)
段落說明:asyncio.run() 會建立新的 event loop,這在已經有 loop 運行的環境(例如 Jupyter)會衝突;正式產品請考慮用 asyncio.get_event_loop().run_until_complete() 或把整條鏈改成 async。今天的範例是命令列執行,asyncio.run() 是最簡潔的選擇。把 sync/async 兩種介面都提供,是 MCP 客戶端與既有同步框架整合的常見模式。
第五步,寫一個把整條鏈跑起來的入口腳本。從連線到呼叫一個真實的工具、把結果印出來:
"""research-agent/scripts/mcp_min/run_agent.py
啟動 MCP client、建 ReAct agent、執行一個任務。
"""
import asyncio
from mcp_client import MCPClient, default_servers
async def main() -> None:
clients = {spec.name: MCPClient(spec) for spec in default_servers()}
try:
for client in clients.values():
await client.connect()
basic = clients["basic"]
echo_tool = next(t for t in basic.tools if t["name"] == "echo")
print("可用的 MCP 工具:", [t["name"] for t in basic.tools])
result = await basic.call_tool("echo", {"text": "從 MCP client 呼叫 echo"})
print("echo 結果:", result)
result = await basic.call_tool("add", {"a": 21, "b": 21})
print("add 結果:", result)
finally:
for client in clients.values():
await client.close()
if __name__ == "__main__":
asyncio.run(main())
段落說明:這個入口腳本不依賴 LangGraph,只驗證「MCP 客戶端能正確啟動 server、列出工具、執行工具」。執行後你會看到類似以下的輸出:
uv run python scripts/mcp_min/run_agent.py
可用的 MCP 工具: ['echo', 'add']
echo 結果: 從 MCP client 呼叫 echo
add 結果: 42
段落說明:這份輸出代表客戶端真的呼叫到 server 並取回結果,可以作為接下來整合 LangGraph 的信心檢查點。Run agent.py 完成後,所有 client 都會在 finally 區塊被關閉,不留子行程殘留。
第六步,把錯誤處理補上。MCP 工具呼叫失敗時,客戶端應該回傳結構化錯誤,而不是讓例外一路冒到上層:
"""research-agent/src/research_agent/mcp_client.py(節錄)
把錯誤轉成結構化訊息,避免例外破壞整條代理流程。
"""
from typing import Any
class MCPClient:
# ...
def call_tool_sync(self, name: str, arguments: dict) -> str:
"""同步呼叫工具;錯誤轉成 JSON 字串回傳。"""
try:
return asyncio.run(self._call_tool_async(name, arguments))
except Exception as exc: # noqa: BLE001
return (
'{"ok": false, "error": "'
+ type(exc).__name__
+ '", "message": "'
+ str(exc).replace('"', "'")
+ '"}'
)
段落說明:把例外轉成 JSON 字串看起來有點醜,但對 LangChain 工具介面是務實做法:工具的回傳值會被原封不動送回模型,而模型對結構化錯誤比對 Python 例外更熟悉。例外訊息裡的雙引號被我們換成單引號,避免 JSON 解析錯誤;正式產品可以改用 json.dumps(...) 並對訊息做 escape,但那會犧牲可讀性,請依需求取捨。
第七步,做一個總驗證:我們同時啟動兩個 MCP server,並示範「命名空間前綴真的能避免碰撞」。這個場景對未來多代理架構很重要,今天先把工具準備好:
"""research-agent/scripts/mcp_min/multi_server.py
同時對接兩個 MCP server,並示範命名空間前綴的處理。
"""
import asyncio
from mcp_client import MCPClient, MCPServerSpec
async def main() -> None:
project_root = "."
specs = [
MCPServerSpec(name="basic-a", command="python",
args=[f"{project_root}/src/research_agent/mcp_servers/basic_server.py"]),
MCPServerSpec(name="basic-b", command="python",
args=[f"{project_root}/src/research_agent/mcp_servers/basic_server.py"]),
]
clients = {spec.name: MCPClient(spec) for spec in specs}
try:
for client in clients.values():
await client.connect()
for name, client in clients.items():
short = name.split("-")[-1]
prefixed = [(f"{short}__{t['name']}", t) for t in client.tools]
print(f"{name} 提供的工具(前綴後):{[n for n, _ in prefixed]}")
a_result = await clients["basic-a"].call_tool("echo", {"text": "from A"})
b_result = await clients["basic-b"].call_tool("echo", {"text": "from B"})
print("A 回應:", a_result)
print("B 回應:", b_result)
finally:
for client in clients.values():
await client.close()
if __name__ == "__main__":
asyncio.run(main())
段落說明:這個範例刻意同時啟動兩個內容完全相同的 server,讓命名空間議題變得明顯。實際的部署情境不會有兩個一模一樣的 server,但它們可能各自提供一個 search、fetch、query,名稱碰撞的機率非常高。把前綴當成設計紀律,模型看到的工具清單就會清楚區分;少了這條紀律,等到上線才發現兩個 server 都提供 query 時,就來不及改了。
常見錯誤與踩雷
第一個錯誤是忘記 await close()。若主程式在例外發生時沒有走進 finally 區塊,子行程就會留在背景。請一律用 try/finally 或 async with 來管理客戶端的生命週期,這是 asyncio 的基本紀律。
第二個錯誤是 connect() 之後忘了 initialize()。雖然 MCPClient 範例有呼叫,但若你抄了部分程式碼自己改,很容易漏掉這一步。握手失敗時伺服器會回 invalid_request,請先檢查 initialize 是否成功。
第三個錯誤是命名空間前綴寫死。把 basic 寫死在工具名稱裡,當 server 改名(例如從 basic 改成 core),所有 client 都會跟著失效。請從 spec.name 動態產生前綴,不要用字串字面值。
第四個錯誤是工具參數的型別轉換。MCP 的 inputSchema 用 JSON Schema 表示型別,但 LangChain 工具介面吃的是 Python 型別 hint。請在包裝工具時,確認 JSON Schema 的 type 與 Python 的型別一致。今天的範例刻意讓 add 用 int,是為了讓這個對應關係簡單;實際遇到 number/string/array 等型別時,請在客戶端加一層轉換。
第五個錯誤是 stdio 子行程的環境變數。當你的 MCP server 需要讀取 API key 或設定檔,請用 StdioServerParameters.env 傳入,而不是依賴父行程的環境變數;某些作業系統在子行程繼承時會過濾掉部分變數,顯式傳入是最安全的做法。
第六個錯誤是把例外原封不動送回模型。AG Day 7 學過「錯誤以回傳值呈現而非例外」,這條原則在 MCP 客戶端同樣適用。把 httpx.ConnectError 直接丟給模型,模型看不懂;轉成 {"ok": false, "error": "..."} 結構化訊息,模型才知道該重試或換策略。
第七個錯誤是訊息配對錯誤。同時對接多個 server 時,若客戶端沒有為每個連線維護獨立的 session,來自 server A 的 response 可能會被誤認到 server B 的 request。請嚴格遵守「每個 server 一個 session」的紀律。
效能與實務提醒
先談連線的啟動成本。每一個 MCP server 啟動時都要 import 套件、載入設定、執行握手;當你同時掛了五個 server,這段時間可能累積到數秒。實務上的兩個做法:把 server 清單啟動平行化(asyncio.gather());把啟動成本高的 server 設計成「常駐服務」(用 systemd 或 Docker 之類的行程管理器)而非每次對話都重新啟動。
再談訊息頻率。stdio 是一條序列化的訊息通道,每秒能處理的訊息數量取決於訊息大小與行程 IPC 效率。當你的代理在高頻呼叫工具(例如多輪 ReAct 一次執行數十次工具呼叫),請考慮把工具呼叫結果批次回傳,而不是一輪一次。規格允許在單一訊息裡放多筆請求,但實務上多數 SDK 還沒實作;今天的範例保留簡單的單次呼叫,未來若需要批次再擴充。
第三是快取。tools/list 的結果在 server 不升級的情況下可以快取很長一段時間。請在客戶端加上「除非 server name 改變否則重用清單」的機制,避免每次對話都重新 list。今天的範例沒有實作快取,是因為我們的測試只跑一輪;正式產品請把快取補起來,並提供強制重新整理的開關。
第四是觀測。每一次工具呼叫都應該被記錄:呼叫了哪個工具、傳了什麼參數、伺服器回什麼、耗時多少。把這些紀錄寫進 SQLite 的 events 表或追蹤平台(AG Day 35),是除錯與成本控管的基礎。今天的範例把結果印到 stdout 充當觀測,正式部署時請改成結構化紀錄。
第五是熱插拔與重連。當某個 server 中途掛掉,客戶端需要能偵測並重連。今天的範例沒有實作重連,因為那會讓程式碼膨脹到一個段落塞不下;正式產品請為每個 client 加上一個 watchdog,定期檢查 session 是否還活著,發現問題就重新走 connect()。這個主題會在 AG Day 31 長任務章節再次出現。
小結
今天我們把視角從 MCP 伺服器換到客戶端,並把客戶端接進研究助理的核心流程。重點整理如下:
- Client 介面:每個 server 一個 client 物件;connect/close 成對;handshake 在 connect 內完成。
- 命名空間:用 server 短名當前綴,把
echo變成basic__echo,避免多 server 工具碰撞。 - 動態註冊:客戶端啟動後把工具包成 LangChain StructuredTool,餵進
create_react_agent。 - 錯誤處理:例外轉成 JSON 結構化訊息,避免直接送回模型造成看不懂。
- 多 server 範例:同時對接兩個 server,示範命名空間前綴與訊息路由。
- 重連與觀測:今天先留下伏筆,AG Day 31 與 AG Day 35 會進一步處理。
到這裡,research-agent 已經具備「工具可插拔」的能力:新增 MCP server 只要改設定檔,不必動核心代理程式。這是從「自己寫工具」到「使用整個生態」的轉折點,也是後續多代理架構(AG Day 29 起)的基礎。
結語
今天我們走完了 MCP 章節的最後一塊拼圖:客戶端。從 AG Day 25 的概念拆解、AG Day 26 的第一個 FastMCP server、AG Day 27 的 resources 與 prompts,到今天的 client 與 LangGraph 整合,這四篇構成一個完整的 MCP 工程閉環。明天開始,我們會把視野從「單一代理」放大到「多代理協作」:把搜尋、檢索、寫報告拆成不同 worker,由 supervisor 統一調度。今天接好的 MCP 工具箱會在多代理架構裡扮演關鍵角色——不同 worker 可以共享同一組工具,也可以各自掛不同的 MCP server。
明天,我們會進入「AG Day 29 多代理架構:supervisor 與 worker」,把目前單一代理的 LangGraph 流程拆成「主管+工作者」的星狀拓樸。我們會用 supervisor 統一做路由決策,讓 worker 各自專心做一件事,並讓這個架構能在沒有 API 金鑰的離線模擬模式下先跑通,之後再把真實模型呼叫接上去。
延伸資源
- MCP Python SDK 官方說明:
https://github.com/modelcontextprotocol/python-sdk。客戶端介面、stdio 與其他傳輸的官方範例。 - Model Context Protocol 官方網站:
https://modelcontextprotocol.io/。方法清單、訊息格式、傳輸選項的權威來源。 - LangGraph 官方說明:
https://langchain-ai.github.io/langgraph/。ReAct、create_react_agent、ToolNode 的 API 與範例。 - LangChain StructuredTool 官方說明:
https://python.langchain.com/docs/modules/tools/。把 MCP 工具包成 LangChain 工具的方法。 - Python
asyncio官方說明:https://docs.python.org/3/library/asyncio.html。asyncio.gather()、async with與asyncio.run()的語意。
留言
張貼留言