手撕AI Agent框架！多智能体+MCP+A2A全链路实战（DeepSeek可跑）

最近在AI Agent开发中踩了不少坑，特别是多智能体协作的场景下，不同框架的兼容性和部署复杂度让人头疼。本文将从零开始构建一个完整的餐厅多智能体系统，涵盖MCP工具集成、A2A协议通信、智能体部署等核心环节，所有代码都兼容DeepSeek模型，可直接运行。

1. AI Agent技术栈选型与架构设计

1.1 为什么需要多智能体架构？

在真实业务场景中，单个AI智能体往往难以胜任复杂任务。比如餐厅服务场景，需要菜单查询、预订管理、客户服务等多个专业能力。多智能体架构的优势在于：

• 职责分离 ：每个智能体专注特定领域，便于维护和迭代
• 独立部署 ：不同智能体可以有独立的更新周期和部署策略
• 弹性扩展 ：可根据业务负载单独扩缩容特定功能模块
• 技术异构 ：不同智能体可以采用最适合的技术栈

1.2 核心技术组件介绍

MCP（Model Context Protocol） MCP是连接AI智能体与工具数据的标准协议，让智能体能够安全地访问外部系统和数据源。在我们的餐厅场景中，MCP用于菜单数据的查询和管理。

A2A（Agent-to-Agent）协议
A2A解决了智能体间的通信问题，标准化了智能体发现、能力描述和任务协作的机制。与MCP主要区别在于：

• MCP：智能体↔工具（无状态函数调用）
• A2A：智能体↔智能体（有状态多轮对话）

ADK（Agent Development Kit） Google开源的AI智能体开发框架，提供统一的开发模式和部署工具。

1.3 系统架构设计

我们的多智能体餐厅系统包含以下组件：

用户请求
    ↓
餐厅主智能体（编排器）
    ├── MCP工具箱（菜单查询）
    └── A2A预订智能体（预订管理）
        ↓
    Agent Runtime（托管服务）

主智能体负责请求路由：菜单查询走MCP协议，预订操作走A2A协议。

2. 开发环境准备与依赖配置

2.1 基础环境要求

# 检查Python版本
python --version  # 需要Python 3.9+
uv --version     # 包管理工具

# 创建项目目录
mkdir ai-agent-restaurant && cd ai-agent-restaurant

2.2 项目依赖配置

创建 pyproject.toml 文件：

[project]
name = "ai-agent-restaurant"
version = "0.1.0"
description = "Multi-agent restaurant system with MCP and A2A"
requires-python = ">=3.9"

dependencies = [
    "google-cloud-aiplatform[agent_engines,adk]>=1.149.0",
    "a2a-sdk>=0.3.26",
    "google-adk>=1.29.0",
    "httpx>=0.25.0",
    "pydantic>=2.0.0",
    "cloudpickle>=2.0.0",
]

