Day 07 - 開放協定：MCP 與 OpenAPI 整合

18th鐵人賽 google adk mcp

seanlin040

團隊HIT靈機一動萬象成章

2026-09-05 01:40:10

141 瀏覽

分享至

Day 7 | 開放協定：用 MCP 與 OpenAPI 接上別人寫好的工具

讀完能做到：讓你的 agent 接上一個外部 MCP server（也能反過來把自己的工具包成 MCP server），並知道從一份 OpenAPI 規格自動生出一整組可呼叫的 API 工具。順便知道「能在筆電上跑」跟「能部署到生產環境」之間，MCP 有一道很多教學都沒講的門檻。

昨天自己寫工具，今天不用自己寫了

Day 6 講的是「自己動手做」——寫函式、包成 FunctionTool。但現實世界裡，你要接的往往不是自己寫的東西：可能是別人維護的檔案系統操作、資料庫查詢、第三方 SaaS API。每個都自己重寫一次工具，不但累，還要自己維護。今天的兩個主題，MCP 與 OpenAPI，處理的正是「怎麼把別人已經定義好的能力，變成你的 agent 可以呼叫的工具」這件事——只是走的是兩條完全不同的路。

Model Context Protocol：一個通訊協定，不是一個函式庫

先把最容易搞混的觀念釐清：MCP 是協定規格，ADK 是框架。McpToolset 做的事，是在 ADK 框架內實作 MCP 協定的客戶端那一半；反過來，如果你要用 Python 寫一個 MCP 伺服器，靠的是 model-context-protocol 這個獨立的函式庫，跟 ADK 本身沒有直接關係。這個區分之所以重要，是因為它決定了你要看哪份文件——「怎麼寫一個 MCP server」跟「怎麼在 ADK 裡接 MCP server」是兩件事。

MCP 遵循 client-server 架構，定義了資料（resources）、互動範本（prompts）、可執行功能（tools）怎麼被 server 端暴露、被 client 端（可能是 LLM host 應用，也可能是另一個 AI agent）消費。官方把它形容成「一個通用的連接機制，簡化 LLM 取得 context、執行動作、跟各種系統互動的方式」——換句話說，MCP 想解決的問題，跟 USB 想解決「每種周邊都要自己的連接器」是同一種問題。

ADK 底層用 FastMCP 處理協定的複雜細節，讓你多數時候只需要「裝飾一個函式」就能把它變成 MCP 工具——這是官方原話，強調的是 FastMCP 設計上刻意 Pythonic、刻意低摩擦。

這一頁真正的乾貨，其實在部署

多數教學讀完 MCP 介紹，就直接跳去展示怎麼接一個檔案系統 MCP server，然後結束。但如果你打算把 agent 真的部署出去，官方文件裡藏著一段更重要的內容——MCP 有幾個跟一般 REST API 完全不同的架構特性，理解不夠就會在部署階段撞牆：

MCP 是有狀態的持久連線，不是無狀態的 REST API。原始設計常假設 client 跟 server 是同機部署，這個假設在雲端多租戶環境下並不成立。
McpToolset 管理連線的生命週期，範例程式碼裡常見的 exit_stack 模式，就是確保 agent 結束時連線（以及可能的 server process）被正確關閉。
McpToolset 支援物件序列化（getstate/setstate），這是為了讓 agent 部署到 Cloud Run 或 GKE 這類受管環境時能保留 context。但要注意——agent 保留了 session state，並不代表 MCP 連線也一起被還原：process 重啟後，agent 會依需要重新初始化跟 MCP server 的連線。這個細節在寫「部署後斷線重連」的除錯文章時特別重要。

用法一：ADK 當 MCP 客戶端，接外部工具

這是最常見的整合模式——你的 agent 需要用某個已經存在、以 MCP 介面暴露能力的服務。核心是 McpToolset 類別，把它加進 agent 的 tools 列表就會自動處理：連線管理（本地 process 用 StdioConnectionParams，遠端服務用 SseConnectionParams）、向 server 查詢可用工具（list_tools）、把 MCP 工具 schema 轉成 ADK 相容的 BaseTool、代理實際呼叫（call_tool）、以及可選的 tool_filter 篩選要暴露哪些工具子集。

