使用 MCP SDK 連至 Developer Knowledge MCP 伺服器

官方 Model Context Protocol (MCP) SDK 可讓您自己的應用程式和 AI 代理,透過程式輔助方式連線至 Developer Knowledge MCP 伺服器。建構自訂代理、後端服務或自己的 MCP 伺服器時,如果需要搜尋及擷取官方 Google Developers 開發人員說明文件,請使用 MCP SDK,而非設定現有的 IDE 或程式開發輔助。

本指南說明如何:

  • 安裝 Python 或 TypeScript 適用的 MCP 用戶端 SDK。
  • 使用 API 金鑰或應用程式預設憑證 (ADC),連至 Developer Knowledge MCP 伺服器。
  • 呼叫 search_documents、get_documents 和 answer_query 工具。
  • 處理驗證、配額和工具錯誤。

事前準備

開始之前,請先完成下列各節。

啟用 API 並建立憑證

在 Google Cloud 雲端專案中啟用 Developer Knowledge API,並為驗證方式建立憑證。如需逐步說明,請參閱「選擇驗證方式」。

本指南中的範例使用 API 金鑰。如要改用 ADC,請參閱「使用應用程式預設憑證進行驗證」。

安裝 MCP SDK

選取程式設計語言,安裝 MCP 用戶端 SDK:

Python

本指南使用第 2 版的 MCP Python SDK,需要 Python 3.10 以上版本。SDK 1.x 版使用不同的類別和函式名稱,因此這些範例不適用:

pip install --upgrade "mcp>=2"

TypeScript

本指南使用 MCP TypeScript SDK,搭配 Node.js 18 以上版本:

npm install @modelcontextprotocol/sdk

設定 API 金鑰

為避免在原始碼中硬式編碼 API 金鑰,請將金鑰儲存在環境變數中。將 YOUR_API_KEY 替換成您的 API 金鑰:

export DEVELOPERKNOWLEDGE_API_KEY="YOUR_API_KEY"

連線至伺服器

Developer Knowledge MCP 伺服器在下列端點使用串流 HTTP 傳輸:

https://developerknowledge.googleapis.com/mcp

如要驗證,請在每個要求的 X-Goog-Api-Key HTTP 標頭中傳遞 API 金鑰。請勿將 API 金鑰做為網址查詢參數傳遞。

以下範例會連線至伺服器,並列出可用工具:

Python

import asyncio
import os

from mcp import Client
from mcp.client.streamable_http import (
    create_mcp_http_client,
    streamable_http_client,
)

MCP_URL = "https://developerknowledge.googleapis.com/mcp"

async def main() -> None:
    headers = {"X-Goog-Api-Key": os.environ["DEVELOPERKNOWLEDGE_API_KEY"]}

    async with create_mcp_http_client(headers=headers) as http_client:
        transport = streamable_http_client(MCP_URL, http_client=http_client)
        async with Client(transport) as client:
            tools = await client.list_tools()
            names = ", ".join(tool.name for tool in tools.tools)
            print("Available tools:", names)

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

TypeScript

import { Client } from '@modelcontextprotocol/sdk/client/index.js';
import {
  StreamableHTTPClientTransport,
} from '@modelcontextprotocol/sdk/client/streamableHttp.js';

const apiKey = process.env.DEVELOPERKNOWLEDGE_API_KEY;
if (!apiKey) {
  throw new Error('Set the DEVELOPERKNOWLEDGE_API_KEY variable.');
}

const transport = new StreamableHTTPClientTransport(
  new URL('https://developerknowledge.googleapis.com/mcp'),
  { requestInit: { headers: { 'X-Goog-Api-Key': apiKey } } },
);

const client = new Client({ name: 'my-docs-agent', version: '1.0.0' });
await client.connect(transport);

const { tools } = await client.listTools();
console.log('Available tools:', tools.map((tool) => tool.name).join(', '));

await client.close();

如果連線成功,輸出內容會列出這三項工具:

Available tools: search_documents, answer_query, get_documents

本指南中的其餘範例,都會在本節的已連線用戶端中執行。TypeScript 範例使用頂層 await,因此請以 ES 模組的形式執行 (例如使用 npx tsx index.ts)。

呼叫工具

每項工具都會以符合工具輸出結構定義的結構化內容傳回結果,因此您可以直接讀取欄位,不必剖析文字。如需完整的輸入和輸出結構定義,請參閱 MCP 參考資料。