[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"

创建环境配置文件 .env.example ：

# 项目配置
GOOGLE_CLOUD_PROJECT=your-project-id
GOOGLE_CLOUD_LOCATION=us-central1
REGION=us-central1

# 服务配置
TOOLBOX_URL=http://127.0.0.1:5000
RESERVATION_AGENT_CARD_URL=

# 部署配置
STAGING_BUCKET=your-project-adk-a2a-agent-runtime

2.3 初始化项目结构

# 创建项目目录结构
mkdir -p {reservation_agent,restaurant_agent,scripts,logs,tools}

# 初始化Python环境
uv sync

项目最终结构如下：

ai-agent-restaurant/
├── reservation_agent/     # 预订智能体
│   ├── agent.py
│   ├── a2a_config.py
│   ├── executor.py
│   └── __init__.py
├── restaurant_agent/      # 餐厅主智能体
│   ├── agent.py
│   └── __init__.py
├── scripts/               # 部署和测试脚本
├── tools/                 # MCP工具配置
├── logs/                  # 日志文件
└── pyproject.toml

3. 构建预订智能体（A2A服务端）

3.1 预订智能体核心逻辑

创建 reservation_agent/agent.py ：

import os
from functools import cached_property
from typing import Any
from google.adk.agents import LlmAgent
from google.adk.models.google_llm import Gemini
from google.adk.tools import ToolContext
from google.genai import Client, types

# 应用级状态前缀，确保预订数据在所有会话间持久化
STATE_PREFIX = "app:reservation:"

class DeepSeekCompatibleGemini(Gemini):
    """兼容DeepSeek的Gemini封装类"""
    
    @cached_property
    def api_client(self) -> Client:
        project = os.getenv("GOOGLE_CLOUD_PROJECT")
        return Client(
            project=project,
            location="global",
            http_options=types.HttpOptions(
                headers=self._tracking_headers(),
                retry_options=self.retry_options,
            ),
        )

def create_reservation(
    phone_number: str,
    name: str,
    party_size: int,
    date: str,
    time: str,
    tool_context: ToolContext,
) -> dict:
    """创建新的餐厅预订"""
    reservation = {
        "name": name,
        "party_size": party_size,
        "date": date,
        "time": time,
        "status": "confirmed",
    }
    
    # 使用电话号码作为键存储预订信息
    tool_context.state[f"{STATE_PREFIX}{phone_number}"] = reservation
    
    return {
        "status": "confirmed",
        "message": f"Reservation created for {name}, party of {party_size} on {date} at {time}. Phone: {phone_number}.",
    }

def check_reservation(phone_number: str, tool_context: ToolContext) -> dict:
    """通过电话号码查询预订"""
    reservation = tool_context.state.get(f"{STATE_PREFIX}{phone_number}")
    
    if reservation:
        return {"found": True, "reservation": reservation}
    
    return {"found": False, "message": f"No reservation found for {phone_number}."}

def cancel_reservation(phone_number: str, tool_context: ToolContext) -> dict:
    """取消现有预订"""
    key = f"{STATE_PREFIX}{phone_number}"
    reservation = tool_context.state.get(key)
    
    if not reservation:
        return {
            "success": False,
            "message": f"No reservation found for {phone_number}.",
        }
    
    if reservation.get("status") == "cancelled":
        return {
            "success": False,
            "message": f"Reservation for {phone_number} is already cancelled.",
        }
    
    reservation["status"] = "cancelled"
    tool_context.state[key] = reservation
    
    return {
        "success": True,
        "message": f"Reservation for {reservation['name']} ({phone_number}) has been cancelled.",
    }

# 创建主智能体实例
root_agent = LlmAgent(
    name="reservation_agent",
    model=DeepSeekCompatibleGemini(model="gemini-3.5-flash"),
    instruction="""You are a friendly reservation assistant for "Foodie Finds" restaurant.
You help diners create, check, and cancel table reservations.

When a diner wants to make a reservation, collect these details:
- Name for the reservation
- Phone number (used as the reservation ID)
- Party size (number of guests)
- Date
- Time

Always confirm the details before creating the reservation.
When checking or cancelling, ask for the phone number if not provided.
Be concise and professional.""",
    tools=[create_reservation, check_reservation, cancel_reservation],
)

3.2 A2A智能体卡片配置

创建 reservation_agent/a2a_config.py ：

from a2a.types import AgentSkill
from vertexai.preview.reasoning_engines.templates.a2a import create_agent_card

# 定义预订技能
reservation_skill = AgentSkill(
    id="manage_reservations",
    name="Restaurant Reservations",
    description="Create, check, and cancel table reservations at Foodie Finds restaurant",
    tags=["reservations", "restaurant", "booking"],
    examples=[
        "Book a table for 4 on Friday at 7pm",
        "Check reservation for 555-0101",
        "Cancel my reservation, phone number 555-0101",
    ],
    input_modes=["text/plain"],
    output_modes=["text/plain"],
)

# 创建智能体卡片
agent_card = create_agent_card(
    agent_name="Reservation Agent",
    description="Handles restaurant table reservations — create, check, and cancel bookings for Foodie Finds restaurant.",
    skills=[reservation_skill],
)

3.3 A2A执行器实现

创建 reservation_agent/executor.py ：

import os
from typing import NoReturn
import vertexai
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, UnsupportedOperationError
from a2a.utils import new_agent_text_message
from a2a.utils.errors import ServerError
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService, VertexAiSessionService
from google.genai import types

from reservation_agent.agent import root_agent as reservation_agent

class ReservationAgentExecutor(AgentExecutor):
    """A2A协议与ADK预订智能体之间的桥梁"""
    
    def __init__(self) -> None:
        self.agent = None
        self.runner = None
    
    def _init_agent(self) -> None:
        if self.agent is not None:
            return
            
        self.agent = reservation_agent
        engine_id = os.environ.get("GOOGLE_CLOUD_AGENT_ENGINE_ID")
        
        # 根据环境选择会话服务
        if engine_id:
            # 生产环境：使用持久化会话
            project = os.environ.get("GOOGLE_CLOUD_PROJECT")
            location = os.environ.get("GOOGLE_CLOUD_LOCATION", "us-central1")
            vertexai.init(project=project, location=location)
            session_service = VertexAiSessionService(
                project=project, location=location, agent_engine_id=engine_id,
            )
            app_name = engine_id
        else:
            # 本地环境：使用内存会话
            session_service = InMemorySessionService()
            app_name = self.agent.name
        
        self.runner = Runner(
            app_name=app_name,
            agent=self.agent,
            artifact_service=InMemoryArtifactService(),
            session_service=session_service,
            memory_service=InMemoryMemoryService(),
        )
    
    async def execute(self, context: RequestContext, event_queue: EventQueue) -> None:
        if self.agent is None:
            self._init_agent()
        
        query = context.get_user_input()
        updater = TaskUpdater(event_queue, context.task_id, context.context_id)
        
        user_id = context.message.metadata.get("user_id", "a2a-user") if context.message.metadata else "a2a-user"
        
        if not context.current_task:
            await updater.submit()
        
        await updater.start_work()
        
        try:
            session = await self._get_or_create_session(context.context_id, user_id)
            content = types.Content(role="user", parts=[types.Part(text=query)])
            
            async for event in self.runner.run_async(
                session_id=session.id, user_id=user_id, new_message=content,
            ):
                if event.is_final_response():
                    parts = event.content.parts
                    answer = " ".join(p.text for p in parts if p.text) or "No response."
                    await updater.add_artifact([TextPart(text=answer)], name="answer")
                    await updater.complete()
                    break
                    
        except Exception as e:
            await updater.update_status(
                TaskState.failed, message=new_agent_text_message(f"Error: {e!s}"),
            )
            raise
    
    async def _get_or_create_session(self, context_id: str, user_id: str):
        app_name = self.runner.app_name
        
        if context_id:
            session = await self.runner.session_service.get_session(
                app_name=app_name, session_id=context_id, user_id=user_id,
            )
            if session:
                return session
        
        session = await self.runner.session_service.create_session(
            app_name=app_name, user_id=user_id, session_id=context_id,
        )
        return session
    
    async def cancel(self, context: RequestContext, event_queue: EventQueue) -> NoReturn:
        raise ServerError(error=UnsupportedOperationError())

4. 本地测试A2A智能体

4.1 创建本地测试脚本

创建 scripts/test_a2a_agent_local.py ：

import asyncio
import json
import os
from pprint import pprint
from dotenv import load_dotenv
from starlette.requests import Request
from vertexai.preview.reasoning_engines import A2aAgent

from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor

load_dotenv()

def build_post_request(data: dict = None, path_params: dict = None) -> Request:
    """构建模拟POST请求"""
    scope = {
        "type": "http", 
        "http_version": "1.1", 
        "headers": [(b"content-type", b"application/json")], 
        "app": None
    }
    
    if path_params:
        scope["path_params"] = path_params
    
    async def receive():
        byte_data = json.dumps(data).encode("utf-8") if data else b""
        return {"type": "http.request", "body": byte_data, "more_body": False}
    
    return Request(scope, receive)

async def wait_for_task(a2a_agent, task_id, max_retries=30):
    """轮询任务直到完成"""
    for _ in range(max_retries):
        request = build_post_request()
        result = await a2a_agent.on_get_task(request=request, context=None)
        state = result.get("status", {}).get("state", "")
        if state in ["completed", "failed"]:
            return result
        await asyncio.sleep(1)
    return result

async def main():
    # 创建本地A2A智能体
    a2a_agent = A2aAgent(
        agent_card=agent_card, 
        agent_executor_builder=ReservationAgentExecutor
    )
    a2a_agent.set_up()
    
    print("=" * 50)
    print("1. 测试智能体卡片检索...")
    print("=" * 50)
    
    # 测试智能体卡片
    request = build_post_request()
    card_response = await a2a_agent.handle_authenticated_agent_card(request=request, context=None)
    print(f"智能体名称: {card_response.get('name')}")
    print(f"可用技能: {[s.get('name') for s in card_response.get('skills', [])]}")
    
    # 测试预订创建
    print("\n" + "=" * 50)
    print("2. 测试预订创建...")
    print("=" * 50)
    
    message_data = {
        "message": {
            "messageId": f"msg-test-001",
            "content": [{"text": "Book a table for 2 on Saturday at 6pm. Name: Bob, Phone: 555-0202"}],
            "role": "ROLE_USER",
        },
    }
    
    request = build_post_request(message_data)
    response = await a2a_agent.on_message_send(request=request, context=None)
    task_id = response["task"]["id"]
    context_id = response["task"].get("contextId")
    
    print(f"任务ID: {task_id}")
    
    # 等待任务完成
    result = await wait_for_task(a2a_agent, task_id)
    print(f"任务状态: {result.get('status', {}).get('state')}")
    
    if result.get("artifacts"):
        for artifact in result.get("artifacts", []):
            if artifact.get("parts") and "text" in artifact["parts"][0]:
                print(f"智能体回复: {artifact['parts'][0]['text']}")
    
    print("\n" + "=" * 50)
    print("本地A2A测试完成!")
    print("=" * 50)

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

4.2 运行本地测试

# 设置环境变量
source .env

# 运行测试
PYTHONPATH=. uv run python scripts/test_a2a_agent_local.py

预期输出：

==================================================
1. 测试智能体卡片检索...
==================================================
智能体名称: Reservation Agent
可用技能: ['Restaurant Reservations']

==================================================
2. 测试预订创建...
==================================================
任务ID: xxxxxxxx-xxxx-xxxx-xxxx-xxxxxxxxxxxx
任务状态: completed
智能体回复: Your reservation for Bob, party of 2, on Saturday at 6:00 PM has been confirmed.

5. 部署A2A智能体到Agent Runtime

5.1 创建部署脚本

创建 scripts/deploy_a2a_agent_runtime.py ：

import os
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types
from vertexai.preview.reasoning_engines import A2aAgent

from reservation_agent.a2a_config import agent_card
from reservation_agent.executor import ReservationAgentExecutor

load_dotenv()

def main():
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
    REGION = os.environ["REGION"]
    STAGING_BUCKET = os.environ.get("STAGING_BUCKET", f"{PROJECT_ID}-adk-a2a-agent-runtime")
    BUCKET_URI = f"gs://{STAGING_BUCKET}"
    
    # 初始化A2A智能体
    a2a_agent = A2aAgent(
        agent_card=agent_card,
        agent_executor_builder=ReservationAgentExecutor,
    )
    
    # 初始化Vertex AI
    vertexai.init(project=PROJECT_ID, location=REGION, staging_bucket=BUCKET_URI)
    
    client = vertexai.Client(
        project=PROJECT_ID,
        location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )
    
    print("正在部署预订智能体到Agent Runtime...")
    print("这可能需要3-5分钟...")
    
    # 部署到Agent Runtime
    remote_agent = client.agent_engines.create(
        agent=a2a_agent,
        config={
            "display_name": agent_card.name,
            "description": agent_card.description,
            "requirements": [
                "google-cloud-aiplatform[agent_engines,adk]==1.149.0",
                "a2a-sdk==0.3.26",
                "google-adk==1.29.0",
                "cloudpickle>=2.0.0",
                "pydantic>=2.0.0"
            ],
            "extra_packages": ["./reservation_agent"],
            "http_options": {"api_version": "v1beta1"},
            "staging_bucket": BUCKET_URI,
        },
    )
    
    resource_name = remote_agent.api_resource.name
    print(f"\n部署完成!")
    print(f"资源名称: {resource_name}")
    
    # 保存资源名称到环境文件
    env_path = Path(".env")
    lines = env_path.read_text().splitlines() if env_path.exists() else []
    lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_RESOURCE_NAME=")]
    lines.append(f"RESERVATION_AGENT_RESOURCE_NAME={resource_name}")
    env_path.write_text("\n".join(lines) + "\n")
    
    print("已将RESERVATION_AGENT_RESOURCE_NAME写入.env文件")

if __name__ == "__main__":
    main()

5.2 执行部署

# 创建存储桶用于暂存
STAGING_BUCKET="${GOOGLE_CLOUD_PROJECT}-adk-a2a-agent-runtime"
gsutil mb -l $REGION -p $GOOGLE_CLOUD_PROJECT gs://$STAGING_BUCKET 2>/dev/null || echo "存储桶已存在"
echo "STAGING_BUCKET=$STAGING_BUCKET" >> .env

# 运行部署
source .env
PYTHONPATH=. uv run python scripts/deploy_a2a_agent_runtime.py

5.3 验证部署

创建测试脚本 scripts/test_deployed_agent.py ：

import asyncio
import os
import time
import vertexai
from a2a.types import TaskState
from dotenv import load_dotenv
from google.genai import types

load_dotenv()

async def main():
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
    REGION = os.environ["REGION"]
    RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]
    
    vertexai.init(project=PROJECT_ID, location=REGION)
    client = vertexai.Client(
        project=PROJECT_ID, 
        location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )
    
    # 获取已部署的智能体
    agent = client.agent_engines.get(name=RESOURCE_NAME)
    
    print("=" * 50)
    print("测试已部署的A2A智能体...")
    print("=" * 50)
    
    # 发送测试消息
    message_data = {
        "messageId": "msg-deploy-test-001",
        "role": "user",
        "parts": [{"kind": "text", "text": "Book a table for 3 on Sunday at noon. Name: Carol, Phone: 555-0303"}],
    }
    
    response = await agent.on_message_send(**message_data)
    
    # 处理响应
    task_object = None
    for chunk in response:
        if isinstance(chunk, tuple) and len(chunk) > 0 and hasattr(chunk[0], "id"):
            task_object = chunk[0]
            break
    
    if task_object:
        print(f"任务ID: {task_object.id}")
        print(f"初始状态: {task_object.status.state}")
        
        # 等待任务完成
        for _ in range(30):
            try:
                result = await agent.on_get_task(id=task_object.id)
                if result.status.state in [TaskState.completed, TaskState.failed]:
                    break
            except Exception:
                pass
            time.sleep(1)
        
        print(f"最终状态: {result.status.state}")
        
        if result.artifacts:
            for artifact in result.artifacts:
                if artifact.parts and hasattr(artifact.parts[0], "text"):
                    print(f"智能体回复: {artifact.parts[0].text}")

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

