Google 会使用 AI 技术将内容翻译成您偏好的语言。AI 翻译可能包含错误。

使用 MCP SDK 连接到开发者知识 MCP 服务器

借助官方 Model Context Protocol (MCP) SDK ，您自己的应用和 AI 智能体可以通过编程方式连接到开发者知识 MCP 服务器。如果您要构建需要搜索和检索 Google Developers 官方开发者文档的自定义智能体、后端服务或您自己的 MCP 服务器，而不是配置现有 IDE 或编码辅助，请使用 MCP SDK。

在本指南中，您将学习如何：

安装适用于 Python 或 TypeScript 的 MCP 客户端 SDK。
使用 API 密钥或应用默认凭据 (ADC) 连接到开发者知识 MCP 服务器。
调用 search_documents、get_documents 和 answer_query 工具。
处理身份验证、配额和工具错误。

准备工作

在开始之前，请完成以下部分。

启用 API 并创建凭据

在 Google Cloud 云项目中启用 Developer Knowledge API，并为您的身份验证方法创建凭据。如需获取分步说明，请参阅选择身份验证方法。

本指南中的示例使用 API 密钥。如需改用 ADC，请参阅使用应用默认凭证进行身份验证。

安装 MCP SDK

选择您的编程语言以安装 MCP 客户端 SDK：

Python

本指南使用 MCP Python SDK的第 2 版，该版本需要 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"

连接到服务器

开发者知识 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 身份验证库向每个请求添加当前令牌。

按照选择 OAuth 凭据类型中的 ADC标签页上的步骤，为本地开发设置 ADC。
将 GOOGLE_CLOUD_PROJECT 环境变量设置为已启用 Developer Knowledge API 的项目。这些示例在 X-Goog-User-Project 标头中发送此项目 ID：
```
export GOOGLE_CLOUD_PROJECT="PROJECT_ID"
```

为您的语言安装 Google 身份验证库：

Python

pip install --upgrade google-auth requests

TypeScript

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);

处理错误

开发者知识 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 错误。

后续步骤

在 MCP 参考文档中探索完整的工具架构。
在语料库参考文档中了解哪些产品已编入索引。
使用客户端库直接调用 Developer Knowledge API。

使用 MCP SDK 连接到开发者知识 MCP 服务器 使用集合让一切井井有条 根据您的偏好保存内容并对其进行分类。

准备工作

启用 API 并创建凭据

安装 MCP SDK

Python

TypeScript

设置 API 密钥

连接到服务器

Python

TypeScript

调用工具

搜索文档

Python

TypeScript

检索完整文档

Python

TypeScript

生成答案

Python

TypeScript

使用应用默认凭证进行身份验证

Python

TypeScript

Python

TypeScript

处理错误

Python

TypeScript

后续步骤

使用 MCP SDK 连接到开发者知识 MCP 服务器