Skip to content

将 ADK 智能体配置为 MCP 服务器

你可以通过在 MCP 服务器中托管你的 ADK 智能体能力,将其暴露给外部 MCP 客户端(例如 Antigravity、Claude Code 或自定义智能体)。 主要有两种实现方式:

  • 暴露整个智能体: 通过简单的一行转换,将完整的多轮智能体推理和内部工具执行封装进一个服务器。

  • 暴露单个工具: 手动构建一个轻量级 MCP 服务器,来封装特定的、独立的 ADK 工具(例如某个 FunctionTool),而不包含智能体的推理循环。

开始使用

最强大的方式是暴露整个 LlmAgent。借助 to_mcp_server() 工具函数,你可以把你的智能体转换成一个标准的 FastMCP 服务器。这使外部客户端能够与该智能体的完整认知能力及其内部工具集进行交互。

from google.adk.agents import LlmAgent
from google.adk.tools.load_web_page import load_web_page
from google.adk.tools.mcp_tool import to_mcp_server

# 1. 定义你的 ADK 智能体
agent = LlmAgent(
    model="gemini-flash-latest", 
    name="web_reader_agent", 
    instruction="为用户获取并总结网页内容。", 
    tools=[load_web_page], 
)

# 2. 将智能体转换成 MCP 服务器
app = to_mcp_server(agent)

if __name__ == "__main__":
    # 将智能体作为标准的 stdio MCP 服务器运行
    app.run()

暴露单个 ADK 工具

如果你只想暴露单个能力(例如某个特定的 FunctionTool)而不包含完整的智能体推理循环,就必须手动构建一个 MCP 服务器。

前提条件: 你必须在你的 ADK 环境中安装 MCP 服务器库:

pip install mcp

实现步骤:

  1. 初始化工具: 实例化你想要暴露的 ADK 工具,例如:FunctionTool(load_web_page)。
  2. 列出工具的处理器: 实现 MCP 服务器的 @app.list_tools() 处理器来公布该工具。你使用 google.adk.tools.mcp_tool.conversion_utils 中的 adk_to_mcp_tool_type 工具函数,把 ADK 工具定义转换成 MCP 模式格式。
  3. 调用工具的处理器: 实现 @app.call_tool() 处理器来接收客户端请求。该处理器必须判断请求是否匹配你封装的工具,执行 ADK 工具的 .run_async() 方法(传入 tool_context=None),并把响应格式化为符合 MCP 的结构,例如 mcp.types.TextContent。

使用 ADK 工具构建 MCP 服务器

  1. 为你的 MCP 服务器创建一个新的 Python 文件,例如:my_adk_mcp_server.py。
  2. 用以下代码在你的新文件中实现服务器逻辑。下面的脚本搭建了一个暴露 ADK load_web_page 工具的 MCP 服务器。
import asyncio
import json
import os
from dotenv import load_dotenv

# MCP 服务器导入
from mcp import types as mcp_types
from mcp.server.lowlevel import Server, NotificationOptions
from mcp.server.models import InitializationOptions
import mcp.server.stdio 

# ADK 工具导入
from google.adk.tools.function_tool import FunctionTool
from google.adk.tools.load_web_page import load_web_page 
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type

load_dotenv()

# 1. 准备 ADK 工具
adk_tool_to_expose = FunctionTool(load_web_page)

# 2. 创建 MCP 服务器实例
app = Server("adk-tool-exposing-mcp-server")

# 3. 实现 list_tools 处理器
@app.list_tools()
async def list_mcp_tools() -> list[mcp_types.Tool]:
    mcp_tool_schema = adk_to_mcp_tool_type(adk_tool_to_expose)
    return [mcp_tool_schema]

# 4. 实现 call_tool 处理器
@app.call_tool()
async def call_mcp_tool(name: str, arguments: dict) -> list[mcp_types.Content]:
    if name == adk_tool_to_expose.name:
        try:
            # 执行 ADK 工具
            adk_tool_response = await adk_tool_to_expose.run_async(
                args=arguments, 
                tool_context=None,
            )
            # 格式化为 MCP TextContent
            response_text = json.dumps(adk_tool_response, indent=2)
            return [mcp_types.TextContent(type="text", text=response_text)]
        except Exception as e:
            error_text = json.dumps({"error": str(e)})
            return [mcp_types.TextContent(type="text", text=error_text)]
    else:
        error_text = json.dumps({"error": f"Tool '{name}' not found."})
        return [mcp_types.TextContent(type="text", text=error_text)]

# 5. MCP 服务器运行器(Stdio)
async def run_mcp_stdio_server():
    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name=app.name,
                server_version="0.1.0",
                capabilities=app.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

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

连接与传输模式

通过这些 MCP 服务器暴露你的 ADK 能力时,你通常使用标准输入/输出连接(mcp.server.stdio)来运行它们。该连接允许作为父进程运行的外部客户端应用生成你的脚本,并通过 stdio 流与之直接通信,从而保持连接的自包含与隔离。

使用 ADK 智能体测试你的自定义 MCP 服务器

要测试你的自定义服务器,你需要构建一个充当客户端的 ADK 智能体。该智能体使用 McpToolset 与你刚创建的服务器脚本建立连接。

  1. 在一个新目录(例如 ./adk_agent_samples/mcp_client_agent/)中设置你的智能体。创建一个 agent.py 文件,并在其旁边包含一个 __init__.py,使其可被发现。
from google.adk.agents import LlmAgent
from google.adk.tools.mcp_tool import McpToolset
from google.adk.tools.mcp_tool.mcp_session_manager import StdioConnectionParams
from mcp import StdioServerParameters

# 重要:请提供你之前构建的服务器脚本的绝对路径
MCP_SERVER_SCRIPT = "/path/to/your/my_adk_mcp_server.py"

root_agent = LlmAgent(
    model='gemini-flash-latest',
    name='web_reader_mcp_client_agent',
    instruction="使用 'load_web_page' 工具获取用户提供的 URL 内容。",
    tools=[
        McpToolset(
            connection_params=StdioConnectionParams(
                server_params=StdioServerParameters(
                    command='python3', 
                    args=[MCP_SERVER_SCRIPT], 
                )
            )
        )
    ],
)
  1. 在终端中导航到你的智能体的父目录:
cd ./adk_agent_samples
adk web
  1. 打开 ADK Web UI 并选择 web_reader_mcp_client_agent。
  2. 用如下提示测试连接:加载 "https://example.com" 的内容。

Google Cloud Genmedia 的 MCP 服务器

Genmedia 服务的 MCP 工具 是一组开源 MCP 服务器,让你能够把 Google Cloud 生成式媒体服务(例如 Imagen、Veo、Chirp 3 HD 音色和 Lyria)集成到你的 AI 应用中。

智能体开发工具包 (ADK) 和 Genkit 为这些 MCP 工具提供了内置支持,让你的 AI 智能体能够有效地编排生成式媒体工作流。有关实现指导,请参阅 ADK 示例智能体和 Genkit 示例。