6. 构建餐厅主智能体（MCP + A2A集成）

6.1 主智能体实现

创建 restaurant_agent/agent.py ：

import os
import httpx
from google.adk.agents import LlmAgent
from google.adk.agents.remote_a2a_agent import RemoteA2aAgent
from google.auth import default
from google.auth.transport.requests import Request as AuthRequest

# MCP工具箱集成（模拟）
class MockToolboxToolset:
    """模拟MCP工具箱用于菜单查询"""
    
    def __init__(self, toolbox_url: str):
        self.toolbox_url = toolbox_url
    
    async def search_menu(self, query: str) -> dict:
        """模拟菜单搜索"""
        # 在实际项目中这里会调用真实的MCP服务
        menu_items = [
            {"name": "Margherita Pizza", "price": 12.99, "category": "Italian"},
            {"name": "Caesar Salad", "price": 8.99, "category": "Salads"},
            {"name": "Spaghetti Carbonara", "price": 14.99, "category": "Italian"},
        ]
        
        return {
            "results": [item for item in menu_items if query.lower() in item["name"].lower() or query.lower() in item["category"].lower()],
            "count": len(menu_items)
        }

# Google Cloud认证处理
class GoogleCloudAuth(httpx.Auth):
    """自动刷新Google Cloud访问令牌"""
    
    def __init__(self):
        self.credentials, _ = default(
            scopes=["https://www.googleapis.com/auth/cloud-platform"]
        )
    
    def auth_flow(self, request):
        if not self.credentials.valid:
            self.credentials.refresh(AuthRequest())
        request.headers["Authorization"] = f"Bearer {self.credentials.token}"
        yield request

