iT邦幫忙

2026 iThome 鐵人賽

DAY 16
0
AI Engineering

AI Agent 系統開發 30 天系列 第 16 篇

用 MCP 接入外部工具伺服器

  • 分享至 

  • xImage
  •  

我們在之前的 讓 LangGraph Agent 使用工具查詢外部資料 這篇的 Tool 是用 @tool 寫在 Agent 專案裡的 Python 函式,執行也發生在同一行程內。這種寫法在工具單純時很直接,但若工具屬於獨立業務系統(例如訂單退貨),或是由 Go、TypeScript 等其他語言撰寫,就無法直接在 Python 行程中執行。過去往往得為每個工具單獨開 HTTP API 或透過 subprocess 呼叫外部程式,並在 Agent 專案裡重複手寫參數 Schema 與請求封裝;只要工具介面一改,所有依賴它的 Agent 專案都得同步修改並重新部署。

https://ithelp.ithome.com.tw/upload/images/20260927/20111896FTBkq3YU88.png

MCP(Model Context Protocol)將「工具宣告與執行」從 Agent 專案中剝離,改由工具端運行獨立的 MCP 伺服器。Agent 啟動時透過標準 JSON-RPC 協定查詢可用工具清單與參數 Schema(tools/list),直接交給模型;當模型決定呼叫時,再將請求轉發給伺服器執行(tools/call)。

這樣一來,工具的程式碼與 Schema 只需在後端維護一份;無論工具端是用 Python、Go 還是 TypeScript 開發,Agent 都不再需要把呼叫邏輯寫死在專案裡,只要透過標準協定連線就能動態載入。

對應的範例程式碼位於 ai-agent-sample/langgraph/langgraph-mcp-tool。

第一步:用 FastMCP 建立獨立的工具伺服器

要提供 MCP 工具,最直接的方式是使用官方 Python SDK 內建的 FastMCP。我們建立一個獨立的 server.py,專門提供訂單退貨狀態查詢:

# server.py
from mcp.server.fastmcp import FastMCP

# 建立名為 OrderService 的 MCP 伺服器
mcp = FastMCP("OrderService")

RETURN_RECORDS = {
    "A-1024": {
        "order_id": "A-1024",
        "status": "processing",
        "reason": "商品尺寸不合",
        "eta": "2026-08-05",
    },
}

@mcp.tool()
def get_return_status(order_id: str) -> dict:
    """查詢指定訂單的退貨進度與退款狀態。

    顧客詢問退貨處理進度、物流進度或退款狀況時使用。
    order_id 需為完整訂單編號,例如 A-1024。
    """
    record = RETURN_RECORDS.get(order_id.strip())
    if not record:
        return {
            "order_id": order_id,
            "status": "not_found",
            "message": "查無此訂單退貨紀錄",
        }
    return record

if __name__ == "__main__":
    mcp.run(transport="stdio")

這裡的 @mcp.tool() 與 LangChain 的 @tool 概念完全相同:函式名稱成為工具名稱,Docstring 成為模型判斷使用時機的說明文字,型別註解(order_id: str)則會被 FastMCP 自動轉換為 MCP 標準的 JSON Schema。

最後的 mcp.run(transport="stdio") 表示以標準輸入輸出(Standard Input / Output)作為溝通管道。

若要在接入 Agent 之前先確認工具伺服器是否正常運作,可以使用官方的視覺化除錯工具 MCP Inspector:

npx @modelcontextprotocol/inspector uv run python src/langgraph_mcp_tool/server.py

