เชื่อมต่อกับเซิร์ฟเวอร์ MCP ของ Developer Knowledge โดยใช้ MCP SDK

SDK อย่างเป็นทางการ ของ Model Context Protocol (MCP) ช่วยให้แอปพลิเคชันและ AI Agent ของคุณเองเชื่อมต่อกับเซิร์ฟเวอร์ MCP ของ Developer Knowledge ได้โดยใช้โปรแกรม ใช้ MCP SDK เมื่อสร้าง Agent ที่กำหนดเอง, บริการแบ็กเอนด์ หรือเซิร์ฟเวอร์ MCP ของคุณเองที่ต้องค้นหาและดึงข้อมูลเอกสารประกอบสำหรับนักพัฒนาแอปอย่างเป็นทางการของ Google แทนที่จะกำหนดค่า IDE หรือเครื่องมือช่วยเขียนโค้ดที่มีอยู่

ในคู่มือนี้ คุณจะได้เรียนรู้วิธีการทำสิ่งต่อไปนี้

  • ติดตั้ง MCP Client SDK สำหรับ Python หรือ TypeScript
  • เชื่อมต่อกับเซิร์ฟเวอร์ MCP ของ Developer Knowledge ด้วยคีย์ API หรือข้อมูลรับรองเริ่มต้นของแอปพลิเคชัน (ADC)
  • เรียกใช้เครื่องมือ search_documents, get_documents และ answer_query
  • จัดการการตรวจสอบสิทธิ์ โควต้า และข้อผิดพลาดของเครื่องมือ

ก่อนเริ่มต้น

ก่อนเริ่มต้น ให้ทำตามขั้นตอนในส่วนต่อไปนี้

เปิดใช้ API และสร้างข้อมูลเข้าสู่ระบบ

เปิดใช้ Developer Knowledge API ในโปรเจ็กต์ที่อยู่ในระบบคลาวด์ของ Google และสร้างข้อมูลเข้าสู่ระบบสำหรับวิธีการตรวจสอบสิทธิ์ ดูวิธีการทีละขั้นตอนได้ที่ หัวข้อ เลือกวิธีการตรวจสอบสิทธิ์

ตัวอย่างในคู่มือนี้ใช้คีย์ API หากต้องการใช้ ADC แทน โปรดดูหัวข้อ ตรวจสอบสิทธิ์ด้วยข้อมูลรับรองเริ่มต้นของแอปพลิเคชัน

ติดตั้ง MCP SDK

เลือกภาษาโปรแกรมเพื่อติดตั้ง MCP Client SDK

Python

คู่มือนี้ใช้ MCP Python SDK เวอร์ชัน 2 ซึ่ง ต้องใช้ Python 3.10 ขึ้นไป SDK เวอร์ชัน 1.x ใช้ชื่อคลาสและฟังก์ชันที่แตกต่างกัน ดังนั้นตัวอย่างเหล่านี้จึงใช้ไม่ได้กับ SDK เวอร์ชันดังกล่าว

pip install --upgrade "mcp>=2"

TypeScript

คู่มือนี้ใช้ MCP TypeScript SDK กับ Node.js 18 ขึ้นไป

npm install @modelcontextprotocol/sdk

ตั้งค่าคีย์ API

เก็บคีย์ API ไว้ในตัวแปรสภาพแวดล้อมเพื่อหลีกเลี่ยงการฮาร์ดโค้ดคีย์ API ในซอร์สโค้ด แทนที่ YOUR_API_KEY ด้วยคีย์ API ของคุณ

export DEVELOPERKNOWLEDGE_API_KEY="YOUR_API_KEY"

เชื่อมต่อกับเซิร์ฟเวอร์

เซิร์ฟเวอร์ MCP ของ Developer Knowledge ใช้ การรับส่งข้อมูล HTTP ที่สตรีมได้ ที่ปลายทางต่อไปนี้

https://developerknowledge.googleapis.com/mcp

หากต้องการตรวจสอบสิทธิ์ ให้ส่งคีย์ API ในส่วนหัว HTTP X-Goog-Api-Key ในทุกคำขอ อย่าส่งคีย์ API เป็นพารามิเตอร์การค้นหาของ URL

ตัวอย่างต่อไปนี้แสดงวิธีเชื่อมต่อกับเซิร์ฟเวอร์และแสดงรายการเครื่องมือที่พร้อมใช้งาน

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

หากเชื่อมต่อสำเร็จ เอาต์พุตจะแสดงรายการเครื่องมือ 3 รายการ

Available tools: search_documents, answer_query, get_documents

ตัวอย่างที่เหลือในคู่มือนี้จะทำงานภายในไคลเอ็นต์ที่เชื่อมต่อจากส่วนนี้ ตัวอย่าง TypeScript ใช้ await ระดับบนสุด ดังนั้นให้เรียกใช้เป็นโมดูล ES (เช่น ใช้ npx tsx index.ts)

เรียกใช้เครื่องมือ

เครื่องมือแต่ละรายการจะแสดงผลลัพธ์เป็นเนื้อหาที่มีโครงสร้างซึ่งตรงกับสคีมาเอาต์พุตของเครื่องมือ คุณจึงอ่านฟิลด์ได้โดยตรงแทนที่จะแยกวิเคราะห์ข้อความ ดูสคีมาอินพุตและเอาต์พุตทั้งหมดได้ในข้อมูลอ้างอิง MCP