# 配置环境变量
TOOLBOX_URL = os.environ.get("TOOLBOX_URL", "http://127.0.0.1:5000")
RESERVATION_AGENT_CARD_URL = os.environ.get("RESERVATION_AGENT_CARD_URL", "")

# 初始化工具箱
toolbox = MockToolboxToolset(TOOLBOX_URL)

# 创建远程A2A智能体（预订服务）
reservation_remote_agent = RemoteA2aAgent(
    name="reservation_agent",
    description="Handles restaurant table reservations — create, check, and cancel bookings.",
    agent_card=RESERVATION_AGENT_CARD_URL,
    httpx_client=httpx.AsyncClient(auth=GoogleCloudAuth(), timeout=60),
)

# 创建主智能体
root_agent = LlmAgent(
    name="restaurant_agent",
    model="gemini-3.5-flash",
    instruction="""You are a friendly and knowledgeable concierge at "Foodie Finds" restaurant. 

核心职责：
- 通过MCP工具箱帮助顾客浏览菜单（按类别或菜系）
- 提供菜品详细信息（成分、价格、饮食信息）
- 根据顾客描述推荐菜品
- 将预订相关请求委托给专门的预订智能体

请求路由逻辑：
1. 菜单查询 → MCP工具箱
2. 预订操作 → A2A预订智能体

对话风格：专业、友好、简洁，确保顾客获得准确信息。""",
    tools=[toolbox.search_menu],  # 实际项目中会有更多MCP工具
    sub_agents=[reservation_remote_agent],
)

