在本文中，我们的重点将是通过作为MCP客户端的ADK代理，利用现有的MCP服务器并借助Gemini LLM，实现使用外部MCP驱动工具所提供的能力来调用工具。在深入代码实现之前，我们将讨论ADK的核心概念。

什么是ADK——代理开发工具包？

代理开发工具包（Agent Development Kit，简称ADK）是一个开源的、以代码为先的Python工具包，用于构建、评估和部署智能AI代理。

ADK使开发人员能够创建代理工作流——从简单的单代理任务到复杂的多代理编排——所有这些都在一个模块化和可扩展的框架内完成。

ADK中的代理是什么？

代理（Agent）是一个自主的、自包含的执行单元，旨在实现特定目标。代理可以

1. 执行任务，
2. 与用户交互，
3. 利用外部工具，以及
4. 与其他代理协作以完成复杂的工作流。

核心代理类别

ADK提供了三种主要的代理类型来支持不同的应用场景：

• LLM代理（如LlmAgent、Agent）：利用大语言模型（LLMs）进行理解、推理、规划和行动——非常适合动态的、以语言驱动的任务。
• 工作流代理（如SequentialAgent、ParallelAgent、LoopAgent）：以可预测的模式编排其他代理，而不依赖于LLM进行流程控制——最适合结构化的、可重复的过程。
• 自定义代理：通过扩展BaseAgent来启用自定义逻辑、专门的工作流或独特的工具集成——非常适合高级的、定制化的解决方案。

在本文中，我们将使用带有MCPTools的LLM代理类型。

ADK中的工具是什么？

在ADK的上下文中，工具（Tool）代表授予AI代理的一种特定能力，使其能够执行操作并与外部世界进行交互，而不仅仅是进行基本的文本生成和推理。

工具通常是一个模块化的代码组件——例如Python函数、类方法，甚至是另一个代理——旨在执行定义好的任务。

代理如何使用工具？

代理通过函数调用机制动态地利用工具，其中LLM根据上下文进行推理，选择并使用生成的输入调用适当的工具，观察结果，并将输出集成到其下一个动作或响应中。

ADK中的工具类型

ADK支持多种类型的工具：

函数工具：专门为我们的应用程序的独特逻辑和工作流构建的自定义工具。

• 函数/方法：注册为工具的标准同步Python函数（def）或类方法。
• 代理作为工具：将专门的代理作为可调用工具在父代理中使用，以实现模块化行为。
• 长时间运行的函数工具：为异步或耗时操作设计的工具。

内置工具：框架中包含的预定义工具，用于执行网络搜索、代码执行或RAG（检索增强生成）等任务。

第三方工具：轻松集成来自流行生态系统（如LangChain或CrewAI）的工具。

架构

在这里我们将使用ADK。

实现

让我们通过将其分解为关键实现步骤，来深入了解如何使用ADK + MCP + Gemini AI构建这个管道。

先决条件

1. 已安装Python 3.8+
2. 通过API密钥访问Google Gemini生成式AI
3. 一个有效的SerpAPI密钥（用于获取实时航班数据）

步骤1：设置虚拟环境

使用以下命令安装依赖项

# Setup virtual environment (mac or Unix )
python -m venv venv && source venv/bin/active 
# Install agent development kit
pip install google-adk
# Install MCP server 
pip install mcp-flight-search
# Install GenAI Python SDK
pip install google-genai

google-sdk：谷歌用于构建代理（agents）的代理开发工具包（Agent Development Kit）。

google-genai：谷歌用于与生成式AI模型（如Gemini）进行交互的库。

mcp-flight-search：使用MCP库并通过SerpAPI获取航班信息的MCP服务器。

设置环境变量：

注意这里的区别：在ADK中，它不是GEMINI_API_KEY，而是GOOGLE_API_KEY。

export GOOGLE_API_KEY="your-google-api-key"
export SERP_API_KEY="your-serpapi-key"

步骤2：安装MCP服务器——mcp-flight-search

为了让Gemini能够与真实世界的API进行交互，我们将使用一个符合MCP规范的服务器。

对于本文，我们将使用mcp-flight-search——一个使用FastMCP构建的轻量级MCP服务器，它公开了一个使用SerpAPI搜索实时航班数据的工具。

验证已安装的MCP服务器包：https://pypi.org/project/mcp-flight-search/

# Install from PyPI ( Already Installed from Step 1)
pip install mcp-flight-search

步骤3：理解ADK作为MCP客户端

from google.adk.agents.llm_agent import LlmAgent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.tools.mcp_tool.mcp_toolset import MCPToolset, StdioServerParameters