一個接本機檔案系統 MCP server 的最小範例：

# ./adk_agent_samples/mcp_agent/agent.py
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

TARGET_FOLDER_PATH = "/absolute/path/to/your/folder"

root_agent = LlmAgent(
    model='gemini-flash-latest',
    name='filesystem_assistant_agent',
    instruction='Help the user manage their files. You can list files, read files, etc.',
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command='npx',
                    args=["-y", "@modelcontextprotocol/server-filesystem", TARGET_FOLDER_PATH],
                ),
            ),
        )
    ],
)

值得注意的前提條件（Prerequisites 一節列的）：MCP 需要 Python 3.9 以上，很多社群 MCP server 是用 Node.js 套件分發、透過 npx 執行，所以本地開發環境要有 Node.js。

如果接的是一個長時間執行的 MCP 工具，McpToolset 還有一個容易被忽略的參數——progress_callback。這是 MCP 協定原生支援的進度推播機制，讓工具在跑的過程中就能即時回報進度，不用等整個工具跑完才知道狀態：

async def my_progress_callback(progress: float, total: float, message: str):
    print(f"Progress: {progress}/{total} - {message}")

toolset = McpToolset(connection_params=..., progress_callback=my_progress_callback)

這跟 Day 6 講的 LongRunningFunctionTool（靠 FunctionResponse 分次回應）是兩條不同的路——一個是 MCP 協定層的推播，一個是 ADK 框架層的中繼回應模式，長任務要接哪一種工具就對應用哪一套進度回報機制。

一個現成、企業級的 MCP server 範例：MCP Toolbox for Databases。 上面的檔案系統範例是最小示範，實務上你更可能接的是一個開源、自己部署、Google 官方支援的資料庫 MCP server——涵蓋 BigQuery、AlloyDB、Spanner、PostgreSQL、MongoDB、Neo4j 等近 40 種資料源：

from google.adk import Agent
from google.adk.tools.toolbox_toolset import ToolboxToolset

toolset = ToolboxToolset(server_url="http://127.0.0.1:5000")
root_agent = Agent(..., tools=[toolset])

它跟今天的 McpToolset 走的是同一套協定，差別只在於工具是別人（Google）已經寫好、你只要部署 server 就能用，不用自己刻。細節見 [[資料庫工具整合 (BigQuery, MCP Toolbox for Databases)]]。

用法二：ADK 當 MCP 伺服器，把自己的工具開放給別人

方向反過來：把既有的 ADK 工具（例如 load_web_page）包一層，變成任何標準 MCP client 都能呼叫的服務。這條路涉及三步：實作 server 的 @app.list_tools() handler（用 adk_to_mcp_tool_type 這個轉換工具，把 ADK 工具定義轉成 MCP schema）、實作 @app.call_tool() handler（收到呼叫請求後執行 ADK 工具的 .run_async()，把結果格式化成 MCP 相容的回應）、以及把整個 server 跑起來（通常是 stdio 模式）。

# my_adk_mcp_server.py
from mcp import types as mcp_types
from mcp.server.lowlevel import Server, NotificationOptions
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.load_web_page import load_web_page
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type

adk_tool_to_expose = FunctionTool(load_web_page)

這條路徑適合的場景是：你的團隊已經在 ADK 裡累積了一堆寫得很好的工具，而現在有其他非 ADK 的 agent（可能用 LangChain、可能是別家框架）也想用——與其重寫，不如把 ADK 工具包一層 MCP server，變成大家都能接的共用服務。

生產部署：一個容易忽略、卻會直接讓部署失敗的規則

官方標成 Critical Deployment Requirement：部署環境要求 agent 與 McpToolset 必須用同步（synchronous）方式定義。adk web 開發環境允許非同步建立 agent，但 Cloud Run、GKE、Agent Runtime 這些部署目標不行。

# ✅ 正確：部署用的同步定義
root_agent = LlmAgent(
    model='gemini-flash-latest',
    name='enterprise_assistant',
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command='npx',
                    args=['-y', '@modelcontextprotocol/server-filesystem', _allowed_path],
                ),
                timeout=5,
            ),
            tool_filter=['read_file', 'list_directory', 'search_files'],  # 生產環境務必收斂
        )
    ],
)

