Skip to content

MCP(模型上下文协议)工具集成

本指南将引导你了解将模型上下文协议 (MCP) 与 ADK 集成的两种方式。

什么是模型上下文协议 (MCP)?

模型上下文协议 (MCP) 是一种开放标准,旨在标准化大型语言模型 (LLM)(如 Gemini 和 Claude)与外部应用程序、数据源和工具的通信方式。可以将其视为一种通用连接机制,简化了 LLM 获取上下文、执行操作和与各种系统交互的方式。

MCP 遵循客户端 - 服务器架构,定义了数据(资源)、交互模板(提示)和可执行函数(工具)如何由 MCP 服务器公开并由 MCP 客户端(可能是 LLM 主机应用程序或 AI 智能体)使用。

本指南涵盖了两种主要的集成模式:

  1. 在 ADK 中使用现有 MCP 服务器: ADK 智能体充当 MCP 客户端,利用外部 MCP 服务器提供的工具。
  2. 通过 MCP 服务器暴露 ADK 工具: 构建一个包装 ADK 工具的 MCP 服务器,使其可被任何 MCP 客户端访问。

先决条件

在开始之前,确保你已设置以下内容:

  • 设置 ADK: 请按照快速入门中的标准 ADK[设置]()说明进行设置。
  • 安装/更新 Python: MCP 需要 Python 3.9 或更高版本。
  • 设置 Node.js 和 npx: 许多社区 MCP 服务器是以 Node.js 包的形式分发并使用npx运行的。如果你还没有安装 Node.js(包括 npx),请进行安装。详情请参见https://nodejs.org/en
  • 验证安装: 确认在激活的虚拟环境中adknpx都在你的 PATH 中:
# 两个命令都应该打印可执行文件的路径。
which adk
which npx

1. adk web 中将 MCP 服务器与 ADK 智能体一起使用(ADK 作为 MCP 客户端)

本节展示了两个将 MCP 服务器与 ADK 智能体一起使用的示例。这是最常见的集成模式。你的 ADK 智能体需要使用由现有服务提供的功能,该服务将自身公开为 MCP 服务器。

MCPToolset

这些示例使用 ADK 中的 MCPToolset 类作为到 MCP 服务器的桥梁。你的 ADK 智能体使用 MCPToolset 来:

  1. 连接: 建立与 MCP 服务器进程的连接。这可以是通过标准输入/输出(StdioServerParameters)通信的本地服务器或使用服务器发送事件(SseServerParams)的远程服务器。
  2. 发现: 查询 MCP 服务器以获取其可用工具(list_tools MCP 方法)。
  3. 适配: 将 MCP 工具模式转换为与 ADK 兼容的 BaseTool 实例。
  4. 公开: 将这些适配好的工具呈现给 ADK LlmAgent
  5. 代理调用:LlmAgent 决定使用这些工具之一时,MCPToolset 将调用(call_tool MCP 方法)转发到 MCP 服务器并返回结果。
  6. 管理连接: 处理与 MCP 服务器进程的连接生命周期,通常需要显式清理。

这些示例假设你使用 adk web 与 MCP 工具交互。如果你不使用 adk web,请参阅下面的"在 adk web 之外的自己的智能体中使用 MCP 工具"部分。

注意:使用 MCP 工具需要稍微不同的语法来导出包含 MCP 工具的智能体。目前正在开发一个更简单的在 ADK 中使用 MCP 的接口。

示例 1:文件系统 MCP 服务器

此示例演示了连接到提供文件系统操作的本地 MCP 服务器。

步骤 1:通过 MCPToolset 将 MCP 服务器附加到你的 ADK 智能体

./adk_agent_samples/mcp_agent/ 中创建 agent.py 并使用以下代码片段定义一个初始化 MCPToolset 的函数。

  • 重要:"/path/to/your/folder" 替换为你系统上实际文件夹的绝对路径
# ./adk_agent_samples/mcp_agent/agent.py
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters


async def create_agent():
  """Gets tools from MCP Server."""
  tools, exit_stack = await MCPToolset.from_server(
      connection_params=StdioServerParameters(
          command='npx',
          args=["-y",    # 命令的参数
            "@modelcontextprotocol/server-filesystem",
            # 重要提示!将下面的路径更改为你系统上的绝对路径。
            "/path/to/your/folder",
          ],
      )
  )

  agent = LlmAgent(
      model='gemini-2.0-flash',
      name='enterprise_assistant',
      instruction=(
          '帮助用户访问他们的文件系统'
      ),
      tools=tools,
  )
  return agent, exit_stack


root_agent = create_agent()

如果有多个 MCP 服务器,请创建一个通用退出栈并将其应用于所有 MCPToolset

# agent.py
from contextlib import AsyncExitStack
from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters, SseServerParams


async def create_agent():
  """从 MCP 服务器获取工具。"""
  common_exit_stack = AsyncExitStack()

  local_tools, _ = await MCPToolset.from_server(
      connection_params=StdioServerParameters(
          command='npx',
          args=["-y",    # 命令的参数
            "@modelcontextprotocol/server-filesystem",
            # 重要提示!将下面的路径更改为你系统上的绝对路径。
            "/path/to/your/folder",
          ],
      ),
      async_exit_stack=common_exit_stack
  )

  remote_tools, _ = await MCPToolset.from_server(
      connection_params=SseServerParams(
          # 重要提示!将下面的路径更改为你的远程 MCP 服务器路径
          url="https://your-mcp-server-url.com/sse"
      ),
      async_exit_stack=common_exit_stack
  )


  agent = LlmAgent(
      model='gemini-2.0-flash',
      name='enterprise_assistant',
      instruction=(
          '帮助用户访问他们的文件系统'
      ),
      tools=[
        *local_tools,
        *remote_tools,
      ],
  )
  return agent, common_exit_stack


root_agent = create_agent()

步骤 2:创建一个 init 文件

在与上面的 agent.py 相同的文件夹中创建一个 __init__.py 文件

# ./adk_agent_samples/mcp_agent/__init__.py
from . import agent

步骤 3:观察结果

从 adk_agent_samples 目录运行 adk web(确保你的虚拟环境已激活):

cd ./adk_agent_samples
adk web

成功的 MCPTool 交互将通过访问你的本地文件系统产生响应,如下所示:

MCP with ADK Web - FileSystem Example

示例 2:Google Maps MCP 服务器

这遵循相同的模式,但针对 Google Maps MCP 服务器。

步骤 1:获取 API 密钥并启用 API

按照 使用 API 密钥 的指示获取 Google Maps API 密钥。

在你的 Google Cloud 项目中启用 Directions API 和 Routes API。有关说明,请参阅 Google Maps Platform 入门 主题。

步骤 2:更新 create_agent

修改 agent.py 中的 create_agent 以连接到 Maps 服务器,通过 StdioServerParameters 的 env 参数传递你的 API 密钥。

# agent.py(根据需要修改 get_tools_async 和其他部分)

from google.adk.agents.llm_agent import LlmAgent
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters


async def create_agent():
  """从 MCP 服务器获取工具。"""

  tools, exit_stack = await MCPToolset.from_server(
      connection_params=StdioServerParameters(
          command='npx',
          args=["-y",
                "@modelcontextprotocol/server-google-maps",
          ],
          # 将 API 密钥作为环境变量传递给 npx 进程
          env={
              "GOOGLE_MAPS_API_KEY": google_maps_api_key
          }
      )
  )

  agent = LlmAgent(
      model='gemini-2.0-flash', # 根据需要调整
      name='maps_assistant',
      instruction='使用可用工具帮助用户进行地图和方向查询。',
      tools=tools,
  )
  return agent, exit_stack


root_agent = create_agent()

步骤 3:创建 init 文件

如果你已经从上面的示例 1 完成了这一步,请跳过此步骤。

在与上面的 agent.py 相同的文件夹中创建一个 __init__.py 文件

# ./adk_agent_samples/mcp_agent/__init__.py
from . import agent

步骤 4:观察结果

从 adk_agent_samples 目录运行 adk web(确保你的虚拟环境已激活):

cd ./adk_agent_samples
adk web