執行後終端機會印出本地 Web 介面網址(預設為 http://127.0.0.1:6274)並開啟瀏覽器。在介面中操作:

  1. 點擊頂部的 Connect 連線至伺服器。
  2. 切換至 Tools 頁籤,點擊 List Tools,會列出剛才定義的 get_return_status 工具與參數 Schema。
  3. 在 order_id 欄位輸入測試訂單編號(例如 A-1024),點擊 Run Tool,即可在下方看到伺服器回傳的 JSON 結果:
{
  "eta": "2026-08-05",
  "order_id": "A-1024",
  "reason": "商品尺寸不合",
  "status": "processing"
}

確認工具回傳無誤後即可關閉 Inspector。在正式執行 Agent 時,伺服器不需要手動常駐開啟;第二步中的 Agent 會在啟動時以子行程(Subprocess)自動執行它,並透過標準輸入輸出交換訊息。

![[Pasted image 20260910172207.png]]

第二步:在 LangGraph Agent 中接入外部工具

在 LangGraph 中,模型的 bind_tools() 與執行節點 ToolNode 收的都是 LangChain 的 BaseTool 清單,兩者不在意底層是本地函式還是遠端調用。

但 MCP 是通用的 JSON-RPC 協定,LangGraph 原生無法直接讀取 MCP 伺服器的宣告。我們使用 langchain-mcp-adapters 充當橋樑:它負責連上 MCP 伺服器取得工具宣告,並動態將其包裝成 LangGraph 認得的 BaseTool。

在 Agent 的主程式(agent.py)中,我們透過非同步函式取得工具並組裝狀態圖:

# agent.py
import asyncio
import os
import sys
from langchain_anthropic import ChatAnthropic
from langchain_core.messages import HumanMessage
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition

def build_graph(model, tools):
    model_with_tools = model.bind_tools(tools)

    def call_model(state: MessagesState):
        response = model_with_tools.invoke(state["messages"])
        return {"messages": [response]}

    builder = StateGraph(MessagesState)
    builder.add_node("agent", call_model)
    builder.add_node("tools", ToolNode(tools))

    builder.add_edge(START, "agent")
    builder.add_conditional_edges("agent", tools_condition)
    builder.add_edge("tools", "agent")

    return builder.compile()

async def main():
    # 1. 連接本機的 MCP 伺服器(Agent 在背後以子行程啟動 python server.py)
    client = MultiServerMCPClient(
        {
            "orders": {
                "transport": "stdio",
                "command": sys.executable,
                "args": ["src/langgraph_mcp_tool/server.py"],
            }
        }
    )

    # 2. 向伺服器動態取得工具清單(自動轉為 BaseTool 物件)
    tools = await client.get_tools()

    # 3. 組裝狀態圖(節點與邊的寫法與本地 Tool 100% 相同)
    model = ChatAnthropic(model_name="claude-haiku-4-5")
    graph = build_graph(model, tools)

    # 4. 執行任務
    query = "幫我查訂單 A-1024 的退貨進度"
    result = await graph.ainvoke({"messages": [HumanMessage(content=query)]})
    print(result["messages"][-1].content)

if __name__ == "__main__":
    asyncio.run(main())

注意到 LangGraph 的核心組裝(StateGraph、ToolNode、tools_condition)完全沒有變動。差異僅在於 tools 不再是專案內的 Python 函式清單,而是透過 await client.get_tools() 向外部伺服器動態取回。MCP 工具提供非同步調用介面,因此整張圖必須使用 await graph.ainvoke(...) 執行。

注意事項

MCP 協定在 2026-07-28 規格修訂中將核心全面轉向無狀態(Stateless)設計。需要釐清的是,MCP 過去所謂的「有狀態」,並非指工具伺服器會替你記住跨工具呼叫的業務資料(工具本質上仍是獨立執行的函式),而是指協定連線層與互動通道的狀態管理。

新舊規格的實際差異體現在兩個層面:

  • 移除連線握手與 Session ID:舊版要求用戶端在連線時先執行 initialize 與 notifications/initialized 雙向握手,協商協定版本與能力,並依賴伺服器配發的 Mcp-Session-Id 維持後續請求的脈絡。新版移除了強制握手與連線 Session,每個請求皆為自包含(Self-describing)的獨立呼叫,直接在 _meta 欄位帶上用戶端資訊與能力,並在標頭宣告協定版本(如 MCP-Protocol-Version: 2026-07-28)。若用戶端想在呼叫前探測伺服器支援的功能,改用選用的 server/discover RPC,不再強制綁定連線狀態。
  • 雙向長串流改為多輪請求(MRTR):舊版依賴持續開啟的雙向長串流(Held-open stream),當工具執行中途需要向使用者索取缺漏參數或確認(Elicitation)時,是由伺服器主動向用戶端發起反向請求。新版改為標準的 Request / Response 架構,透過 MRTR(Multi Round-Trip Requests) 取代反向串流:當工具需要補充資訊時,伺服器直接回傳 resultType: "input_required" 結束當次請求;用戶端備齊答案後,再將回覆包在 inputResponses 中重新發起呼叫,在純無狀態通訊下完成多輪人機互動。

若不同工具呼叫之間有相依的業務狀態(例如查單後發起退貨),應由工具回傳明確的識別碼或 handle(如 order_id),由模型作為參數傳入下一個工具,並由 LangGraph 的 State(如 MessagesState)管理上下文歷程。


上一篇
用 Prompt Caching 降低成本與延遲
下一篇
讓客服 Agent 透過 RAG 查詢退貨規定並回答
系列文
AI Agent 系統開發 30 天 共 21 篇
圖片
  熱門推薦
圖片
{{ item.channelVendor }} | {{ item.webinarstarted }} |
{{ formatDate(item.duration) }}
直播中

尚未有邦友留言

立即登入留言