1. 개요
이 Codelab에서는 여러 ADK 에이전트가 Agent2Agent (A2A) 프로토콜을 사용하여 통신하고 공동작업하는 멀티 에이전트 시스템을 빌드합니다.
학습할 내용
- 독립적인 ADK 에이전트를 여러 개 만드는 방법
- 각 에이전트에게 에이전트 카드를 부여하고 A2A 서버로 래핑하는 방법
- 원격 에이전트를 오케스트레이션하는 호스트 에이전트를 빌드하는 방법
- 원격 에이전트 연결을 설정하는 방법
- 멀티 에이전트 시스템을 로컬에서 테스트하는 방법
필요한 항목
- 결제가 사용 설정된 Google Cloud 프로젝트
- 웹브라우저(예: Chrome)
- Python 3.12 이상
이 Codelab은 Python 및 Google Cloud에 익숙한 중급 개발자를 대상으로 합니다.
이 Codelab을 완료하는 데 약 15분이 소요됩니다.
이 Codelab에서 만든 리소스의 비용은 5달러 미만이어야 합니다.
2. 환경 설정
Google Cloud 프로젝트 만들기
- Google Cloud 콘솔의 프로젝트 선택기 페이지에서 Google Cloud 프로젝트를 선택하거나 만듭니다.
- Cloud 프로젝트에 결제가 사용 설정되어 있는지 확인합니다. 프로젝트에 결제가 사용 설정되어 있는지 확인하는 방법을 알아보세요.
Cloud Shell 편집기 시작
Google Cloud 콘솔에서 Cloud Shell 세션을 시작하려면 Google Cloud 콘솔에서 Cloud Shell 활성화를 클릭합니다.
그러면 Google Cloud 콘솔 하단 창에서 세션이 시작됩니다.
편집기를 실행하려면 Cloud Shell 창의 툴바에서 편집기 열기를 클릭합니다.
환경 구성
터미널에서 다음 명령어를 실행하여 A2A 시스템의 프로젝트 폴더 구조를 만듭니다. 이 데모에서는 $HOME 디렉터리의 절대 경로를 사용합니다.
mkdir -p $HOME/app/src/agents $HOME/app/src/host
touch $HOME/app/.env $HOME/app/pyproject.toml
이제 일반적인 아키텍처가 있으므로 환경 구성을 채워 보겠습니다. 다음 코드 세그먼트를 새 .env 파일에 복사합니다.
새 .env 파일에 GCP 프로젝트 ID와 GCP 리전을 입력합니다. 빌드하려는 에이전트로부터 보고서 이메일을 받으려면 MAIL_TO에 원하는 이메일을 입력해도 됩니다. GitHub 검색 에이전트를 용이하게 하기 위해 GitHub 개인 액세스 토큰 GITHUB_TOKEN을 지속적으로 추가할 수 있습니다.
다음 bash 명령어를 사용하여 새 .env 파일을 엽니다.
cloudshell edit .env
그런 다음 다음 파일을 app.env의 .env 파일에 복사합니다. 중요: 본인의 값을 입력해야 합니다.
# --- Google Cloud ---
GOOGLE_GENAI_USE_VERTEXAI=TRUE
GOOGLE_CLOUD_PROJECT=<your-gcp-project-id>
GOOGLE_CLOUD_LOCATION=<your-gcp-project-region>
# --- GitHub ---
# Personal Access Token with "repo" scope
# Create one at: https://github.com/settings/tokens
# Generate new token --> Fine-grained, repo-secured --> (populate) Token name --> (scroll down) Generate token
GITHUB_TOKEN=<your-github-pat>
MCP_SERVER_HOST=https://api.githubcopilot.com/mcp/
TARGET_REPOS=["google/adk-python", "langchain-ai/langchain"]
DEFAULT_ISSUE_COUNT=5
DEFAULT_PR_COUNT=5
# --- Email (Optional) ---
# Only needed if you want the emailer agent to send reports
RESEND_API_KEY=<your-resend-API-key>
RESEND_DOMAIN=<your-domain>
MAIL_TO=<insert-email-to-receive-reports>
# --- Local Development Overrides ---
RETRIEVAL_AGENT_URL=http://localhost:8001
EVALUATOR_AGENT_URL=http://localhost:8002
EMAILER_AGENT_URL=http://localhost:8003
ORCHESTRATOR_PORT=http://localhost:8000
이제 환경 변수를 채웠으므로 uv 환경을 구성해야 합니다. 다음 코드 세그먼트를 pyproject.toml 파일에 복사합니다.
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project]
name = "app"
version = "0.1.0"
description = "Multi-agent system designed to assess the newest issues and pull requests from numerous agentic development platforms"
authors = [
{ name = "Thomas Wagner" },
{ name = "Kris Overholt" }
]
license = "Apache-2.0"
requires-python = ">=3.11,<3.14"
dependencies = [
"resend>=2.21.0",
"uvicorn>=0.40.0",
"a2a-sdk>=0.3.26,<0.4.0",
"google-adk[a2a]>=1.27.0,<1.30.0",
"google-generativeai>=0.8.4",
"httpx>=0.28.1",
"pydantic>=2.12.5, <3.0.0",
"python-dotenv>=1.2.0",
"nest-asyncio>=1.6.0",
]
[project.optional-dependencies]
test = [
"pytest>=8.3.2",
"pytest-asyncio>=0.23.8",
"pytest-mock>=3.14.0",
"respx>=0.21.0",
]
[tool.ruff]
target-version = "py313"
line-length = 80
[tool.ruff.lint]
select = ["E", "F", "I", "C", "PL", "B", "UP", "RUF"]
ignore = ["E501", "C901"]
[tool.ruff.lint.per-file-ignores]
"__init__.py" = ["F401"]
[tool.ruff.lint.isort]
known-first-party = ["src"]
[tool.hatch.build.targets.wheel]
packages=["src/"]
가상 환경 만들기
이제 방금 만든 app 디렉터리 내에서 터미널에서 다음 bash 스크립트를 실행합니다. 이렇게 하면 Python 가상 환경이 설정되고 pyproject.toml 파일에서 필요한 모든 종속 항목이 설치됩니다.
# Ensure you are in the 'app' directory before running this
# Exit immediately if a command exits with a non-zero status.
set -e
# 1. Install uv, the Python package manager used for this project
echo "Installing uv..."
if ! command -v uv &> /dev/null
then
pip install uv
else
echo "uv is already installed."
fi
# 2. Create and activate virtual environment
echo "Setting up virtual environment..."
# --clear ensures you start with a fresh environment
uv venv --clear
# 3. Install dependencies
echo "Installing Python dependencies..."
uv sync
echo "Installation and setup complete."
이제 멀티 에이전트 시스템을 만들 준비가 되었습니다.
3. 리트리버 에이전트 만들기
Eva는 새로운 문제와 PR로 업데이트되는 GitHub 저장소를 최신 상태로 유지하고 싶어 하는 소프트웨어 개발자입니다. 따라서 그녀는 요청한 GitHub 데이터를 가져오는 ADK 에이전트를 만듭니다.
이 데모를 위해 프로젝트의 루트에서 다음 명령어를 실행하여 검색기 에이전트에 필요한 디렉터리와 파일을 만듭니다.
mkdir -p $HOME/app/src/agents/retriever
touch $HOME/app/src/agents/retriever/agent.py
touch $HOME/app/src/agents/retriever/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/retriever/__init__.py
다음 ADK 코드 세그먼트를 $HOME/app/src/agents/retriever/agent.py에 복사합니다.
import logging
import os
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.tools.mcp_tool.mcp_session_manager import (
StreamableHTTPConnectionParams,
)
from google.adk.tools.mcp_tool.mcp_toolset import McpToolset
logger = logging.getLogger(__name__)
load_dotenv()
GITHUB_MCP_URL = os.getenv(
"GITHUB_MCP_URL", "https://api.githubcopilot.com/mcp/"
)
GITHUB_TOKEN = os.getenv("GITHUB_TOKEN")
if GITHUB_TOKEN is None:
raise ValueError("GITHUB_TOKEN env is not set")
TARGET_REPOS = os.getenv("TARGET_REPOS")
DEFAULT_ISSUE_COUNT = os.getenv("DEFAULT_ISSUE_COUNT")
DEFAULT_PR_COUNT = os.getenv("DEFAULT_PR_COUNT")
# --- Prompt ---
GITHUB_RETRIEVAL_INSTRUCTIONS = f"""
You are a specialized GitHub data retrieval agent.
Your only purpose is to fetch data from GitHub using the available MCP tools and format it for another agent.
**DEFAULT CONFIGURATION:**
If the orchestrator does not specify which repositories or how many items to fetch, you MUST use the following defaults:
- **Repositories:** {TARGET_REPOS}
- **PR Count:** {DEFAULT_PR_COUNT} per repository
- **Issue Count:** {DEFAULT_ISSUE_COUNT} per repository
**Your Task:**
1. Analyze the task. If no specific repo is mentioned, iterate through the Default Repositories list above.
2. Use the `GitHub` MCP tool to fetch the requested data (Pull Requests and/or Issues).
3. **CRITICAL:** After the tool has finished running, you MUST take the raw output and compile it into a single response.
The orchestrator is waiting for this raw data to pass to the Evaluator. Do not summarize it.
"""
# --- Agent Initialization ---
root_agent = Agent(
model="gemini-2.5-flash",
name="retriever",
instruction=GITHUB_RETRIEVAL_INSTRUCTIONS,
description="""
Connects to the GitHub MCP server to retrieve real-time development
insights, actively reading repository issues, pull requests, and commit
histories.
""",
tools=[
McpToolset(
connection_params=StreamableHTTPConnectionParams(
url=GITHUB_MCP_URL,
headers={
"Authorization": f"Bearer {GITHUB_TOKEN}",
"X-MCP-Toolsets": "repos,issues,pull_requests",
"X-MCP-Readonly": "true",
},
)
)
],
)
이 에이전트는 독립형 도구로 작동합니다. 독립적으로 테스트하려면 다음 명령어를 실행하여 **/agents/ 디렉터리에서 uv run adk web을 실행하세요.
cd $HOME/app/src/agents
uv run adk run retriever
'uv run adk web' 터미널에서 localhost 링크를 열고 에이전트를 선택하여 사용해 봅니다.
4. 평가자 에이전트 만들기
Richard는 기술에 익숙하지 않은 고객과 동료를 위해 많은 기술 용어를 이해하기 쉽게 설명해야 하는 경우가 많습니다. 동일한 용어를 정의하고 동일한 프로젝트를 간결하게 설명해야 하는 데 지친 Richard는 기술 명칭을 쉽게 이해할 수 있는 텍스트로 요약하는 요약 에이전트를 만듭니다.
이 데모에서는 다음 명령어를 실행하여 평가자 에이전트의 파일을 만듭니다.
mkdir -p $HOME/app/src/agents/evaluator
touch $HOME/app/src/agents/evaluator/agent.py
touch $HOME/app/src/agents/evaluator/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/evaluator/__init__.py
아래 에이전트 코드 세그먼트를 $HOME/app/src/agents/evaluator/agent.py에 복사합니다.
import json
from google.adk.agents.llm_agent import Agent
# --- Prompt ---
EVAL_AND_SUMMARIZATION_INSTRUCTIONS = """
You are a specialized analysis and summarization agent. Your only purpose is to take raw, structured text about GitHub repositories and transform it into a concise, human-readable Markdown report.
**Your Task:**
1. You will receive a block of text containing raw data about pull requests and issues from an orchestrator.
2. Analyze the provided data. For pull requests, evaluate their significance. For issues, identify key themes and problems.
3. **CRITICAL:** You MUST generate a comprehensive summary in Markdown format based on your analysis. Your final output should be ONLY this Markdown report. Do not add any conversational text or explanations (e.g., "Here is the summary..."). The orchestrator needs to pass your clean report to another agent or directly to the user.
**Output Format Rules:**
- The report MUST be in Markdown.
- Structure the report by repository.
- For each repository, provide a concise overview of significant pull requests and important issues.
- Conclude with overall insights.
The orchestrator is waiting for this report. Ensure your final response consists of nothing but the complete Markdown summary.
"""
# --- Agent ---
root_agent = Agent(
model="gemini-2.5-flash",
name="evaluator",
instruction=EVAL_AND_SUMMARIZATION_INSTRUCTIONS,
description="""
Distills and summarizes complex GitHub data, breaking down pull
requests, code changes, and issue discussions into easily understandable
insights.
""",
)
이 에이전트는 독립형 도구로 작동합니다. 독립적으로 테스트하려면 새 터미널에서 다음 명령어를 실행하여 **/agents/ 디렉터리에서 uv run adk web을 실행하세요.
cd $HOME/app/src/agents
uv run adk run evaluator
터미널에서 localhost 링크를 열고 평가자 에이전트를 선택하여 사용해 봅니다.
5. Emailer 에이전트 만들기
Ivan은 쉽게 사용할 수 있는 텍스트를 요약하고 다시 포맷해야 하는 이메일의 양에 질렸습니다. 그래서 그는 지정된 텍스트를 다시 포맷하여 제공된 이메일 계정으로 이메일을 보내는 이메일 에이전트를 만들었습니다.
이 데모를 위해 터미널에서 다음 명령어를 실행하여 이메일러 에이전트의 파일을 만듭니다.
mkdir -p $HOME/app/src/agents/emailer
touch $HOME/app/src/agents/emailer/agent.py
touch $HOME/app/src/agents/emailer/__init__.py
echo "from . import agent" >> $HOME/app/src/agents/emailer/__init__.py
다음 에이전트 코드 세그먼트를 app/src/agents/emailer/agent.py에 복사합니다.
import logging
import os
import resend
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
load_dotenv()
logger = logging.getLogger(__name__)
RESEND_API_KEY = os.getenv("RESEND_API_KEY")
RESEND_DOMAIN = os.getenv("RESEND_DOMAIN")
MAIL_TO = os.getenv("MAIL_TO")
if RESEND_API_KEY:
resend.api_key = RESEND_API_KEY
# --- Tools ---
def send_report_email(recipient: str, subject: str, body: str) -> str:
"""
Sends an email of the given subject and body to the specified recipient
via Resend. If recipient is None, it defaults to the MAIL_TO environment variable.
Args:
recipient (str | None): The email address of the recipient. If None, defaults to MAIL_TO.
subject (str): The subject of the email.
body (str): The body of the email.
Returns:
str: A success or failure message.
"""
if not recipient:
recipient = MAIL_TO
if not all([RESEND_API_KEY, RESEND_DOMAIN, recipient]):
error_msg = "Error: Email tool configuration missing (API Key, Domain, or Recipient)"
logger.error(error_msg)
return error_msg
try:
html_body = f"<div style='font-family: sans-serif; white-space: pre-wrap;'>{body}</div>"
params: resend.Emails.SendParams = {
"from": f"Research Agent <agent@{RESEND_DOMAIN}>",
"to": [recipient], #type: ignore
"subject": subject,
"text": body,
"html": html_body,
}
email = resend.Emails.send(params)
logger.info(f"Email sent successfully. ID: {email['id']}")
return f"Email sent! ID: {email['id']}"
except Exception as e:
error_msg = f"Failed to send email: {e}"
logger.error(error_msg)
return error_msg
# --- Prompt ---
EMAIL_INSTRUCTIONS = """
You are an emailer agent responsible for formatting and sending a research report via email.
INPUT: You will receive a Markdown-formatted string (the report) and the email recipient.
YOUR TASK:
1. Take the Markdown content and format it appropriately for an email body.
The goal is to render the Markdown effectively so it is readable and well-presented in an email client.
2. Based on the summary, generate a concise and informative subject line for the email.
The subject line should reflect the main themes or key insights from the report.
3. Use the `send_report_email` tool with the extracted recipient, the generated subject line, and the formatted email body.
If no email recipient is provided, output a message indicating that no recipient was specified and do NOT call the tool.
"""
# --- Agent ---
root_agent = Agent(
model="gemini-2.5-flash",
name="emailer",
instruction=EMAIL_INSTRUCTIONS,
description="""
Acts as the final delivery mechanism in the pipeline, dispatching
generated summaries and reports to designated email addresses.
""",
tools=[
send_report_email
],
)
이 에이전트는 독립형 도구로 작동합니다. 독립적으로 테스트하려면 새 터미널에서 다음 명령어를 실행하여 **/agents/ 디렉터리에서 uv run adk web을 실행하세요.
cd $HOME/app/src/agents
uv run adk run emailer
터미널에서 localhost 링크를 열고 에이전트를 선택하여 사용해 봅니다.
6. 상담사 카드 만들기
지금까지 각자 고유한 에이전트를 보유한 세 가지 독립 개발자 스토리를 살펴봤습니다. 이러한 모든 상담사는 게시되어 회사 상담사 라이브러리에 로그인되어 있습니다. 또 다른 개발자 다이앤은 이러한 개별 아이디어를 하나의 멀티 에이전트 시스템으로 통합하려고 합니다.
그녀가 달성하고자 하는 목표는 다양한 에이전트 개발 프레임워크의 문제와 PR에 대한 최신 정보를 제공하는 주문형 연구 에이전트를 갖는 것입니다. 또한 생태계의 모든 새로운 개발 사항을 요약한 이메일을 보내기를 원합니다. 모든 에이전트는 서로 다른 환경에서 개별적으로 개발되므로 A2A는 이러한 기존 에이전트를 통합하는 데 이상적인 솔루션입니다.
일련의 원격 에이전트를 조립하는 첫 번째 단계는 필요한 각 원격 에이전트에게 에이전트 카드를 제공하는 것입니다. 이를 에이전트의 명함이라고 생각하면 됩니다. 호스트 에이전트가 이름, 설명, 기능, 입력/출력 스키마로 에이전트를 식별할 수 있도록 하는 데 사용되는 JSON 파일입니다.
먼저 프로젝트 루트에서 다음 명령어를 실행하여 cards 디렉터리와 에이전트 카드에 필요한 모든 JSON 파일을 만듭니다.
mkdir -p $HOME/app/cards/
touch $HOME/app/cards/github_retrieval_agent_card.json
touch $HOME/app/cards/content_evaluator_agent_card.json
touch $HOME/app/cards/emailer_agent_card.json
이러한 파일은 다음 bash 명령어를 통해서만 액세스할 수 있습니다.
cloudshell edit $HOME/app/cards/github_retrieval_agent_card.json
다음 JSON 콘텐츠를 app/cards/github_retrieval_agent_card.json 폴더에 복사합니다. 이 파일에 액세스할 수 있습니다.
{
"name": "GitHub_Retrieval_Agent",
"description": "Connects to the GitHub MCP server to retrieve real-time development insights, actively reading repository issues, pull requests, and commit histories.",
"url": "http://localhost:8001",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"json",
"text/plain"
],
"skills": [
{
"id": "read_github_repos",
"name": "GitHub_Retriever",
"description": "Fetches and analyzes recent content updates from GitHub repositories, including open/closed issues, pull request statuses, and detailed commit logs.",
"tags": [
"Find the newest pull requests from",
"Check recent issues in",
"Summarize commits for",
"GitHub repository status"
],
"examples": [
"Find the 10 most recent pull requests from the langchain-ai/langgraph repository along with their commits.",
"List all open issues tagged with 'bug' in the current repository."
]
}
]
}
다음 콘텐츠를 app/cards/content_evaluator_agent_card.json 파일에 복사합니다. 이렇게 하면 호스트 에이전트에게 이 에이전트가 할 수 있는 일, 제공할 수 있는 데이터의 종류, 반환할 데이터에 관한 설명이 제공됩니다.
이전과 마찬가지로 다음 명령어를 사용하여 평가자 에이전트 카드를 수정합니다.
cloudshell edit $HOME/app/cards/content_evaluator_agent_card.json
{
"name": "Content_Evaluation_Agent",
"description": "Distills and summarizes complex GitHub data, breaking down pull requests, code changes, and issue discussions into easily understandable insights.",
"url": "http://localhost:8002",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"defaultInputModes": [
"text",
"json",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"id": "evaluate_github_content",
"name": "Evaluate GitHub Content",
"description": "Analyzes and synthesizes provided GitHub data—including code diffs, commit histories, issue threads, and PR comments—into clear, structured summaries.",
"tags": [
"summarize pull request",
"evaluate GitHub changes",
"break down commits",
"distill issue discussion",
"code review summary"
],
"examples": [
"Make a thorough summary of the code changes and commit history from this pull request data.",
"Break down the main arguments and proposed solutions from this provided GitHub issue discussion."
]
}
]
}
다음은 이메일 에이전트의 에이전트 카드입니다. 동일한 Cloud Shell 명령어를 사용하여 app/cards/emailer_agent_card.json 파일에 복사합니다.
cloudshell edit $HOME/app/cards/emailer_agent_card.json
{
"name": "Email_Agent",
"description": "Acts as the final delivery mechanism in the pipeline, dispatching generated summaries and reports to designated email addresses.",
"url": "http://localhost:8003",
"capabilities": {
"streaming": true,
"pushNotifications": true,
"stateTransitionHistory": false
},
"defaultInputModes": [
"text",
"text/plain"
],
"defaultOutputModes": [
"text",
"text/plain"
],
"skills": [
{
"id": "dispatch_email_report",
"name": "Dispatch_Report_via_Email",
"description": "Takes synthesized text data and securely emails it to a specified recipient or distribution list.",
"tags": [
"send email",
"email summary",
"dispatch report",
"forward to inbox"
],
"examples": [
"Email me the summarized evaluations from the retrieved pull request and issue data.",
"Take this code review breakdown and send it in an email to the dev team."
]
}
]
}
7. 원격 에이전트 실행기 만들기
원격 에이전트가 호스트 에이전트에 의해 '호출'되거나 '대화'하려면 각 에이전트에는 A2A 서버에 표시되고 에이전트 카드로 식별되는 에이전트 실행기가 필요합니다. 실행기는 execute() 및 cancel() 메서드가 있는 추상 클래스로, 원격 에이전트를 호출하여 작업이나 메시지를 제공하거나 작업을 완전히 취소합니다.
다음은 메시지 전송과 원격 에이전트에 대한 작업 순서 지정이 모두 TextPart 아티팩트를 교환하는 방식으로 이루어지도록 지원하는 추상 ADK 에이전트 실행기의 구현입니다. 따라서 에이전트 간에 공유되는 공통 에이전트 실행기를 사용할 수 있습니다. 새 실행기 파일을 만듭니다.
touch $HOME/app/src/agents/executor.py
다음 코드 세그먼트를 새 $HOME/app/src/agents/executor.py 파일에 복사합니다.
# $HOME/app/src/agents/executor.py
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.agent_execution.context import RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import TaskState, TextPart, Part
from a2a.utils import new_agent_text_message, new_task
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
from google.genai import types
class BaseAgentExecutor(AgentExecutor):
"""An AgentExecutor that runs a remote ADK Agent"""
def __init__(
self,
agent,
status_message='Processing request...',
artifact_name='response',
):
"""Initialize a generic ADK agent executor.
Args:
agent: The ADK agent instance
status_message: Message to display while processing
artifact_name: Name for the response artifact
"""
self.agent = agent
self.status_message = status_message
self.artifact_name = artifact_name
self.runner = Runner(
app_name=agent.name,
agent=agent,
artifact_service=InMemoryArtifactService(),
session_service=InMemorySessionService(),
memory_service=InMemoryMemoryService(),
)
async def execute(
self,
context: RequestContext,
event_queue: EventQueue,
) -> None:
query = context.get_user_input()
task = context.current_task or new_task(context.message) # type: ignore
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
if context.call_context:
user_id = context.call_context.user.user_name
else:
user_id = "a2a_user"
try:
# Update status with custom message
await updater.update_status(
TaskState.working,
new_agent_text_message(
self.status_message,
task.context_id,
task.id
),
)
# Process with ADK agent
session = await self.runner.session_service.create_session(
app_name=self.agent.name,
user_id=user_id,
state={},
session_id=task.context_id,
)
content = types.Content(
role="user", parts=[types.Part.from_text(text=query)]
)
response_text = ""
async for event in self.runner.run_async(
user_id=user_id, session_id=session.id, new_message=content
):
if event.is_final_response() and event.content and event.content.parts:
for part in event.content.parts:
if hasattr(part, "text") and part.text:
response_text += part.text + "\n"
elif hasattr(part, "function_call"):
# Log or handle function calls if needed
pass # Function calls are handled internally by ADK
# Add response as artifact with custom name
await updater.add_artifact(
[Part(root=TextPart(text=response_text))],
name=self.artifact_name,
)
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.failed,
new_agent_text_message(f"Error: {e!s}", task.context_id, task.id),
final=True,
)
async def cancel(
self,
context: RequestContext,
event_queue: EventQueue
) -> None:
"""Cancel the execution of a specific task."""
raise NotImplementedError("Cancel not implemented for RetrieverAgentExecutor")
이제 추상 에이전트 실행기가 있으므로 Python 상속을 사용하여 각 에이전트의 특정 기능과 요구사항에 맞게 실행기를 미세 조정할 수 있습니다. 여기서는 페이로드로 에이전트를 호출하고, 응답을 기다리고, 응답 페이로드를 가져오는 실행이 워크플로에 보편적입니다. 따라서 특별한 변경사항은 필요하지 않습니다. 하지만 각 에이전트 A2A 서버에는 에이전트 실행기가 필요합니다. 다음 명령어를 실행하여 세 에이전트 모두의 실행기 파일을 만듭니다.
touch $HOME/app/src/agents/retriever/executor.py
touch $HOME/app/src/agents/evaluator/executor.py
touch $HOME/app/src/agents/emailer/executor.py
이제 각 파일에 해당하는 콘텐츠를 추가합니다.
# $HOME/app/src/agents/retriever/executor.py
from ..executor import BaseAgentExecutor
class RetrieverAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Retriever Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
# $HOME/app/src/agents/evaluator/executor.py
from ..executor import BaseAgentExecutor
class EvaluatorAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Evaluator Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
# $HOME/app/src/agents/emailer/executor.py
from ..executor import BaseAgentExecutor
class EmailerAgentExecutor(BaseAgentExecutor):
"""
An AgentExecutor that runs the Emailer Agent
All agent specific implementations for execute() and cancel() can be
overloaded here, along with any other desired funcitonality.
"""
pass
8. 원격 에이전트를 A2A 서버에 노출
이제 각 에이전트가 자체 명함을 가지므로 A2A 서버를 통해 엔드포인트에 노출되면 검색할 수 있습니다. 이 Codelab에서는 각 원격 에이전트 엔드포인트에 localhost를 사용하여 전체 프로세스를 로컬로 유지합니다. 하지만 실제로는 원하는 엔드포인트에 노출될 수 있으며, 에이전트 카드의 url 필드에 참조되는 한 검색 가능합니다.
각 원격 에이전트의 다음 단계는 자체 A2A 서버에 노출하는 것입니다. 다음 명령어를 실행하여 세 원격 에이전트 모두의 server.py 파일을 만듭니다.
touch $HOME/app/src/agents/retriever/server.py
touch $HOME/app/src/agents/evaluator/server.py
touch $HOME/app/src/agents/emailer/server.py
이제 각 에이전트의 서버 코드를 추가합니다. 리트리버 에이전트부터 시작합니다.
# $HOME/app/src/agents/retriever/server.py
import logging
import os
import json
import uvicorn
from a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as retriever_agent
from .executor import RetrieverAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/github_retrieval_agent_card.json", "r") as f:
card_data = json.load(f)
github_retrieval_agent_card = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=RetrieverAgentExecutor(
agent=retriever_agent
),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=github_retrieval_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8001))
print(f"Starting Retriever Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
그런 다음 평가 에이전트의 서버를 만듭니다.
# $HOME/app/src/agents/evaluator/server.py
import logging
import os
import json
import uvicorn
from a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as evaluator_agent
from .executor import EvaluatorAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/content_evaluator_agent_card.json", "r") as f:
card_data = json.load(f)
content_evaluator_agent_card = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=EvaluatorAgentExecutor(agent=evaluator_agent),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=content_evaluator_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8002))
print(f"Starting Evaluator Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
마지막으로 이메일 에이전트의 경우:
# $HOME/app/src/agents/emailer/server.py
import logging
import os
import json
import uvicorn
from a2a.types import AgentCard
from a2a.server.apps import A2AStarletteApplication
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.tasks import InMemoryTaskStore
from .agent import root_agent as emailer_agent
from .executor import EmailerAgentExecutor
logging.basicConfig(level=logging.INFO)
with open("cards/emailer_agent_card.json", "r") as f:
card_data = json.load(f)
emailer_agent_card = AgentCard(**card_data)
request_handler = DefaultRequestHandler(
agent_executor=EmailerAgentExecutor(
agent=emailer_agent
),
task_store=InMemoryTaskStore(),
)
server = A2AStarletteApplication(
http_handler=request_handler,
agent_card=emailer_agent_card,
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8003))
print(f"Starting Emailer Agent on Port {port}...")
uvicorn.run(server.build(), host="0.0.0.0", port=port)
9. 호스트 에이전트 빌드
지금까지 한 작업은 기본적으로 각 에이전트를 호스트 에이전트가 핑할 수 있는 API로 전환하는 것입니다. 이제 원격 에이전트가 자체 A2A 서버에 노출되었으므로 이를 검색하고 오케스트레이션하는 호스트 에이전트를 빌드해야 합니다.
이를 위해 호스트에서 구축해야 하는 몇 가지 측면이 있습니다. 먼저 각 원격 에이전트와 서버의 개별 클라이언트를 빌드하는 방법을 만들어야 합니다. 이는 호스트가 에이전트 카드를 검색할 수 있는 서버에서 호스팅되는 각 원격 에이전트 엔드포인트에 연결을 설정하는 A2A 클라이언트 팩토리를 만들어 수행됩니다. 호스트와 각 원격 에이전트 간의 연결은 RemoteAgentConnections 객체로 초기화할 수 있습니다.
touch $HOME/app/src/host/remote_agent_connection.py
touch $HOME/app/src/host/agent.py
touch $HOME/app/src/host/__init__.py
echo "from . import agent" >> $HOME/app/src/host/__init__.py
원격 에이전트 연결의 다음 구현을 src/host/remote_agent_connection.py 파일에 복사합니다.
# src/host/remote_agent_connection.py
import traceback
from a2a.client import (
Client,
ClientFactory,
)
from a2a.types import (
AgentCard,
Message,
Task,
TaskState,
)
class RemoteAgentConnection:
"""A class to hold the connections to the remote agents."""
def __init__(self, client_factory: ClientFactory, agent_card: AgentCard):
self.agent_client: Client = client_factory.create(agent_card)
self.card: AgentCard = agent_card
def get_agent(self) -> AgentCard:
return self.card
async def send_message(self, message: Message) -> Task | Message | None:
lastTask: Task | None = None
try:
async for event in self.agent_client.send_message(message):
if isinstance(event, Message):
return event
if self.is_terminal_or_interrupted(event[0]):
return event[0]
lastTask = event[0]
except Exception as e:
print('Exception found in send_message')
traceback.print_exc()
raise e
return lastTask
def is_terminal_or_interrupted(self, task: Task) -> bool:
return task.status.state in [
TaskState.completed,
TaskState.canceled,
TaskState.failed,
TaskState.input_required,
TaskState.unknown,
]
이제 지정된 클라이언트와 상담사 카드를 통해 클라이언트와 서버 간의 연결을 설정할 수 있습니다. 이는 호스트 에이전트가 원격 에이전트 서버와의 연결을 설정하는 데 사용하는 도구입니다.
이제 호스트 에이전트를 만들어 보겠습니다. 먼저 다음 코드 세그먼트를 app/src/host/agent.py 파일에 복사합니다. 원격 에이전트 엔드포인트와 표준 httpx 클라이언트가 필요한 Pythonic 클래스의 초기화입니다. 이 에이전트 클래스를 초기화하면 제공된 원격 주소를 통해 연결이 설정됩니다.
import asyncio
import json
import os
import uuid
import httpx
from typing import Any
from a2a.client import A2ACardResolver, ClientConfig, ClientFactory
from a2a.types import (
AgentCard,
Message,
Part,
Role,
Task,
TaskState,
TextPart,
TransportProtocol,
)
from google.adk import Agent
from google.adk.agents.callback_context import CallbackContext
from google.adk.agents.readonly_context import ReadonlyContext
from google.adk.tools.tool_context import ToolContext
from dotenv import load_dotenv
from .remote_agent_connection import RemoteAgentConnection
load_dotenv()
######################################
# --- Coordinator Agent Definition ---
######################################
class CoordinatorAgent:
"""
The Coordinator agent.
This is the agent responsible for sending tasks to agents.
"""
def __init__(
self,
remote_agent_addresses: list[str],
http_client: httpx.AsyncClient,
):
self.http_client = http_client
self.remote_agent_addresses = remote_agent_addresses
self.client_factory = None
self.remote_agent_connections: dict[str, RemoteAgentConnection] = {}
self.cards: dict[str, AgentCard] = {}
self.agents: str = ''
호스트 에이전트의 다음 측면은 원격 에이전트에 대한 연결을 설정하는 것입니다. __init__ (이벤트 루프가 존재하기 전 모듈 가져오기 시간에 실행됨)에서 연결하는 대신 ensure_initialized 메서드는 에이전트가 실제로 사용될 때까지 이 작업을 지연합니다. 연결이 이미 설정되었는지 확인하고, 설정되지 않은 경우 A2A 클라이언트를 생성하고 각 원격 에이전트에 병렬로 연결합니다. 이 세그먼트를 app/src/host/agent.py의 이전 세그먼트 바로 아래에 복사합니다.
#####################################
# --- Remote Agent Initialization ---
#####################################
async def ensure_initialized(self):
if not self.remote_agent_connections:
config = ClientConfig(
httpx_client=self.http_client,
supported_transports=[
TransportProtocol.jsonrpc,
TransportProtocol.http_json,
],
)
self.client_factory = ClientFactory(config=config)
async with asyncio.TaskGroup() as task_group:
for address in self.remote_agent_addresses:
task_group.create_task(self.retrieve_card(address))
async def retrieve_card(self, address: str):
card_resolver = A2ACardResolver(self.http_client, address)
card = await card_resolver.get_agent_card()
card.url = address # Use the actual address, not the hardcoded URL in the card JSON
remote_connection = RemoteAgentConnection(self.client_factory, card)
self.remote_agent_connections[card.name] = remote_connection
self.cards[card.name] = card
agent_info = []
for card in self.cards.values():
agent_info.append(
json.dumps({"name": card.name, "description": card.description})
)
self.agents = "\n".join(agent_info)
다음은 LLM 에이전트 구현입니다. 이 경우 오케스트레이터에 ADK 에이전트를 사용하므로 프롬프트 지침, 콜백, 도구와 같은 일반적인 부가기능이 필요합니다. app/src/host/agent.py 파일의 호스트 에이전트 클래스 내에 이러한 구성요소를 모두 복사합니다. 다음 콜백을 호스트 에이전트 클래스에 넣습니다. 에이전트가 제대로 트리거되기 전에 이 콜백은 에이전트의 상태에서 아직 초기화되지 않은 경우 에이전트를 초기화합니다.
############################
# -- Implement the ADK Agent
############################
# --- Before Model Callback ---
def before_model_callback(
self, callback_context: CallbackContext, llm_request
):
"""
A callback to set up the session state before the model processes the
request
"""
state = callback_context.state
if 'session_active' not in state or not state['session_active']:
if 'session_id' not in state:
state['session_id'] = str(uuid.uuid4())
state['session_active'] = True
호스트 에이전트의 다음 구성요소는 프롬프트 명령어입니다. 이것은 오케스트레이터 에이전트이므로 코디네이터에게 세션에서 현재 어떤 일이 일어나고 있는지 알려주기 위해 사용할 수 있는 원격 에이전트와 활성 상태인 현재 에이전트를 알아야 합니다. 콜백 아래에 다음 프롬프트와 도우미 함수를 추가합니다.
# --- Prompt ---
def root_instruction(self, context: ReadonlyContext) -> str:
current_agent = self.check_state(context)
return f"""
You are an expert orchestrator that can delegate user requests to the
appropriate remote agents to generate a GitHub research report.
**Your Goal:** To fulfill user requests for GitHub repository data, evaluate it, and optionally email a report.
**Workflow Steps:**
1. **Understand the User Request**:
- Identify repository names (e.g., "google/adk-python").
- Determine if the user wants Pull Requests, Issues, or both.
- Extract any specified email address for sending the report (e.g., "user@example.com").
- Note the number of PRs/issues requested per repository. If not specified, the Retrieval Agent has defaults.
2. **Retrieve Data (GitHub_Retrieval_Agent)**:
- Use the `send_message` tool to send a message to the "GitHub_Retrieval_Agent".
- Your message to the Retrieval Agent should clearly state which repositories to fetch data for, and specify if you need issues, pull requests, and the respective limits.
- Example message to Retrieval Agent: "Fetch 5 issues and 3 pull requests for google/adk-python."
3. **Evaluate and Summarize Data (Content_Evaluation_Agent)**:
- Once you receive the raw JSON data from the "GitHub_Retrieval_Agent", use the `send_message` tool to send this data to the "Content_Evaluation_Agent".
- Your message to the Evaluation Agent should include the raw JSON data you received.
- The Evaluation Agent will return a Markdown-formatted report.
4. **Email Report (Email_Agent - if email provided)**:
- If the user's initial request included an email address, use the `send_message` tool to send the Markdown report from the "Content_Evaluation_Agent" to the "Email_Agent".
- Your message to the Email Agent should include the report and the recipient's email address.
- Example message to Email Agent: "Send this report to user@example.com: [Markdown Report Content]".
5. **Respond to User**:
- Based on the outcome of the steps above, formulate a concise and informative response to the user.
- If a report was generated and emailed, confirm that. If only a report was generated, provide it directly to the user.
- If an email address was requested but the email failed to send, inform the user.
- If any step failed, inform the user about the failure.
**Available Tools:**
- `list_remote_agents()`: Use this to see what agents are available (though you already know their names for this workflow).
- `send_message(agent_name: str, message: str)`: Use this to interact with remote agents.
**Crucial Guidelines:**
- **Rely on Tools**: ALWAYS use `send_message` to interact with the remote agents. Do NOT attempt to perform the tasks yourself.
- **No Conversational Filler**: Only communicate the essential information to the remote agents or back to the user.
- **Error Handling**: If a remote agent returns an error, acknowledge it and try to provide a helpful message to the user.
Agents:
{self.agents}
Current agent: {current_agent['active_agent']}
"""
def check_state(self, context: ReadonlyContext):
state = context.state
if (
'context_id' in state
and 'session_active' in state
and state['session_active']
and 'agent' in state
):
return {'active_agent': f'{state["agent"]}'}
return {'active_agent': 'None'}
이제 호스트 에이전트에서 가장 중요한 부분인 도구가 나옵니다. 호스트는 send_message() 도구와 list_remote_agents() 도구에 액세스할 수 있습니다. list_remote_agent() 도구는 더 간단합니다. 클래스의 에이전트 카드를 읽어 각 에이전트의 이름과 설명이 포함된 사전 목록을 반환합니다. send_message()는 좀 더 고급입니다. 에이전트 이름과 메시지 문자열이 주어지면 메시지나 작업, 스트리밍 또는 비스트리밍을 원격 에이전트에 전송할 수 있습니다. 이 코드 세그먼트를 app/src/host/agent.py의 프롬프트 섹션 아래에 있는 동일한 호스트 에이전트 Python 클래스 내에 넣습니다.
# --- Agent Tools ---
def list_remote_agents(self):
"""List the available remote agents you can use to delegate the task."""
if not self.remote_agent_connections:
return []
remote_agent_info = []
for card in self.cards.values():
remote_agent_info.append(
{'name': card.name, 'description': card.description}
)
return remote_agent_info
async def send_message(
self, agent_name: str, message: str, tool_context: ToolContext
):
"""Sends a task either streaming (if supported) or non-streaming.
This will send a message to the remote agent named agent_name.
Args:
agent_name: The name of the agent to send the task to.
message: The message to send to the agent for the task.
tool_context: The tool context this method runs in.
Yields:
A dictionary of JSON data.
"""
await self.ensure_initialized()
if agent_name not in self.remote_agent_connections:
raise ValueError(f'Agent {agent_name} not found')
state = tool_context.state
state['agent'] = agent_name
client = self.remote_agent_connections[agent_name]
if not client:
raise ValueError(f'Client not available for {agent_name}')
task_id = state.get('task_id', None)
context_id = state.get('context_id', None)
message_id = state.get('message_id', None)
task: Task
if not message_id:
message_id = str(uuid.uuid4())
request_message = Message(
role=Role.user,
parts=[Part(root=TextPart(text=message))],
message_id=message_id,
context_id=context_id,
task_id=task_id,
)
response = await client.send_message(request_message)
if isinstance(response, Message):
return await convert_parts(response.parts, tool_context)
task: Task = response # type: ignore
# Assume completion unless a state returns that isn't complete
state['session_active'] = not client.is_terminal_or_interrupted(task)
if task.context_id:
state['context_id'] = task.context_id
state['task_id'] = task.id
if task.status.state == TaskState.input_required:
# Force user input back
tool_context.actions.skip_summarization = True
tool_context.actions.escalate = True
elif task.status.state == TaskState.canceled:
# Open question, should we return some info for cancellation instead
raise ValueError(f'Agent {agent_name} task {task.id} is cancelled')
elif task.status.state == TaskState.failed:
# Raise error for failure
raise ValueError(f'Agent {agent_name} task {task.id} failed')
response = []
if task.status.message:
response.extend(
await convert_parts(task.status.message.parts, tool_context)
)
if task.artifacts:
for artifact in task.artifacts:
response.extend(
await convert_parts(artifact.parts, tool_context)
)
return response
마지막으로 이러한 모든 구성요소를 통합하는 ADK 에이전트를 구현합니다. 이는 BeforeModelCallback이 실행된 후 사용할 수 있는 도구를 사용하여 제공된 프롬프트를 실행하는 오케스트레이터 클래스 에이전트의 브레인입니다.
다음 코드 세그먼트를 app/src/host/agent.py 파일의 도구 섹션 바로 아래에 배치합니다.
def create_agent(self) -> Agent:
"""Create an instance of the CoordinatorAgent."""
return Agent(
model='gemini-2.5-flash',
name='host',
instruction=self.root_instruction,
before_model_callback=self.before_model_callback,
description=(
'This coordinator agent orchestrates the retriever, evaluator, and emailer agents'
),
tools=[
self.send_message,
self.list_remote_agents,
],
)
에이전트에는 A2A 부분을 일반 텍스트와 데이터로 변환하기 위한 send_message() 도구의 도우미 함수가 몇 개 필요합니다. CoordinatorAgent 클래스 외부에 이 세그먼트를 복사합니다.
##########################
# --- Helper Functions ---
##########################
async def convert_part(part: Part, tool_context: ToolContext):
"""Convert a part to text. Only text parts are supported."""
if isinstance(part.root, TextPart):
return part.root.text
if part.root.kind == "data":
return part.root.data
return f"Unknown type: {part}"
async def convert_parts(parts: list[Part], tool_context: ToolContext):
"""Convert parts to text."""
rval = []
for p in parts:
rval.append(await convert_part(p, tool_context))
return rval
uv run adk web 또는 adk run를 통한 로컬 테스트 및 상호작용을 용이하게 하려면 app/src/host/agent.py 파일 하단에 root_agent 초기화를 추가합니다.
root_agent = CoordinatorAgent(
remote_agent_addresses=[
os.getenv('RETRIEVAL_AGENT_URL', 'http://localhost:8001'),
os.getenv('EVALUATOR_AGENT_URL', 'http://localhost:8002'),
os.getenv('EMAILER_AGENT_URL', 'http://localhost:8003'),
],
http_client=httpx.AsyncClient(timeout=30),
).create_agent()
축하합니다. 첫 번째 A2A 호스트 에이전트를 만들었습니다. root_agent 변수로 초기화했으므로 다른 원격 에이전트와 마찬가지로 로컬 uv run adk 웹 배포를 통해 상호작용할 수 있습니다. 다음 명령어를 사용하여 호스트와 상호작용합니다.
cd $HOME/app/src/
uv run adk run host
10. 로컬에서 테스트
전체 멀티 에이전트 시스템을 테스트하려면 호스트 에이전트가 사용할 수 있는 원격 에이전트 서버를 시작해야 합니다. 이렇게 하면 모든 에이전트 서버가 한 번에 시작됩니다.
# Make sure you have activated your virtual environment first!
# source .venv/bin/activate
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Function to kill all background processes
cleanup() {
echo "Caught signal, terminating background processes..."
# The negative PID kills the entire process group
kill -TERM -$$
wait
echo "All processes terminated."
exit
}
# Trap TERM, INT, and EXIT signals and call the cleanup function
trap cleanup TERM INT EXIT
# Activate virtual environment
uv sync
# Start the agent servers in the background
echo "Starting agent servers..."
uv run python3 -m src.agents.retriever.server --port 8001 &
uv run python3 -m src.agents.evaluator.server --port 8002 &
uv run python3 -m src.agents.emailer.server --port 8003 &
cd $HOME/app/src/
RETRIEVER_AGENT_URL=http://localhost:8001
EVALUATOR_AGENT_URL=http://localhost:8002
EMAILER_AGENT_URL=http://localhost:8003
uv run adk run host
# Wait for all background processes to finish
wait
이 bash 명령어는 네 개의 에이전트 서버 (검색기, 평가자, 이메일러, 호스트)를 모두 시작합니다. 터미널에 각 서버의 로그 출력이 표시됩니다. 서버가 실행되면 새 터미널 창을 열고 (동일한 app 디렉터리에 있고 가상 환경이 활성화되어 있는지 확인) uv run adk 웹 인터페이스를 실행합니다.
# In a new terminal, from the 'app' directory
source .venv/bin/activate
cd $HOME/app/src/
uv run adk web
터미널에 표시되는 localhost 링크를 엽니다. 여기에서 왼쪽 상단의 드롭다운 메뉴에서 호스트 에이전트를 선택하여 직접 상호작용할 수 있습니다.
11. 삭제
요금이 계속 청구되지 않도록 이 Codelab에서 만든 리소스를 삭제합니다.
실행 중인 모든 에이전트 서버를 중지한 다음 이 Codelab을 위해 특별히 만든 프로젝트가 있다면 삭제합니다. 원격 에이전트를 실행한 터미널로 이동하여 Ctrl+C를 실행한 다음 이 Google Cloud 프로젝트를 삭제합니다.
gcloud projects delete ${GOOGLE_CLOUD_PROJECT}
12. 축하합니다
Agent2Agent로 멀티 에이전트 시스템을 빌드했습니다.
학습한 내용
- 자체 도구를 사용하는 독립적인 ADK 에이전트를 만드는 방법
- 상담사 카드로 상담사가 검색 가능한 ID를 제공하는 방법
- A2A 서버를 통해 에이전트를 노출하는 방법
- 원격 에이전트를 오케스트레이션하는 호스트 에이전트를 빌드하는 방법
- A2A 프로토콜이 독립적으로 개발된 에이전트 간의 통신을 지원하는 방식
다음 단계
- 프로덕션용으로 Cloud Run에 시스템 배포
- 원격 에이전트를 더 추가하여 시스템 기능 확장
- A2A의 스트리밍 및 푸시 알림 기능 살펴보기