ค้นหาเอกสารประกอบ

เรียกใช้ search_documents ด้วยสตริง query เพื่อดึงข้อมูลเอกสารประกอบที่เกี่ยวข้องมากที่สุด ผลลัพธ์แต่ละรายการจะมี 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 อย่างน้อย 1 ค่าไปยัง 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 เพื่อสร้างคำตอบที่อิงตาม คลังข้อมูล Developer Knowledge ผลลัพธ์จะมี 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 ได้ เอกสารเดียวกันอาจปรากฏมากกว่า 1 ครั้ง ดังนั้นให้ลบรายการที่ซ้ำกันออกก่อนที่จะดึงข้อมูลเอกสาร เครื่องมือ answer_query มีโควต้าต่ำกว่าเครื่องมืออื่นๆ ดูข้อมูลเพิ่มเติมได้ที่หัวข้อ โควต้าและขีดจำกัด

ตรวจสอบสิทธิ์ด้วยข้อมูลรับรองเริ่มต้นของแอปพลิเคชัน

คุณสามารถตรวจสอบสิทธิ์ด้วย ข้อมูลรับรองเริ่มต้นของแอปพลิเคชัน (ADC)แทนคีย์ API ADC ทำงานร่วมกับข้อมูลเข้าสู่ระบบของผู้ใช้ในระหว่างการพัฒนาในเครื่อง และทำงานร่วมกับบัญชีบริการเมื่อโค้ดทำงานบน Google Cloud

เมื่อใช้ ADC ให้ส่งส่วนหัวต่อไปนี้ในทุกคำขอ

  • Authorization: โทเค็นเพื่อการเข้าถึง OAuth 2.0 Bearer ที่มีขอบเขต https://www.googleapis.com/auth/cloud-platform
  • X-Goog-User-Project: รหัสของโปรเจ็กต์ที่เปิดใช้ Developer Knowledge API และมีการเรียกเก็บเงินสำหรับโควต้า บัญชีของคุณต้องมีสิทธิ์ serviceusage.services.use ในโปรเจ็กต์นี้

โทเค็นเพื่อการเข้าถึงจะหมดอายุ ดังนั้นไคลเอ็นต์ที่ทำงานเป็นเวลานานจึงต้องรีเฟรชโทเค็น ตัวอย่างต่อไปนี้ใช้ไลบรารีการตรวจสอบสิทธิ์ของ Google เพื่อเพิ่มโทเค็นปัจจุบันลงในแต่ละคำขอ

  1. ตั้งค่า ADC สำหรับการพัฒนาในเครื่องโดยทำตามขั้นตอนในแท็บ ADC ในหัวข้อ เลือกประเภทข้อมูลเข้าสู่ระบบ OAuth

  2. ตั้งค่าตัวแปรสภาพแวดล้อม GOOGLE_CLOUD_PROJECT เป็นโปรเจ็กต์ที่เปิดใช้ Developer Knowledge API ตัวอย่างจะส่งรหัสโปรเจ็กต์นี้ในส่วนหัว X-Goog-User-Project ดังนี้

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

จัดการข้อผิดพลาด

ข้อผิดพลาดจากเซิร์ฟเวอร์ MCP ของ Developer Knowledge จะไปถึงโค้ดของคุณได้ 2 วิธี

  • ข้อผิดพลาดของคำขอ: ระบบจะแสดงข้อผิดพลาดในการตรวจสอบสิทธิ์ สิทธิ์ และโควต้า พร้อมสถานะข้อผิดพลาดของ HTTP เช่น 400, 403, หรือ 429 SDK จะแสดงข้อผิดพลาดเหล่านี้เป็นข้อยกเว้น ได้แก่ MCPError ใน Python และ StreamableHTTPError ใน TypeScript ดักจับข้อยกเว้นเหล่านี้รอบๆ การเรียกใช้เครื่องมือ
  • ข้อผิดพลาดของเครื่องมือ: เมื่อเครื่องมือทำงานแต่ไม่สามารถดำเนินการตามคำขอให้เสร็จสมบูรณ์ ผลลัพธ์จะมี is_error (Python) หรือ isError (TypeScript) ตั้งค่าเป็น true และ ฟิลด์ content จะมีคำอธิบายข้อความของปัญหา ตรวจสอบฟิลด์นี้ก่อนอ่านเนื้อหาที่มีโครงสร้าง

ตัวอย่างต่อไปนี้แสดงวิธีจัดการข้อผิดพลาดทั้ง 2 ประเภท

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 ซึ่งมีโควต้าสูงกว่า ดูกลยุทธ์การลองอีกครั้ง รวมถึง Exponential Backoff ได้ที่ หัวข้อการจัดการข้อผิดพลาดและการจัดการโควต้า ดูสาเหตุที่พบบ่อยของข้อผิดพลาดในการตรวจสอบสิทธิ์ได้ที่หัวข้อ ข้อผิดพลาดเกี่ยวกับคีย์ API และ ข้อผิดพลาดเกี่ยวกับ OAuth และ ADC

ขั้นตอนถัดไป