成功的 MCPTool 交互将产生一个带有路线计划的响应,如下所示:

MCP with ADK Web - Google Maps Example

示例 3:FastMCP 服务器

此示例演示了连接到提供数学运算(如加法)的远程 FastMCP 服务器。

步骤 0:将 FastMCP 服务器部署到 Cloud Run

#server.py
from fastmcp import FastMCP
import asyncio

mcp = FastMCP("FastMCP Demo Server")

@mcp.tool()
def add(a: int, b: int) -> int:
    """计算两个数字的和"""
    return a + b

if __name__ == "__main__":
    asyncio.run(mcp.run_sse_async(host="0.0.0.0", port=8080))
确保你的 MCP 服务器项目在根目录(如 ./fastmcp-demo)中有以下文件:

  • server.py:使用 FastMCP 的主要应用程序代码。

  • requirements.txt:列出 Python 依赖项。

    fastmcp
    asyncio
    

  • Procfile:告诉 Cloud Run 如何启动你的 Web 服务器。

    web: python server.py
    
    (注意:这假设你的 FastMCP 实例在 server.py 文件中命名为 mcp。如果你的文件名或实例名不同,请相应调整 server:mcp。)

从你的 FastMCP 服务器目录(如 ./fastmcp-demo)执行 Cloud Run 部署命令:

    gcloud run deploy fastmcp-demo \
        --source . \
        --region YOUR_REGION \
        --allow-unauthenticated

步骤 1:通过 MCPToolset 将 FastMCP 服务器附加到你的 ADK 智能体

./adk_agent_samples/fastmcp_agent/ 中创建 agent.py 并使用以下代码片段定义一个初始化 MCPToolset 的函数。

  • 重要: 将 Cloud Run 服务 URL 替换为你在上一步中部署的 URL。
# ./adk_agent_samples/fastmcp_agent/agent.py

import os
from contextlib import AsyncExitStack

import google.auth
from google.adk.agents import Agent
from google.adk.tools.tool_context import ToolContext
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams

_, project_id = google.auth.default()
os.environ.setdefault("GOOGLE_CLOUD_PROJECT", project_id)
os.environ.setdefault("GOOGLE_CLOUD_LOCATION", "us-central1")
os.environ.setdefault("GOOGLE_GENAI_USE_VERTEXAI", "True")


async def get_sum(a: int, b: int) -> int:
    """计算两个数字的和。

    Args:
        a: 数字
        b: 数字

    Returns:
        两个数字的和。
    """
    common_exit_stack = AsyncExitStack()

    tools, _ = await MCPToolset.from_server(
        connection_params=SseServerParams(
            url="https://fastmcp-demo-00000000000.us-central1.run.app/sse",
            project_id="YOUR-GCP-PROJECT-ID",
            location="us-central1",
        ),
        async_exit_stack=common_exit_stack
    )

    return await tools[0].run_async(
        args={
            "a": a,
            "b": b,
        },
        tool_context=None,
    )

root_agent = Agent(
    name="root_agent",
    model="gemini-2.0-flash",
    instruction="你是一个设计用来提供准确有用信息的 AI 助手。",
    tools=[get_sum],
)

步骤 2:创建一个 init 文件

在与上面的 agent.py 相同的文件夹中创建一个 __init__.py 文件

# ./adk_agent_samples/fastmcp_agent/__init__.py
from . import agent

步骤 3:观察结果

从 adk_agent_samples 目录运行 adk web(确保你的虚拟环境已激活):

cd ./adk_agent_samples
adk web

成功的交互将通过访问你的远程 FastMCP 服务器产生响应,如下所示:

FastMCP with ADK Web - 计算数字总和示例

2. 使用 ADK 工具构建 MCP 服务器(MCP 服务器公开 ADK)

此模式允许你包装 ADK 的工具并使其可用于任何标准 MCP 客户端应用程序。本节中的示例通过 MCP 服务器公开 load_web_page ADK 工具。

步骤概述