LlmAgent——是ADK中的一个核心组件，充当我们应用程序的“思考”部分，它利用大型语言模型（LLM）的力量进行推理、理解自然语言、做出决策、生成响应以及与工具进行交互。

Runner（运行器）负责在代理（agent）的生命周期中协调各个组件之间的交互。Runner 使用内存实现（in-memory implementations）来处理工件（artifact）、会话（session）和内存（memory）服务，为代理执行提供了一个轻量级且独立的环境。

InMemorySessionService 是 ADK 的 SessionService 接口的一种实现，它将所有会话数据（如对话历史记录、状态和元数据）直接存储在应用程序的内存中。这意味着当应用程序停止或重新启动时，所有会话信息都会丢失。

当用户与 AI 代理交互时，会创建一个 Session 对象来跟踪对话。

MCPToolSet —— 是 ADK 中的一个类，它使我们的 AI 代理能够与外部 MCP 服务器建立连接。这些服务器会暴露一些工具（例如 API 或服务），代理可以利用这些工具来执行特定任务。通过使用 MCPToolset，代理可以无缝地发现、调用和管理这些外部工具。

StdioServerParameters 是一个配置类，用于指定代理应如何通过标准输入/输出流连接到 MCP 服务器。当 MCP 服务器是一个通过控制台进行通信的本地进程时，这尤其有用。

MCPToolSet 和 StdioServerParameters 是如何协同工作的？

当结合使用时，MCPToolset 和 StdioServerParameters 允许 ADK 代理执行以下操作：

1. 建立连接：使用 StdioServerParameters，定义启动 MCP 服务器进程所需的命令和参数。
2. 发现可用工具：MCPToolset 连接到 MCP 服务器，并检索代理可以使用的可用工具列表。
3. 将工具集成到代理中：将发现的工具适配为与 ADK 代理兼容的格式，以便在代理执行期间无缝调用。
4. 管理连接生命周期：MCPToolset 处理与 MCP 服务器的连接的建立和拆除，确保资源得到适当管理。

步骤4：连接到 MCP 服务器

StdioServerParameters 使用 MCPToolSet 定义了 MCP 配置，以异步方式列出和监听。

# --- Step 1: Get tools from MCP server ---
async def get_tools_async():
    """Gets tools from the Flight Search MCP Server."""
    print("Attempting to connect to MCP Flight Search server...")
    server_params = StdioServerParameters(
        command="mcp-flight-search",
        args=["--connection_type", "stdio"],
        env={"SERP_API_KEY": os.getenv("SERP_API_KEY")},
    )
    
    tools, exit_stack = await MCPToolset.from_server(
        connection_params=server_params
    )
    print("MCP Toolset created successfully.")
    return tools, exit_stack

步骤5：使用ADK创建代理

如上所述，我们正在使用的是Llm代理，它是应用程序的“思考”部分。

# --- Step 2: Define ADK Agent Creation ---
async def get_agent_async():
    """Creates an ADK Agent equipped with tools from the MCP Server."""
    tools, exit_stack = await get_tools_async()
    print(f"Fetched {len(tools)} tools from MCP server.")
    
    # Create the LlmAgent matching the example structure
    root_agent = LlmAgent(
        model=os.getenv("GEMINI_MODEL", "gemini-2.5-pro-preview-03-25"),
        name='flight_search_assistant',
        instruction='Help user to search for flights using available tools based on prompt. If return date not specified, use an empty string for one-way trips.',
        tools=tools,
    )
    
    return root_agent, exit_stack

步骤6：将代理创建、会话管理与运行器（Runner）的编排进行集成

async def async_main():
    # Create services
    session_service = InMemorySessionService()
    # Create a session
    session = session_service.create_session(
        state={}, app_name='flight_search_app', user_id='user_flights'
    )
    # Define the user prompt
    query = "Find flights from Atlanta to Las Vegas 2025-05-05"
    print(f"User Query: '{query}'")
    
    # Format input as types.Content
    content = types.Content(role='user', parts=[types.Part(text=query)])
    # Get agent and exit_stack
    root_agent, exit_stack = await get_agent_async()
    # Create Runner
    runner = Runner(
        app_name='flight_search_app',
        agent=root_agent,
        session_service=session_service,
    )
    print("Running agent...")
    events_async = runner.run_async(
        session_id=session.id, 
        user_id=session.user_id, 
        new_message=content
    )
    # Process events
    final_content = None
    async for event in events_async:
        print(f"Event received: {event}")
    
    # Always clean up resources
    print("Closing MCP server connection...")
    await exit_stack.aclose()
    print("Cleanup complete.")

