Kết nối với máy chủ MCP Kiến thức cho nhà phát triển bằng SDK MCP

SDK Giao thức ngữ cảnh mô hình (MCP) chính thức cho phép các ứng dụng và tác nhân AI của riêng bạn kết nối với chương trình máy chủ MCP Kiến thức của nhà phát triển theo cách có lập trình. Sử dụng SDK MCP khi bạn đang xây dựng một tác nhân tuỳ chỉnh, một dịch vụ phụ trợ hoặc máy chủ MCP của riêng bạn cần tìm kiếm và truy xuất tài liệu chính thức dành cho nhà phát triển của Google, thay vì định cấu hình một IDE hoặc trợ lý lập trình hiện có.

Trong hướng dẫn này, bạn sẽ tìm hiểu cách:

  • Cài đặt SDK ứng dụng MCP cho Python hoặc TypeScript.
  • Kết nối với máy chủ MCP Kiến thức cho nhà phát triển bằng khoá API hoặc Thông tin xác thực mặc định của ứng dụng (ADC).
  • Gọi các công cụ search_documents, get_documents và answer_query.
  • Xử lý lỗi xác thực, hạn mức và công cụ.

Trước khi bắt đầu

Trước khi bắt đầu, hãy hoàn tất các phần sau.

Bật API và tạo thông tin xác thực

Bật Developer Knowledge API trong dự án trên đám mây của bạn và tạo thông tin xác thực cho phương pháp xác thực của bạn. Để xem hướng dẫn từng bước, hãy tham khảo bài viết Chọn phương pháp xác thực.

Các ví dụ trong hướng dẫn này sử dụng khoá API. Để sử dụng ADC, hãy tham khảo bài viết Xác thực bằng Thông tin xác thực mặc định của ứng dụng.

Cài đặt SDK MCP

Chọn ngôn ngữ lập trình để cài đặt SDK ứng dụng MCP:

Python

Hướng dẫn này sử dụng phiên bản 2 của MCP Python SDK, yêu cầu Python 3.10 trở lên. SDK phiên bản 1.x sử dụng các tên lớp và hàm khác nhau, vì vậy, những ví dụ này không hoạt động với phiên bản đó:

pip install --upgrade "mcp>=2"

TypeScript

Hướng dẫn này sử dụng MCP TypeScript SDK với Node.js 18 trở lên:

npm install @modelcontextprotocol/sdk

Đặt khoá API

Để tránh mã hoá cứng khoá API trong mã nguồn, hãy lưu trữ khoá đó trong một biến môi trường. Thay thế YOUR_API_KEY bằng khoá API của bạn:

export DEVELOPERKNOWLEDGE_API_KEY="YOUR_API_KEY"

Kết nối với máy chủ

Máy chủ MCP Kiến thức cho nhà phát triển sử dụng giao thức truyền tải HTTP có thể truyền trực tuyến tại điểm cuối sau:

https://developerknowledge.googleapis.com/mcp

Để xác thực, hãy truyền khoá API của bạn trong tiêu đề HTTP X-Goog-Api-Key trên mọi yêu cầu. Không truyền khoá API dưới dạng tham số truy vấn URL.

Ví dụ sau đây kết nối với máy chủ và liệt kê các công cụ có sẵn:

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

Nếu kết nối thành công, kết quả sẽ liệt kê 3 công cụ:

Available tools: search_documents, answer_query, get_documents

Các ví dụ còn lại trong hướng dẫn này chạy bên trong ứng dụng đã kết nối từ phần này. Các ví dụ về TypeScript sử dụng await cấp cao nhất, vì vậy, hãy chạy các ví dụ đó dưới dạng mô-đun ES (ví dụ: bằng npx tsx index.ts).

Gọi công cụ

Mỗi công cụ trả về kết quả dưới dạng nội dung có cấu trúc khớp với giản đồ đầu ra của công cụ, vì vậy, bạn có thể đọc trực tiếp các trường thay vì phân tích cú pháp văn bản. Để biết giản đồ đầu vào và đầu ra đầy đủ, hãy tham khảo tài liệu tham khảo về MCP.

Tìm tài liệu

Gọi search_documents bằng chuỗi query để truy xuất các đoạn tài liệu phù hợp nhất. Mỗi kết quả chứa đoạn content và tên tài nguyên parent của tài liệu chứa đoạn đó:

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

Truy xuất toàn bộ tài liệu