你将使用 model-context-protocol 库创建一个标准的 Python MCP 服务器应用程序。在此服务器中,你将:

  1. 实例化你想要公开的 ADK 工具(例如,FunctionTool(load_web_page))。
  2. 实现 MCP 服务器的 @app.list_tools 处理程序来公布 ADK 工具,使用 adk_to_mcp_tool_type 将 ADK 工具定义转换为 MCP 模式。
  3. 实现 MCP 服务器的 @app.call_tool 处理程序来接收来自 MCP 客户端的请求,识别请求是否针对你包装的 ADK 工具,执行 ADK 工具的 .run_async() 方法,并将结果格式化为符合 MCP 的响应(例如,types.TextContent)。

先决条件

在与 ADK 相同的环境中安装 MCP 服务器库:

pip install mcp

步骤 1:创建 MCP 服务器脚本

创建一个新的 Python 文件,例如 adk_mcp_server.py。

步骤 2:实现服务器逻辑

添加以下代码,它设置了一个公开 ADK load_web_page 工具的 MCP 服务器。

# adk_mcp_server.py
import asyncio
import json
from dotenv import load_dotenv

# MCP 服务器导入
from mcp import types as mcp_types # 使用别名以避免与 genai.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 # 示例 ADK 工具
# ADK <-> MCP 转换工具
from google.adk.tools.mcp_tool.conversion_utils import adk_to_mcp_tool_type

# --- 加载环境变量(如果 ADK 工具需要它们) ---
load_dotenv()

# --- 准备 ADK 工具 ---
# 实例化你想要公开的 ADK 工具
print("初始化 ADK load_web_page 工具...")
adk_web_tool = FunctionTool(load_web_page)
print(f"ADK 工具 '{adk_web_tool.name}' 已初始化。")
# --- ADK 工具准备结束 ---

# --- MCP 服务器设置 ---
print("创建 MCP 服务器实例...")
# 创建一个命名的 MCP 服务器实例
app = Server("adk-web-tool-mcp-server")

# 实现 MCP 服务器的 @app.list_tools 处理程序
@app.list_tools()
async def list_tools() -> list[mcp_types.Tool]:
  """MCP 处理程序,列出可用工具。"""
  print("MCP 服务器:收到 list_tools 请求。")
  # 将 ADK 工具的定义转换为 MCP 格式
  mcp_tool_schema = adk_to_mcp_tool_type(adk_web_tool)
  print(f"MCP 服务器:公布工具:{mcp_tool_schema.name}")
  return [mcp_tool_schema]

# 实现 MCP 服务器的 @app.call_tool 处理程序
@app.call_tool()
async def call_tool(
    name: str, arguments: dict
) -> list[mcp_types.TextContent | mcp_types.ImageContent | mcp_types.EmbeddedResource]:
  """MCP 处理程序,执行工具调用。"""
  print(f"MCP 服务器:收到 call_tool 请求,用于 '{name}',参数:{arguments}")

  # 检查请求的工具名称是否与我们包装的 ADK 工具匹配
  if name == adk_web_tool.name:
    try:
      # 执行 ADK 工具的 run_async 方法
      # 注意:tool_context 为 None,因为我们不在完整的 ADK Runner 调用中
      adk_response = await adk_web_tool.run_async(
          args=arguments,
          tool_context=None, # 这里没有可用的 ADK 上下文
      )
      print(f"MCP 服务器:ADK 工具 '{name}' 成功执行。")
      # 将 ADK 工具的响应(通常是字典)格式化为 MCP 格式。
      # 这里,我们在 TextContent 中将响应字典序列化为 JSON 字符串。
      # 根据特定 ADK 工具的输出和客户端需求调整格式化。
      response_text = json.dumps(adk_response, indent=2)
      return [mcp_types.TextContent(type="text", text=response_text)]

    except Exception as e:
      print(f"MCP 服务器:执行 ADK 工具 '{name}' 时出错:{e}")
      # 以 MCP 格式返回错误消息
      # 创建适当的 MCP 错误响应可能更为健壮
      error_text = json.dumps({"error": f"无法执行工具 '{name}':{str(e)}"})
      return [mcp_types.TextContent(type="text", text=error_text)]
  else:
      # 处理对未知工具的调用
      print(f"MCP 服务器:未找到工具 '{name}'。")
      error_text = json.dumps({"error": f"未实现工具 '{name}'。"})
      # 为简单起见,将错误作为 TextContent 返回
      return [mcp_types.TextContent(type="text", text=error_text)]