步骤7：演示

MCP客户端中的用户查询：

query = "Find flights from Atlanta to Las Vegas 2025-05-05"

使用标准日志的演示

使用调试日志的演示

将 MCP 与 ADK 集成的关键注意事项

1. MCP 与 ADK 的对比

• MCP 是一种开放协议，用于标准化 AI 模型与外部工具和数据源的交互方式。
• ADK 是一个基于 Python 的框架，用于构建和部署 AI 代理。
• MCPToolset：作为 MCP 和 ADK 之间的桥梁，使 ADK 代理能够使用 MCP 服务器暴露的工具。

要在 Python 中构建 MCP 服务器，可使用 model-context-protocol 库。

2. 工具类型与集成

• ADK 工具：专为在 ADK 代理中直接使用而设计的 Python 对象（如 BaseTool、FunctionTool）。
• MCP 工具：由 MCP 服务器暴露的功能，MCPToolset 会对其进行适配，以便在 ADK 代理中使用。
• 第三方工具：像 LangChain 和 CrewAI 这样的库提供的工具，可通过 LangchainTool 和 CrewaiTool 等包装器集成到 ADK 中。

3. 异步架构

• ADK 和 MCP Python 库均基于 Python 的 asyncio 框架构建。
• 工具实现和服务器处理程序应为异步（async def），以确保非阻塞操作。

4. MCP 中的有状态会话

• 与典型的无状态 REST API 不同，MCP 在客户端和服务器之间建立持久的有状态连接。
• 这种有状态性允许在交互之间保留上下文，但需要仔细的会话管理。

5. 部署注意事项

• MCP 连接的持久性可能会给扩展和部署带来挑战，尤其是对于处理多个用户的远程服务器。
• 基础设施方面的考虑包括负载均衡和会话亲和性，以维持连接稳定性。

6. 在 ADK 中管理 MCP 连接

• MCPToolset 在 ADK 中管理 MCP 连接的生命周期。
• 使用 exit_stack 可确保在代理执行完成时正确终止连接。

故障排查

默认情况下，adk 库期望使用 GCP 项目 Vertex AI、位置和 VertexAI 配置。请确保使用 GOOGLE_API_KEY 而不是 GEMINI_API_KEY，因为错误消息可能不会明确指示缺少的环境变量。

解决方案：确保将变量设置为 GOOGLE_API_KEY。

ValueError: Missing key inputs argument! To use the Google AI API,provide (`api_key`) arguments. To use the Google Cloud API, provide (`vertexai`, `project` & `location`) arguments.

频繁出现 429 速率限制错误和 500 内部服务器错误。

解决方案：切换到 Gemini 2 Flash，因为“Gemini API 的‘免费层级’是通过 API 服务提供的，其速率限制较低，仅用于测试目的”。

google.genai.errors.ClientError: 429 RESOURCE_EXHAUSTED. {'error': {'code': 429, 'message': 'You exceeded your current quota, please check your plan and billing details. For more information on this error, head to: https://ai.google.dev/gemini-api/docs/rate-limits.', 'status': 'RESOURCE_EXHAUSTED', 'details': [{'@type': 'type.googleapis.com/google.rpc.QuotaFailure', 'violations': [{'quotaMetric': 'generativelanguage.googleapis.com/generate_content_free_tier_requests', 'quotaId': 'GenerateRequestsPerDayPerProjectPerModel-FreeTier', 'quotaDimensions': {'model': 'gemini-2.0-pro-exp', 'location': 'global'}, 'quotaValue': '25'}]}, {'@type': 'type.googleapis.com/google.rpc.Help', 'links': [{'description': 'Learn more about Gemini API quotas', 'url': 'https://ai.google.dev/gemini-api/docs/rate-limits'}]}, {'@type': 'type.googleapis.com/google.rpc.RetryInfo', 'retryDelay': '27s'}]}}
An error occurred during execution: 500 INTERNAL. {'error': {'code': 500, 'message': 'An internal error has occurred. Please retry or report in https://developers.generativeai.google/guide/troubleshooting', 'status': 'INTERNAL'}}

Gemini API 费用计费

Gemini 2.5 Pro 预览版——根据谷歌文档，Gemini API 的“免费层级”是通过 API 服务提供的，其速率限制较低，仅用于测试目的。Gemini API 的“付费层级”则具有更高的速率限制、附加功能以及不同的数据处理方式。

结论

在本文中，我们探讨了如何使用开源的 ADK，通过模型上下文协议（MCP）并借助 Google Gemini LLM 作为 MCP 客户端，来构建一个代理设置助手。