6.2 智能体卡片解析脚本

创建 scripts/resolve_agent_card_url.py ：

import asyncio
import os
from pathlib import Path
import vertexai
from dotenv import load_dotenv
from google.genai import types

load_dotenv()

async def main():
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
    REGION = os.environ["REGION"]
    RESOURCE_NAME = os.environ["RESERVATION_AGENT_RESOURCE_NAME"]
    
    vertexai.init(project=PROJECT_ID, location=REGION)
    client = vertexai.Client(
        project=PROJECT_ID, 
        location=REGION,
        http_options=types.HttpOptions(api_version="v1beta1"),
    )
    
    # 获取已部署的智能体
    agent = client.agent_engines.get(name=RESOURCE_NAME)
    
    # 获取智能体卡片
    card = await agent.handle_authenticated_agent_card()
    card_url = f"{card.url}/v1/card"
    
    print(f"智能体: {card.name}")
    print(f"卡片URL: {card_url}")
    
    # 更新环境配置文件
    for env_path in [Path("restaurant_agent/.env"), Path(".env")]:
        lines = env_path.read_text().splitlines() if env_path.exists() else []
        lines = [l for l in lines if not l.startswith("RESERVATION_AGENT_CARD_URL=")]
        lines.append(f"RESERVATION_AGENT_CARD_URL={card_url}")
        env_path.write_text("\n".join(lines) + "\n")
        print(f"已更新 {env_path}")

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

