从零开始，亲手开发你的第一个AI大模型（三）Agent实战

本篇文章将进一步深入，重点讲解如何借助 ADK 中的 Agent，以 MCP 客户端的形式调用外部 MCP 工具的能力，配合 Gemini 大模型，实现智能 Agent 的工具链集成。话不多说，我们直接进入代码实现！

Python老猿

924人浏览 · 2025-08-18 11:01:18

Python老猿 · 2025-08-18 11:01:18 发布

📖说明

本篇文章是：从零开始，亲手开发你的第一个AI大模型！的最终篇。在第二篇的基础上配合 Gemini 大模型，实现智能 Agent 的工具链集成。

在我上一篇文章中，我们探讨了如何通过 Model Context Protocol（MCP）将 Gemini 大模型集成为 MCP 客户端，从而实现结构化、具备工具调用能力的智能交互，同时也介绍了 MCP 的核心理念、典型应用场景，并展示了相关演示。

而本篇文章将进一步深入，重点讲解如何借助 ADK 中的 Agent，以 MCP 客户端的形式调用外部 MCP 工具的能力，配合 Gemini 大模型，实现智能 Agent 的工具链集成。话不多说，我们直接进入代码实现！

🎯 我们的最终目标

我们将使用 Google 开源的 Agent Development Kit（ADK），结合 Gemini 大语言模型和 MCP 工具协议，从零构建一个智能航班查询 Agent，实现以下功能：

💡 当用户输入诸如“帮我查查从亚特兰大到拉斯维加斯 5 月 5 号的航班”这样的请求时，Agent 会自动调用 MCP 接口，查找实时航班信息，并以结构化方式返回给用户。

代码结构

我们将延续上一篇文章中使用的架构，并在此基础上，引入 ADK（Agent Development Kit）进行集成。

实现部分

接下来，我们将通过拆解关键步骤，逐步构建这个由 ADK + MCP + Gemini AI 组成的智能 Agent 流水线。

准备工作

在开始实现之前，请确保已完成以下环境配置：

已安装 Python 3.8 及以上版本
拥有 Google Gemini 生成式 AI 的 API Key（用于调用 Gemini 大模型）
有效的 SerpAPI 密钥（用于获取实时航班数据）

上述KEY如何申请？在本系列文章的第二篇均有介绍！

Step1：设置虚拟环境

我们首先需要创建并激活一个 Python 虚拟环境，然后安装所需的依赖库。

# 设置虚拟环境（适用于 macOS 或 Unix 系统）
python -m venv venv && source venv/bin/active 

# 安装 Agent Development Kit（代理开发工具包）
pip install google-adk

# 安装 MCP server 
pip install mcp-flight-search

# 安装 GenAI Python SDK
pip install google-genai

google-sdk：Google 推出的 Agent Development Kit，用于构建智能代理（Agent）。
google-genai：Google 提供的库，用于与生成式 AI 模型（如 Gemini）进行交互。
mcp-flight-search：一个基于 MCP 协议的服务端，利用 SerpAPI 通过 MCP 库实现航班搜索功能。

设置环境变量：

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

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

Step2:安装 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

Step3:理解 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 生命周期中协调各个组件之间的交互。它采用内存级实现来管理 artifact（工件）、session（会话）和 memory（记忆）服务，从而为 Agent 的运行提供一个轻量级、独立的执行环境。

InMemorySessionService 是 ADK 中 SessionService 接口的一个实现，它将所有会话数据（包括对话历史、状态信息和元数据）直接存储在应用的内存中。这意味着一旦应用停止或重启，所有会话信息都将丢失。

当用户与 AI Agent 进行交互时，系统会创建一个 Session 对象，用于跟踪整个对话过程。

MCPToolSet 是 ADK 中的一个类，用于让我们的 AI Agent 能够连接外部的 MCP 服务端。这些服务端通常会暴露一些工具（如 API 或服务），供 Agent 调用以完成特定任务。通过使用 MCPToolSet，Agent 可以实现对这些外部工具的自动发现、调用与管理，从而实现无缝集成。

StdioServerParameters 是一个配置类，用于指定 Agent 如何通过标准输入/输出流（stdin/stdout）与 MCP 服务端建立连接。当 MCP 服务端作为本地进程并通过控制台进行通信时，这个类尤其有用。

💡 MCPToolSet 与 StdioServerParameters 如何协同工作？

将 MCPToolSet 与 StdioServerParameters 结合使用时，ADK Agent 可以实现以下功能：

建立连接：通过 StdioServerParameters，定义启动 MCP 服务进程所需的命令及参数，从而初始化连接。
发现可用工具：MCPToolSet 连接 MCP 服务端，并获取该服务端所暴露的所有可用工具列表。
集成工具到 Agent：将发现的工具适配成 ADK Agent 可识别和调用的格式，实现运行过程中的无缝工具调用。
管理连接生命周期：MCPToolSet 负责 MCP 服务连接的建立与断开，确保资源得到妥善管理。