Nếu một đoạn không chứa đủ ngữ cảnh, hãy truyền một hoặc nhiều giá trị parent đến get_documents để truy xuất toàn bộ nội dung Markdown của mỗi trang. Bạn có thể yêu cầu tối đa 20 tài liệu trong một lệnh gọi. Vì một số đoạn có thể đến từ cùng một tài liệu, nên trước tiên, hãy xoá các giá trị parent trùng lặp:

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

Tài liệu đầy đủ có thể có kích thước lớn, vì vậy, chỉ gọi get_documents khi các khối tìm kiếm không đủ. Để tìm hiểu thêm, hãy tham khảo bài viết Tối ưu hoá cửa sổ ngữ cảnh và mức sử dụng mã thông báo.

Tạo câu trả lời

Gọi answer_query để tạo câu trả lời dựa trên Kho kiến thức dành cho nhà phát triển. Kết quả chứa answerText và references được dùng để tạo kết quả đó:

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

Trường references chứa tên tài nguyên tài liệu mà bạn có thể truyền đến get_documents. Cùng một tài liệu có thể xuất hiện nhiều lần, vì vậy, hãy xoá các bản sao trước khi truy xuất tài liệu. Công cụ answer_query có hạn mức thấp hơn các công cụ khác. Để tìm hiểu thêm, hãy tham khảo bài viết Hạn mức và giới hạn.

Xác thực bằng Thông tin xác thực mặc định của ứng dụng

Thay vì khoá API, bạn có thể xác thực bằng Thông tin xác thực mặc định của ứng dụng (ADC). ADC hoạt động với thông tin xác thực người dùng trong quá trình phát triển cục bộ và với tài khoản dịch vụ khi mã của bạn chạy trên Google Cloud.

Khi bạn sử dụng ADC, hãy gửi các tiêu đề sau trong mọi yêu cầu:

  • Authorization: mã truy cập OAuth 2.0 Bearer có phạm vi https://www.googleapis.com/auth/cloud-platform.
  • X-Goog-User-Project: mã nhận dạng của dự án đã bật Developer Knowledge API và được tính phí theo hạn mức. Tài khoản của bạn cần có quyền serviceusage.services.use đối với dự án này.

Mã truy cập sẽ hết hạn, vì vậy, các ứng dụng chạy trong thời gian dài phải làm mới mã truy cập. Các ví dụ sau đây sử dụng thư viện xác thực của Google để thêm mã thông báo hiện tại vào mỗi yêu cầu.

  1. Thiết lập ADC cho hoạt động phát triển tại máy bằng cách làm theo các bước trên thẻ ADC trong phần Chọn loại thông tin xác thực OAuth.

  2. Đặt biến môi trường GOOGLE_CLOUD_PROJECT thành dự án đã bật Developer Knowledge API. Các ví dụ này gửi mã dự án này trong tiêu đề X-Goog-User-Project:

    export GOOGLE_CLOUD_PROJECT="PROJECT_ID"

  3. Cài đặt thư viện xác thực của Google cho ngôn ngữ của bạn:

    Python

    pip install --upgrade google-auth requests

    TypeScript

    npm install google-auth-library

  4. Tạo ứng dụng bằng 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);

Xử lý lỗi

Các lỗi từ máy chủ MCP của Developer Knowledge sẽ đến mã của bạn theo hai cách:

  • Lỗi yêu cầu: lỗi xác thực, lỗi về quyền và lỗi hạn mức được trả về bằng trạng thái lỗi HTTP, chẳng hạn như 400, 403 hoặc 429. SDK sẽ đưa ra các trường hợp ngoại lệ sau: MCPError trong Python và StreamableHTTPError trong TypeScript. Nắm bắt những trường hợp ngoại lệ này trong các lệnh gọi công cụ.
  • Lỗi công cụ: khi một công cụ chạy nhưng không thể hoàn tất yêu cầu, kết quả sẽ có is_error (Python) hoặc isError (TypeScript) được đặt thành true và trường content chứa nội dung mô tả bằng văn bản về vấn đề. Hãy kiểm tra trường này trước khi đọc nội dung có cấu trúc.

Ví dụ sau đây xử lý cả hai loại lỗi:

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

Nếu answer_query trả về lỗi hạn mức (429 RESOURCE_EXHAUSTED), hãy quay lại search_documents vì phương thức này có hạn mức cao hơn. Đối với các chiến lược thử lại, bao gồm cả thuật toán thời gian đợi luỹ thừa, hãy tham khảo phần Xử lý lỗi và quản lý hạn mức. Để biết các nguyên nhân phổ biến gây ra lỗi xác thực, hãy tham khảo bài viết Lỗi khoá API và Lỗi OAuth và ADC.

Bước tiếp theo