7. 完整系统集成测试

7.1 集成测试脚本

创建 scripts/test_integrated_system.py ：

import asyncio
import os
from dotenv import load_dotenv
from restaurant_agent.agent import root_agent as restaurant_agent
from google.adk.runners import Runner
from google.adk.sessions import InMemorySessionService
from google.adk.artifacts import InMemoryArtifactService
from google.adk.memory.in_memory_memory_service import InMemoryMemoryService
from google.genai import types

load_dotenv()

async def test_integrated_system():
    """测试完整的多智能体系统"""
    
    # 初始化餐厅智能体运行器
    runner = Runner(
        app_name="restaurant_test",
        agent=restaurant_agent,
        artifact_service=InMemoryArtifactService(),
        session_service=InMemorySessionService(),
        memory_service=InMemoryMemoryService(),
    )
    
    test_cases = [
        {
            "name": "菜单查询测试",
            "query": "What Italian dishes do you have?",
            "expected_keywords": ["pizza", "pasta", "Italian"]
        },
        {
            "name": "预订创建测试", 
            "query": "I want to book a table for 4 people this Friday at 7 PM",
            "expected_keywords": ["reservation", "book", "table"]
        },
        {
            "name": "预订查询测试",
            "query": "Can you check my reservation for phone number 555-0202?",
            "expected_keywords": ["reservation", "check", "found"]
        }
    ]
    
    for i, test_case in enumerate(test_cases):
        print(f"\n{'='*50}")
        print(f"测试 {i+1}: {test_case['name']}")
        print(f"{'='*50}")
        
        print(f"用户查询: {test_case['query']}")
        
        # 运行智能体
        session = await runner.session_service.create_session(
            app_name="restaurant_test", 
            user_id=f"test_user_{i}",
            session_id=f"test_session_{i}"
        )
        
        content = types.Content(role="user", parts=[types.Part(text=test_case['query'])])
        
        async for event in runner.run_async(
            session_id=session.id, 
            user_id=f"test_user_{i}", 
            new_message=content,
        ):
            if event.is_final_response():
                response_text = " ".join(p.text for p in event.content.parts if p.text)
                print(f"智能体回复: {response_text}")
                
                # 验证回复包含预期关键词
                keywords_found = [
                    keyword for keyword in test_case['expected_keywords'] 
                    if keyword.lower() in response_text.lower()
                ]
                print(f"找到关键词: {keywords_found}")
                break

