将开发者知识 MCP 服务器与自定义 AI 智能体集成

本教程介绍如何将远程 Developer Knowledge MCP 服务器连接到使用 LangChain、LlamaIndex 和 AutoGen 构建的自定义 AI 智能体框架。

通过将自定义 AI 智能体连接到远程 Model Context Protocol (MCP) 服务器，您的智能体可以动态搜索和检索权威且最新的 Google Developers 开发者文档，其中涵盖 Google Cloud、Firebase、Android、Google Maps Platform、Chrome 等。

前提条件

在开始之前，请完成以下前提条件：

  1. 生成 Developer Knowledge API 密钥：按照 API 密钥设置指南中的说明，在您的 Google Cloud 云项目中启用 Developer Knowledge API 并创建 API 密钥。
  2. 生成 Gemini API 密钥：在 Google AI Studio 中创建 API 密钥以对 Gemini 模型进行身份验证，或者重复使用您的 Google Cloud API 密钥（如果已在该密钥上启用 Generative Language API）。

  3. 设置环境变量：在终端中将 API 密钥设置为环境变量：

    export DEVELOPERKNOWLEDGE_API_KEY="YOUR_API_KEY"
export GEMINI_API_KEY="YOUR_GEMINI_API_KEY"

  4. 安装 Python：验证您的开发环境是否使用 Python 3.10 或更高版本。

使用 LangChain 进行连接

LangChain 应用可以使用 langchain-mcp-adapters 软件包或标准 MCP 客户端传输连接到远程 HTTP MCP 服务器。

安装依赖项

安装 LangChain、Gemini 和 MCP 所需的 Python 软件包：

pip install langgraph langchain-google-genai langchain-mcp-adapters mcp

配置 LangChain 代理

以下示例将 Developer Knowledge MCP 服务器连接到由 Gemini 提供支持的 LangChain ReAct 代理。该脚本使用 streamable_http 传输实例化 MultiServerMCPClient，附加 X-Goog-Api-Key HTTP 标头，并将远程工具加载到 create_react_agent 中：

import asyncio
import os

from langchain_google_genai import ChatGoogleGenerativeAI
from langchain_mcp_adapters.client import MultiServerMCPClient
from langgraph.prebuilt import create_react_agent

async def run_langchain_agent():
    api_key = os.environ.get("DEVELOPERKNOWLEDGE_API_KEY")
    if not api_key:
        raise ValueError(
            "DEVELOPERKNOWLEDGE_API_KEY environment variable is required."
        )

    gemini_api_key = os.environ.get("GEMINI_API_KEY")
    if not gemini_api_key:
        raise ValueError("GEMINI_API_KEY environment variable is required.")

    # Configure remote MCP connection for Developer Knowledge
    client = MultiServerMCPClient(
        {
            "google-developer-knowledge": {
                "url": "https://developerknowledge.googleapis.com/mcp",
                "transport": "streamable_http",
                "headers": {
                    "X-Goog-Api-Key": api_key,
                },
            }
        }
    )
    tools = await client.get_tools()

    # Initialize Gemini model and ReAct agent
    llm = ChatGoogleGenerativeAI(
        model="gemini-2.5-flash",
        temperature=0,
        google_api_key=gemini_api_key,
    )
    agent = create_react_agent(llm, tools)

    # Run agent query
    response = await agent.ainvoke(
        {
            "messages": [
                (
                    "user",
                    "How do I list Google Cloud Storage buckets in Python?",
                )
            ]
        }
    )
    print("\nAgent Response:\n", response["messages"][-1].content)

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

使用 LlamaIndex 进行连接

LlamaIndex 支持通过工具规范集成 MCP 服务器，从而使检索代理能够在推理期间查询开发者知识文档。

安装依赖项

安装 LlamaIndex 核心软件包、Gemini LLM 和 MCP 工具集成：

pip install llama-index llama-index-tools-mcp llama-index-llms-google-genai

配置 LlamaIndex 代理

以下示例使用 BasicMCPClient 连接到远程 MCP 服务器，使用 McpToolSpec 对其进行封装，并将检索到的工具绑定到由 Gemini 提供支持的 LlamaIndex FunctionAgent：

import asyncio
import os

from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.google_genai import GoogleGenAI
from llama_index.tools.mcp import BasicMCPClient, McpToolSpec