搜尋說明文件

使用 query 字串呼叫 search_documents,擷取最相關的文件區塊。每個結果都包含 content 區塊和來源文件的 parent 資源名稱:

Python

search = await client.call_tool(
    "search_documents",
    {"query": "How do I create a Cloud Storage bucket?"},
)
results = search.structured_content["results"]
for chunk in results[:3]:
    print(f"{chunk['parent']}\n{chunk['content'][:200]}...\n")

TypeScript

const search = await client.callTool({
  name: 'search_documents',
  arguments: { query: 'How do I create a Cloud Storage bucket?' },
});
const { results } = search.structuredContent as {
  results: { parent: string; id: string; content: string }[];
};
for (const chunk of results.slice(0, 3)) {
  console.log(`${chunk.parent}\n${chunk.content.slice(0, 200)}...\n`);
}

擷取完整文件

如果區塊的背景資訊不足,請將一或多個 parent 值傳遞至 get_documents,以擷取每個網頁的完整 Markdown 內容。單次通話最多可要求 20 份文件。由於多個區塊可能來自同一份文件,請先移除重複的 parent 值:

Python

names = list(dict.fromkeys(chunk["parent"] for chunk in results[:3]))
get = await client.call_tool("get_documents", {"names": names})
for document in get.structured_content["documents"]:
    print(f"{document['title']} ({document['uri']})")
    print(document["content"])

TypeScript

const names = [...new Set(results.slice(0, 3).map((c) => c.parent))];
const get = await client.callTool({
  name: 'get_documents',
  arguments: { names },
});
const { documents } = get.structuredContent as {
  documents: {
    name: string;
    uri: string;
    title: string;
    content: string;
  }[];
};
for (const document of documents) {
  console.log(`${document.title} (${document.uri})`);
  console.log(document.content);
}

完整文件可能很大,因此只有在搜尋區塊不夠時,才呼叫 get_documents。詳情請參閱「最佳化脈絡視窗和權杖用量」。

生成答案

呼叫 answer_query,根據開發人員知識語料庫生成答案。結果包含 answerText 和用於生成結果的 references:

Python

answer = await client.call_tool(
    "answer_query",
    {"query": "How do I create a Cloud Storage bucket?"},
)
print(answer.structured_content["answerText"])
for reference in answer.structured_content["references"]:
    print(reference)

TypeScript

const answer = await client.callTool({
  name: 'answer_query',
  arguments: { query: 'How do I create a Cloud Storage bucket?' },
});
const { answerText, references } = answer.structuredContent as {
  answerText: string;
  references: string[];
};
console.log(answerText);
console.log(references.join('\n'));

references 欄位包含可傳遞至 get_documents 的文件資源名稱。同一份文件可能會重複出現,因此請先移除重複項目,再擷取文件。answer_query 工具的配額低於其他工具。詳情請參閱「配額與限制」。

使用應用程式預設憑證進行驗證

您可以使用應用程式預設憑證 (ADC) 進行驗證,不必使用 API 金鑰。在本機開發期間,ADC 會搭配您的使用者憑證運作;程式碼在 Google Cloud 上執行時,則會搭配服務帳戶運作。

使用 ADC 時,請在每個要求中傳送下列標頭:

  • Authorization:具有 https://www.googleapis.com/auth/cloud-platform 範圍的 Bearer OAuth 2.0 存取權杖。
  • X-Goog-User-Project:已啟用 Developer Knowledge API 且配額計費的專案 ID。您的帳戶必須具備這項專案的 serviceusage.services.use 權限。