# --- MCP 服务器运行器 ---
async def run_server():
  """通过标准输入/输出运行 MCP 服务器。"""
  # 使用 MCP 库中的 stdio_server 上下文管理器
  async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
    print("MCP 服务器开始握手...")
    await app.run(
        read_stream,
        write_stream,
        InitializationOptions(
            server_name=app.name, # 使用上面定义的服务器名称
            server_version="0.1.0",
            capabilities=app.get_capabilities(
                # 定义服务器功能 - 查阅 MCP 文档了解选项
                notification_options=NotificationOptions(),
                experimental_capabilities={},
            ),
        ),
    )
    print("MCP 服务器运行循环已完成。")

if __name__ == "__main__":
  print("启动公开 ADK 工具的 MCP 服务器...")
  try:
    asyncio.run(run_server())
  except KeyboardInterrupt:
    print("\nMCP 服务器被用户停止。")
  except Exception as e:
    print(f"MCP 服务器遇到错误:{e}")
  finally:
    print("MCP 服务器进程退出。")
# --- MCP 服务器结束 ---

步骤 3:使用 ADK 测试你的 MCP 服务器

按照"示例 1:文件系统 MCP 服务器"中的相同说明创建一个 MCP 客户端。这次使用你上面创建的 MCP 服务器文件作为输入命令:

# ./adk_agent_samples/mcp_agent/agent.py

# ...

async def get_tools_async():
  """从文件系统 MCP 服务器获取工具。"""
  print("尝试连接到 MCP 文件系统服务器...")
  tools, exit_stack = await MCPToolset.from_server(
      # 使用 StdioServerParameters 进行本地进程通信
      connection_params=StdioServerParameters(
          command='python3', # 运行服务器的命令
          args=[
                "/absolute/path/to/adk_mcp_server.py"],
      )
  )

类似上面从你的终端执行智能体脚本(确保在你的环境中安装了必要的库,如 model-context-protocol 和 google-adk):

cd ./adk_agent_samples
python3 ./mcp_agent/agent.py

脚本将打印启动消息,然后等待 MCP 客户端通过其标准输入/输出连接到 adk_mcp_server.py 中的 MCP 服务器。现在任何符合 MCP 的客户端(如 Claude Desktop 或使用 MCP 库的自定义客户端)都可以连接到此进程,发现 load_web_page 工具并调用它。服务器将打印日志消息,指示收到的请求和 ADK 工具执行情况。参考 文档,使用 Claude Desktop 尝试。

adk web 之外的自己的智能体中使用 MCP 工具

如果以下情况符合你的需求,本节内容与你相关:

  • 你正在使用 ADK 开发自己的智能体
  • 且你使用 adk web
  • 且你通过自己的 UI 公开智能体

使用 MCP 工具需要与使用常规工具不同的设置,因为 MCP 工具的规范是从远程运行或在另一个进程中运行的 MCP 服务器异步获取的。

以下示例是从上面的"示例 1:文件系统 MCP 服务器"示例修改而来的。主要区别是:

  1. 你的工具和智能体是异步创建的
  2. 你需要正确管理退出栈,以便在与 MCP 服务器的连接关闭时正确销毁你的智能体和工具。
# agent.py(根据需要修改 get_tools_async 和其他部分)
# ./adk_agent_samples/mcp_agent/agent.py
import asyncio
from dotenv import load_dotenv
from google.genai import types
from google.adk.agents.llm_agent import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.artifacts.in_memory_artifact_service import InMemoryArtifactService # 可选
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, SseServerParams, StdioServerParameters

# 从父目录的 .env 文件加载环境变量
# 在使用环境变量(如 API 密钥)之前,将此放在靠前的位置
load_dotenv('../.env')