async def run_llamaindex_agent():
    api_key = os.environ.get("DEVELOPERKNOWLEDGE_API_KEY")
    if not api_key:
        raise ValueError(
            "DEVELOPERKNOWLEDGE_API_KEY environment variable is required."
        )

    gemini_api_key = os.environ.get("GEMINI_API_KEY")
    if not gemini_api_key:
        raise ValueError("GEMINI_API_KEY environment variable is required.")

    # Connect to the remote Developer Knowledge MCP endpoint
    mcp_client = BasicMCPClient(
        "https://developerknowledge.googleapis.com/mcp",
        headers={"X-Goog-Api-Key": api_key},
    )
    mcp_tool_spec = McpToolSpec(client=mcp_client)
    tools = await mcp_tool_spec.to_tool_list_async()

    # Create LlamaIndex FunctionAgent with Gemini
    llm = GoogleGenAI(model="gemini-2.5-flash", api_key=gemini_api_key)
    agent = FunctionAgent(tools=tools, llm=llm)

    # Query the agent
    response = await agent.run(
        "What are the default limits and quotas for the Developer Knowledge"
        " API?"
    )
    print("\nAgent Response:\n", str(response))

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

使用 AutoGen 进行连接

Microsoft AutoGen 智能体可以使用 autogen-ext MCP 扩展程序与远程 MCP 服务器进行交互，从而实现多智能体协作，并访问 Google Developers 官方开发者文档。

安装依赖项

安装 AutoGen 核心和 MCP 扩展软件包：

pip install autogen-agentchat "autogen-ext[mcp,openai]"

配置 AutoGen 代理

以下示例使用 StreamableHttpServerParams 连接到远程 MCP 服务器，并使用由 Gemini 提供支持的 AutoGen AssistantAgent 注册开发者知识工具：

import asyncio
import os

from autogen_agentchat.agents import AssistantAgent
from autogen_agentchat.teams import RoundRobinGroupChat
from autogen_ext.models.openai import OpenAIChatCompletionClient
from autogen_ext.tools.mcp import StreamableHttpServerParams, mcp_server_tools

async def run_autogen_agent():
    api_key = os.environ.get("DEVELOPERKNOWLEDGE_API_KEY")
    if not api_key:
        raise ValueError(
            "DEVELOPERKNOWLEDGE_API_KEY environment variable is required."
        )

    gemini_api_key = os.environ.get("GEMINI_API_KEY")
    if not gemini_api_key:
        raise ValueError("GEMINI_API_KEY environment variable is required.")

    # Retrieve remote MCP tools for Developer Knowledge by using Streamable HTTP
    server_params = StreamableHttpServerParams(
        url="https://developerknowledge.googleapis.com/mcp",
        headers={"X-Goog-Api-Key": api_key},
    )
    dk_tools = await mcp_server_tools(server_params)

    # Initialize model client with Gemini and assistant agent
    model_client = OpenAIChatCompletionClient(
        model="gemini-2.5-flash",
        api_key=gemini_api_key,
        base_url="https://generativelanguage.googleapis.com/v1beta/openai/",
    )
    developer_assistant = AssistantAgent(
        name="google_docs_assistant",
        model_client=model_client,
        tools=dk_tools,
        system_message=(
            "You are a developer assistant specializing in Google APIs and"
            " SDKs. Use Developer Knowledge tools to retrieve verified"
            " documentation."
        ),
    )

    # Run query with team runner
    team = RoundRobinGroupChat([developer_assistant], max_turns=5)
    stream = team.run_stream(
        task=(
            "Explain how to configure Firebase Cloud Messaging topic"
            " subscriptions."
        )
    )

    async for message in stream:
        print(message)

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

代理集成最佳实践

使用开发者知识 MCP 服务器构建自定义 AI 智能体时，请遵循以下准则：

  • 优化搜索查询：在自然语言查询中包含具体的产品名称和相关关键字，而不是通用字词，因为 search_documents 和 answer_query 工具会根据查询语义进行匹配。
  • 管理上下文窗口令牌：使用 answer_query 从文档语料库中获取直接答案，并避免使用 get_documents 请求完整页面内容，除非 search_documents 中的代码段详细信息不足。
  • 保护 API 密钥：将 Developer Knowledge API 和 Gemini API 密钥存储在环境变量或标准 Secret Manager 系统中，切勿在代码中硬编码凭据。

后续步骤

如需详细了解开发者知识 MCP 服务器，请参阅以下资源：