存取權杖會過期,因此長時間執行的用戶端必須重新整理權杖。下列範例使用 Google 驗證程式庫,在每個要求中加入目前的權杖。

  1. 按照「選擇 OAuth 憑證類型」中的「ADC」ADC分頁標籤步驟,為本機開發設定 ADC。

  2. 將 GOOGLE_CLOUD_PROJECT 環境變數設為已啟用 Developer Knowledge API 的專案。範例會在 X-Goog-User-Project 標頭中傳送這個專案 ID:

    export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
    
  3. 安裝您所用語言的 Google Authenticator 程式庫:

    Python

    pip install --upgrade google-auth requests
    

    TypeScript

    npm install google-auth-library
    
  4. 使用 ADC 建立用戶端:

    Python

    import os
    
    import google.auth
    import google.auth.transport.requests
    import httpx2  # Installed automatically with mcp>=2
    from mcp import Client
    from mcp.client.streamable_http import (
        create_mcp_http_client,
        streamable_http_client,
    )
    
    MCP_URL = "https://developerknowledge.googleapis.com/mcp"
    
    class GoogleCredentialsAuth(httpx2.Auth):
        """Adds a current ADC access token and quota project to requests."""
    
        def __init__(self, project_id: str) -> None:
            self.credentials, _ = google.auth.default(
                scopes=["https://www.googleapis.com/auth/cloud-platform"]
            )
            self.project_id = project_id
    
        def auth_flow(self, request):
            if not self.credentials.valid:
                self.credentials.refresh(
                    google.auth.transport.requests.Request()
                )
            request.headers["Authorization"] = (
                f"Bearer {self.credentials.token}"
            )
            request.headers["X-Goog-User-Project"] = self.project_id
            yield request
    
    async def main() -> None:
        auth = GoogleCredentialsAuth(os.environ["GOOGLE_CLOUD_PROJECT"])
        async with create_mcp_http_client(auth=auth) as http_client:
            transport = streamable_http_client(
                MCP_URL, http_client=http_client
            )
            async with Client(transport) as client:
                ...
    

    TypeScript

    import { GoogleAuth } from 'google-auth-library';
    import { Client } from '@modelcontextprotocol/sdk/client/index.js';
    import {
      StreamableHTTPClientTransport,
    } from '@modelcontextprotocol/sdk/client/streamableHttp.js';
    
    const auth = new GoogleAuth({
      scopes: 'https://www.googleapis.com/auth/cloud-platform',
    });
    const projectId = process.env.GOOGLE_CLOUD_PROJECT;
    if (!projectId) {
      throw new Error('Set the GOOGLE_CLOUD_PROJECT variable.');
    }
    
    const transport = new StreamableHTTPClientTransport(
      new URL('https://developerknowledge.googleapis.com/mcp'),
      {
        // Add a current access token and the quota project to each request.
        fetch: async (url, init) => {
          const headers = new Headers(init?.headers);
          for (const [name, value] of await auth.getRequestHeaders()) {
            headers.set(name, value);
          }
          headers.set('X-Goog-User-Project', projectId);
          return fetch(url, { ...init, headers });
        },
      },
    );
    
    const client = new Client({ name: 'my-docs-agent', version: '1.0.0' });
    await client.connect(transport);
    

處理錯誤

Developer Knowledge MCP 伺服器的錯誤會透過兩種方式傳送至程式碼:

  • 要求錯誤:驗證、權限和配額錯誤會以 HTTP 錯誤狀態傳回,例如 400、403 或 429。SDK 會將這些視為例外狀況:Python 中的 MCPError 和 TypeScript 中的 StreamableHTTPError。在工具呼叫周圍擷取這些例外狀況。
  • 工具錯誤:如果工具執行但無法完成要求,結果會將 is_error (Python) 或 isError (TypeScript) 設為 true,且 content 欄位會包含問題的文字說明。讀取結構化內容前,請先檢查這個欄位。

以下範例會處理這兩種錯誤類型:

Python

from mcp import MCPError

try:
    result = await client.call_tool(
        "search_documents", {"query": "Cloud Storage buckets"}
    )
except MCPError as error:
    print(f"Request failed ({error.code}): {error.message}")
else:
    if result.is_error:
        print("Tool error:", result.content)
    else:
        results = result.structured_content["results"]

TypeScript

import {
  StreamableHTTPError,
} from '@modelcontextprotocol/sdk/client/streamableHttp.js';

try {
  const result = await client.callTool({
    name: 'search_documents',
    arguments: { query: 'Cloud Storage buckets' },
  });
  if (result.isError) {
    console.error('Tool error:', result.content);
  } else {
    const { results } = result.structuredContent as {
      results: { parent: string; id: string; content: string }[];
    };
  }
} catch (error) {
  if (error instanceof StreamableHTTPError) {
    console.error(`Request failed (${error.code}): ${error.message}`);
  } else {
    throw error;
  }
}

如果 answer_query 傳回配額錯誤 (429 RESOURCE_EXHAUSTED),請改用配額較高的 search_documents。如需重試策略 (包括指數輪詢),請參閱「錯誤處理和配額管理」。如要瞭解驗證錯誤的常見原因,請參閱「API 金鑰錯誤」和「OAuth 和 ADC 錯誤」。

後續步驟