# --- 步骤 1:智能体定义 ---
async def get_agent_async():
  """创建一个配备 MCP 服务器工具的 ADK 智能体。"""
  tools, exit_stack = await MCPToolset.from_server(
      # 使用 StdioServerParameters 进行本地进程通信
      connection_params=StdioServerParameters(
          command='npx', # 运行服务器的命令
          args=["-y",    # 命令的参数
                "@modelcontextprotocol/server-filesystem",
                # 重要提示!将下面的路径更改为你系统上的绝对路径。
                "/path/to/your/folder"],
      )
      # 对于远程服务器,你会使用 SseServerParams:
      # connection_params=SseServerParams(url="http://remote-server:port/path", headers={...})
  )
  print(f"从 MCP 服务器获取了 {len(tools)} 个工具。")
  root_agent = LlmAgent(
      model='gemini-2.0-flash', # 根据可用性调整模型名称(如果需要)
      name='filesystem_assistant',
      instruction='使用可用工具帮助用户与本地文件系统交互。',
      tools=tools, # 向 ADK 智能体提供 MCP 工具
  )
  return root_agent, exit_stack

# --- 步骤 2:主执行逻辑 ---
async def async_main():
  session_service = InMemorySessionService()
  # 此示例可能不需要 Artifact 服务
  artifacts_service = InMemoryArtifactService()

  session = session_service.create_session(
      state={}, app_name='mcp_filesystem_app', user_id='user_fs'
  )

  # 提示:将查询更改为与你指定的文件夹相关的内容。
  # 例如,"列出 'documents' 子文件夹中的文件"或"读取 'notes.txt' 文件"
  query = "列出测试文件夹中的文件"
  print(f"用户查询:'{query}'")
  content = types.Content(role='user', parts=[types.Part(text=query)])

  root_agent, exit_stack = await get_agent_async()

  runner = Runner(
      app_name='mcp_filesystem_app',
      agent=root_agent,
      artifact_service=artifacts_service, # 可选
      session_service=session_service,
  )

  print("正在运行智能体...")
  events_async = runner.run_async(
      session_id=session.id, user_id=session.user_id, new_message=content
  )

  async for event in events_async:
    print(f"收到事件:{event}")

  # 关键清理:确保 MCP 服务器进程连接已关闭。
  print("正在关闭 MCP 服务器连接...")
  await exit_stack.aclose()
  print("清理完成。")

if __name__ == '__main__':
  try:
    asyncio.run(async_main())
  except Exception as e:
    print(f"发生错误:{e}")

关键考虑事项

使用 MCP 和 ADK 时,请记住以下几点:

  • 协议与库: MCP 是一种协议规范,定义通信规则。ADK 是用于构建智能体的 Python 库/框架。MCPToolset 通过在 ADK 框架中实现 MCP 协议的客户端端来桥接这两者。相反,在 Python 中构建 MCP 服务器需要使用 model-context-protocol 库。

  • ADK 工具与 MCP 工具:

    • ADK 工具(BaseTool, FunctionTool, AgentTool 等)是为在 ADK 的 LlmAgent 和 Runner 中直接使用而设计的 Python 对象。
    • MCP 工具是根据协议的模式由 MCP 服务器公开的能力。MCPToolset 使这些对 LlmAgent 看起来像 ADK 工具。
    • Langchain/CrewAI 工具是这些库中的特定实现,通常是简单的函数或类,缺乏 MCP 的服务器/协议结构。ADK 提供了一些互操作性的包装器(LangchainTool, CrewaiTool)。
  • 异步特性: ADK 和 MCP Python 库都严重基于 Python 的 asyncio 库。工具实现和服务器处理程序通常应该是异步函数。

  • 有状态会话(MCP): MCP 在客户端和服务器实例之间建立有状态的持久连接。这与典型的无状态 REST API 不同。

    • 部署: 这种有状态性可能会给扩展和部署带来挑战,特别是对于处理多个用户的远程服务器。原始 MCP 设计通常假定客户端和服务器位于同一位置。管理这些持久连接需要仔细的基础设施考虑(例如,负载均衡,会话亲和性)。
    • ADK MCPToolset: 管理这种连接生命周期。示例中显示的 exit_stack 模式对于确保连接(可能还有服务器进程)在 ADK 智能体完成时正确终止至关重要。

进一步资源