Step4:链接MCP服务器

StdioServerParameters 用于定义 MCP 的配置，使 MCPToolSet 能够以异步方式进行工具列表的获取与监听。

# --- Step 1: 从 MCP 服务端获取工具 ---
asyncdef get_tools_async():
    """从航班搜索 MCP 服务端获取工具。"""
    print("正在尝试连接 MCP 航班搜索服务端...")
    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 工具集创建成功。")
    return tools, exit_stack

Step5:使用ADK创建Agent

如上所述，我们使用的是 LlmAgent，它是应用中的“思考核心”部分。

# --- 第二步：定义 ADK Agent 的创建 ---
asyncdef get_agent_async():
    """创建一个配备了 MCP 工具的 ADK Agent。"""
    tools, exit_stack = await get_tools_async()
    print(f"已从 MCP 服务端获取 {len(tools)} 个工具。")
    
    # 创建符合示例结构的 LlmAgent
    root_agent = LlmAgent(
        model=os.getenv("GEMINI_MODEL", "gemini-2.5-pro-preview-03-25"),
        name='flight_search_assistant',
        instruction='帮助用户根据提示使用可用工具搜索航班。如果未指定返程日期，则将其设为空字符串表示单程。',
        tools=tools,
    )
    
    return root_agent, exit_stack

Step6:将 Agent 创建、会话管理与运行协调整合到 Runner 中

async def async_main():
    # 创建服务
    session_service = InMemorySessionService()

    # 创建一个会话
    session = session_service.create_session(
        state={}, app_name='flight_search_app', user_id='user_flights'
    )

    # 定义用户输入的查询语句
    query = "Find flights from Atlanta to Las Vegas 2025-05-05"
    print(f"用户查询：'{query}'")
    
    # 将输入格式化为 types.Content 类型
    content = types.Content(role='user', parts=[types.Part(text=query)])

    # 获取 Agent 和资源释放堆栈
    root_agent, exit_stack = await get_agent_async()

    # 创建 Runner
    runner = Runner(
        app_name='flight_search_app',
        agent=root_agent,
        session_service=session_service,
    )

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

    # 处理事件流
    final_content = None
    asyncfor event in events_async:
        print(f"接收到事件：{event}")

    # 始终清理资源
    print("正在关闭 MCP 服务端连接...")
    await exit_stack.aclose()
    print("资源清理完成。")

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

1. MCP 与 ADK 的区别

MCP 是一个开放协议，用于标准化 AI 模型如何与外部工具和数据源进行交互。
ADK 是一个基于 Python 的框架，用于构建和部署智能代理（AI Agents）。
MCPToolset：连接 MCP 与 ADK，使 ADK Agent 能够调用 MCP 服务端所暴露的工具功能。
若需在 Python 中构建 MCP 服务端，请使用 model-context-protocol 库。

2. 工具类型与集成方式

ADK 工具：如 BaseTool、FunctionTool 等 Python 对象，可直接在 ADK Agent 中使用。
MCP 工具：由 MCP 服务端提供的功能，经过 MCPToolset 适配后，可供 ADK Agent 调用。
第三方工具：如 LangChain 和 CrewAI 提供的工具，可通过 LangchainTool 或 CrewaiTool 包装后集成到 ADK 中。

3. 异步架构设计

ADK 和 MCP 的 Python 实现均基于 Python 的 asyncio 框架构建。
工具实现和服务端处理器应采用 async def 定义，以确保非阻塞执行。

4. MCP 中的有状态会话

与常见的无状态 REST API 不同，MCP 建立的是持久、有状态的客户端与服务端连接。
有状态机制支持上下文保留，但也要求开发者妥善管理会话状态。

5. 部署注意事项

MCP 的长连接特性在部署和扩展时可能带来挑战，尤其是在服务远程部署、同时处理多个用户的情况下。
基础设施层面需考虑负载均衡与会话亲和性（session affinity），以维持连接稳定性。

6. 在 ADK 中管理 MCP 连接

MCPToolset 在 ADK 内部负责 MCP 连接的生命周期管理。
使用 exit_stack 可确保在 Agent 执行结束后正确关闭连接，防止资源泄露。

🛠️ 常见问题排查（Troubleshooting）

默认情况下，ADK 库会要求配置 GCP 项目、Vertex AI 所在区域以及 VertexAI 的相关设置。请务必使用 GOOGLE_API_KEY，而不是 GEMINI_API_KEY，因为错误提示可能不会明确指出缺少的是哪个环境变量。

✅ 解决方案：确保将 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？

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 2 Flash 模型。

原因：Gemini API 的“免费套餐（free tier）”是通过 API 服务提供的，适用于测试用途，其速率限制较低，容易触发请求限制。使用 Gemini 2 Flash 可有效缓解该问题。