官方 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範圍的BearerOAuth 2.0 存取權杖。X-Goog-User-Project:已啟用 Developer Knowledge API 且配額計費的專案 ID。您的帳戶必須具備這項專案的serviceusage.services.use權限。
存取權杖會過期,因此長時間執行的用戶端必須重新整理權杖。下列範例使用 Google 驗證程式庫,在每個要求中加入目前的權杖。
將
GOOGLE_CLOUD_PROJECT環境變數設為已啟用 Developer Knowledge API 的專案。範例會在X-Goog-User-Project標頭中傳送這個專案 ID:export GOOGLE_CLOUD_PROJECT="PROJECT_ID"安裝您所用語言的 Google Authenticator 程式庫:
Python
pip install --upgrade google-auth requestsTypeScript
npm install google-auth-library使用 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 錯誤」。