# ❌ 錯誤：非同步模式在部署環境行不通
async def get_agent():
    toolset = await create_mcp_toolset_async()
    return LlmAgent(tools=[toolset])

這是很多人在本地 adk web 跑得好好的、一部署就報錯的根本原因——本地開發時用的非同步寫法看起來更「乾淨」，卻正是部署環境不支援的那種寫法。

官方文件列了三種部署模式，各有取捨：

模式	適用情境	取捨
Self-Contained Stdio（把 MCP server 跟 agent 包進同一個 container）	npm 套件形式的 server（如 `@modelcontextprotocol/server-filesystem`）	簡單、process 隔離好、適合容器化，但不適合高規模部署——每個連線都是一個 process
Remote MCP Servers（Streamable HTTP）	需要可擴展的生產部署	基於網路、可支援多個 client，但要處理額外的網路基礎設施與認證複雜度
Sidecar（GKE）	GKE 部署，MCP server 以 sidecar 容器形式跟 agent 同 pod	介於前兩者之間

Production Deployment Checklist 裡最值得摘出來寫的幾條：用 tool_filter 限制暴露的功能（不要把 MCP server 全部工具都開放給 agent）、檔案系統類 MCP server 要用限制性路徑（範例直接示範 os.path.dirname(os.path.abspath(__file__)) 而不是任意路徑）、遠端連線要用認證 header、以及stdio 連線要監控記憶體——因為每個 stdio MCP server 都是獨立 process，高流量時記憶體會隨連線數線性成長。

不想把 server endpoint 寫死在程式碼裡？ Google Cloud Agent Registry 提供另一條路：在執行期用 API 動態查找已登記在治理目錄裡的 MCP server，拿到的直接是能用的 McpToolset，不用自己管連線字串：

from google.adk.integrations.agent_registry import AgentRegistry

registry = AgentRegistry(project_id=project_id, location="global")
mcp_server_name = f"projects/{project_id}/locations/global/mcpServers/YOUR_MCP_SERVER_ID"
my_mcp_toolset = registry.get_mcp_toolset(mcp_server_name=mcp_server_name)

適合企業內有多團隊共用一批已治理 MCP server 的場景——endpoint 換了不用改各團隊的程式碼。細節見 [[Agent Registry (Google Cloud, Preview)]]。

OpenAPI：另一條路，不需要協定，只需要一份規格

如果你要接的是一個有 OpenAPI（v3.x）規格的傳統 REST API，走 MCP 反而是繞遠路——ADK 提供 OpenAPIToolset，直接從規格生成整組可呼叫工具，不需要對方額外實作任何協定層。

from google.adk.agents import LlmAgent
from google.adk.tools.openapi_tool.openapi_spec_parser.openapi_toolset import OpenAPIToolset

toolset = OpenAPIToolset(spec_str=openapi_spec_json, spec_str_type="json")

my_agent = LlmAgent(
    name="api_interacting_agent",
    model="gemini-flash-latest",
    tools=[toolset],
)

運作邏輯：OpenAPIToolset 解析規格、解掉內部 $ref 參照，找出 paths 底下每個操作（GET/POST/PUT/DELETE），為每個操作生成一個 RestApiTool 實例。工具名稱來自規格的 operationId（轉成 snake_case，上限 60 字元；沒有 operationId 就依 method 與 path 自動命名），工具描述來自 summary/description。每個 RestApiTool 會動態生成 FunctionDeclaration（告訴 LLM 該怎麼呼叫）、在被呼叫時組出實際的 HTTP 請求（用 httpx 非同步執行）、處理認證（初始化 OpenAPIToolset 時設定的 auth_scheme/auth_credential 會自動套用到所有生成的工具）。

OpenAPIToolset 在 0.1.0 版本就有的功能——這是整個系列裡少數從 ADK 1.x 時代就存在、幾乎沒怎麼變過的機制，代表性夠穩，適合當「快速把內部既有 REST API 接進 agent」的預設選項。至於更細緻的五層認證策略（API Key、OAuth 等），留給 Day 28 的工具認證專題細講。