async def main():
    print("开始集成系统测试...")
    await test_integrated_system()
    print("\n测试完成!")

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

7.2 运行集成测试

# 解析智能体卡片URL
uv run python scripts/resolve_agent_card_url.py

# 运行集成测试
source .env
PYTHONPATH=. uv run python scripts/test_integrated_system.py

8. 生产环境部署与优化

8.1 Cloud Run部署配置

创建 scripts/deploy_to_cloud_run.py ：

import os
import subprocess
from dotenv import load_dotenv

load_dotenv()

def deploy_restaurant_agent():
    """部署餐厅主智能体到Cloud Run"""
    
    PROJECT_ID = os.environ["GOOGLE_CLOUD_PROJECT"]
    REGION = os.environ["REGION"]
    RESERVATION_AGENT_CARD_URL = os.environ["RESERVATION_AGENT_CARD_URL"]
    
    # 授予Cloud Run服务账号访问Agent Runtime的权限
    project_number_cmd = f"gcloud projects describe {PROJECT_ID} --format='value(projectNumber)'"
    project_number = subprocess.check_output(project_number_cmd, shell=True).decode().strip()
    
    # 设置IAM权限
    iam_cmd = f"""
    gcloud projects add-iam-policy-binding {PROJECT_ID} \
        --member="serviceAccount:{project_number}-compute@developer.gserviceaccount.com" \
        --role="roles/aiplatform.user"
    """
    
    subprocess.run(iam_cmd, shell=True, check=True)
    
    # 部署到Cloud Run
    deploy_cmd = f"""
    gcloud run deploy restaurant-agent \
        --source . \
        --region={REGION} \
        --allow-unauthenticated \
        --update-env-vars="RESERVATION_AGENT_CARD_URL={RESERVATION_AGENT_CARD_URL}" \
        --min-instances=0 \
        --max-instances=1 \
        --memory=1Gi \
        --port=8080
    """
    
    print("正在部署餐厅智能体到Cloud Run...")
    subprocess.run(deploy_cmd, shell=True, check=True)
    
    # 获取服务URL
    url_cmd = f"gcloud run services describe restaurant-agent --region={REGION} --format='value(status.url)'"
    service_url = subprocess.check_output(url_cmd, shell=True).decode().strip()
    
    print(f"部署完成! 服务URL: {service_url}")
    return service_url

if __name__ == "__main__":
    deploy_restaurant_agent()

8.2 性能优化配置

创建 optimization_config.yaml ：

# 智能体性能优化配置
optimization:
  # 会话管理
  session:
    timeout: 3600  # 会话超时时间（秒）
    max_size: 1000  # 最大会话数
    
  # 缓存配置
  cache:
    enabled: true
    ttl: 300  # 缓存存活时间
    
  # 并发控制
  concurrency:
    max_workers: 10
    queue_size: 100
    
  # 监控配置
  monitoring:
    enabled: true
    metrics:
      - response_time
      - error_rate
      - throughput

9. 常见问题与解决方案

9.