1. Genel Bakış
Bu codelab'de, Agent2Agent (A2A) protokolünü kullanarak birden fazla ADK aracısının iletişim kurduğu ve işbirliği yaptığı bir çoklu aracı sistemi oluşturacaksınız.
Neler öğreneceksiniz?
- Birden fazla bağımsız ADK aracısı oluşturma
- Her ajana nasıl Ajan Kartı verilir ve A2A sunucusu olarak nasıl sarmalanır?
- Uzak aracılarınızı düzenleyen bir ana makine aracısı oluşturma
- Uzak aracı bağlantıları oluşturma
- Çok agentlı sistemi yerel olarak test etme
İhtiyacınız olanlar
- Faturalandırmanın etkin olduğu bir Google Cloud projesi
- Chrome gibi bir web tarayıcısı
- Cloud Shell yerine kodu kendi makinenizde çalıştırıyorsanız Python 3.12+ ve uv
Bu codelab, Python ve Google Cloud hakkında bilgi sahibi olan orta düzey geliştiriciler içindir.
Bu codelab'in tamamlanması yaklaşık 15 dakika sürer.
Bu codelab'de oluşturulan kaynakların maliyeti 5 ABD dolarından az olmalıdır.
2. Ortamınızı ayarlama
Google Cloud projesi oluşturma
- Google Cloud Console'daki proje seçici sayfasında bir Google Cloud projesi seçin veya oluşturun.
- Cloud projeniz için faturalandırmanın etkinleştirildiğinden emin olun. Bir projede faturalandırmanın etkin olup olmadığını nasıl kontrol edeceğinizi öğrenin.
Cloud Shell Düzenleyici'yi başlatma
Google Cloud Console'dan Cloud Shell oturumu başlatmak için Google Cloud Console'da Cloud Shell'i etkinleştir'i tıklayın.
Bu işlem, Google Cloud Console'unuzun alt bölmesinde bir oturum başlatır.
Düzenleyiciyi başlatmak için Cloud Shell penceresinin araç çubuğunda Open Editor'ı (Düzenleyiciyi Aç) tıklayın.
API'leri etkinleştir
Terminalde, ADK'nın Gemini'ı çağırmak için kullandığı API'yi etkinleştirmek üzere aşağıdaki komutu çalıştırın:
gcloud services enable aiplatform.googleapis.com
Ortamınızı yapılandırma
A2A sisteminiz için proje klasörü yapısını oluşturmak üzere terminalinizde aşağıdaki komutu çalıştırarak başlayın. Bu demoda, $HOME dizininizden mutlak yol oluşturma yöntemini kullanacağız:
mkdir -p $HOME/app/src/agents $HOME/app/src/host
touch $HOME/app/.env $HOME/app/pyproject.toml
Genel mimariyi oluşturduğumuza göre şimdi ortam yapılandırmalarını dolduralım. Aşağıdaki kod segmentini yeni .env dosyasına kopyalayın:
Yeni .env dosyanızı Google Cloud proje kimliğinizle doldurun. GOOGLE_CLOUD_LOCATION değerini global olarak ayarlayın. Bu, bu codelab'deki Gemini modelinin kullanılabildiği yerdir. Oluşturmak üzere olduğunuz aracıdan rapor e-postaları almak istiyorsanız MAIL_TO alanına istediğiniz bir e-posta adresini de girebilirsiniz. Ayrıca, GitHub alma aracını kolaylaştırmak için sürekli olarak bir GitHub kişisel erişim jetonu GITHUB_TOKEN ekleyebilirsiniz.
Yeni .env dosyanızı aşağıdaki bash komutuyla açın:
cloudshell edit $HOME/app/.env
Ardından aşağıdaki dosyayı app/.env konumundaki .env dosyasına kopyalayın. ÖNEMLİ NOT: KENDİ değerlerinizi girdiğinizden emin olun.
# --- Google Cloud ---
GOOGLE_GENAI_USE_ENTERPRISE=True
GOOGLE_CLOUD_PROJECT=<your-gcp-project-id>
GOOGLE_CLOUD_LOCATION=global
# --- 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>
GITHUB_MCP_URL=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
Ortam değişkenlerinizi doldurduktan sonra uv ortamımızı yapılandırmamız gerekir. pyproject.toml dosyasını açın:
cloudshell edit $HOME/app/pyproject.toml
Aşağıdaki kod segmentini pyproject.toml dosyasına kopyalayın:
[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"
dependencies = [
"markdown",
"resend",
"uvicorn",
"a2a-sdk",
"google-adk[a2a,mcp]",
"httpx",
"python-dotenv",
]
[project.optional-dependencies]
test = [
"pytest",
"pytest-asyncio",
"pytest-mock",
"respx",
]
[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/"]
Bağımlıları yükleme
Şimdi terminalinizde aşağıdaki komutları çalıştırarak yeni oluşturduğunuz app dizinine gidin ve pyproject.toml dosyasından gerekli tüm bağımlılıkları yükleyin:
cd $HOME/app
uv sync
uv, bu codelab için izole bir Python ortamı oluşturur. Bu nedenle herhangi bir şeyi etkinleştirmeniz gerekmez. Python komutlarının önüne uv run ekleyin.
Artık çoklu ajan sistemimizi oluşturmak için her şey hazır.
3. Retriever Agent'ı oluşturma
Eva, yeni sorunlar ve PR'ler ile güncellenen GitHub depolarından haberdar olmak isteyen bir yazılım geliştiricidir. Bu nedenle, istediği GitHub verilerini almak için bir ADK aracısı oluşturur.
Bu demoda, alıcı aracısı için gerekli dizini ve dosyayı oluşturmak üzere projenizin kökünden aşağıdaki komutu çalıştırın:
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
Aşağıdaki ADK kodu segmentini $HOME/app/src/agents/retriever/agent.py dosyasına kopyalayın:
import logging
import os
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
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(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
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",
},
)
)
],
)
Bu ajan bağımsız bir araç olarak çalışır. Bağımsız olarak test etmek için agents dizininden adk run kullanarak sohbet etmek üzere aşağıdaki komutları çalıştırın:
cd $HOME/app/src/agents
uv run adk run retriever
Denemek için [user]: istemine bir mesaj yazın, ardından aracıyı durdurmak için exit yazın.
4. Değerlendirici Ajanı oluşturma
Richard, teknik olmayan müşterileri ve iş arkadaşları için genellikle çok fazla teknik jargonu basitleştirmesi gerektiğini düşünüyor. Aynı terimleri tanımlamaktan ve aynı projeyi özetlenmiş bir şekilde açıklamaktan bıkan Richard, teknik terminolojiyi kolayca anlaşılabilir bir metin haline getiren bir özetleme ajanı oluşturur.
Bu demo için değerlendirici aracının dosyasını oluşturmak üzere aşağıdaki komutu çalıştırın:
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
Aşağıdaki aracı kod segmentini $HOME/app/src/agents/evaluator/agent.py içine kopyalayın.
import json
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
load_dotenv()
# --- 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(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
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.
""",
)
Bu ajan bağımsız bir araç olarak çalışır. Bağımsız olarak test etmek için YENİ bir terminalde aşağıdaki komutları çalıştırarak agents dizinindeki adk run ile sohbet edin:
cd $HOME/app/src/agents
uv run adk run evaluator
Denemek için [user]: istemine bir mesaj yazın, ardından aracıyı durdurmak için exit yazın.
5. E-posta Gönderen Ajanı oluşturma
İvan, kolayca erişilebilen bir metni özetleyip yeniden biçimlendirmesi gereken e-postaları yazmaktan bıktı. Bu nedenle, belirli bir metni yeniden biçimlendirip sağlanan e-posta hesabına gönderecek bir e-posta aracı oluşturdu.
Bu demoda, e-posta gönderen aracısı için dosyayı oluşturmak üzere terminalinizde aşağıdaki komutu çalıştırın:
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
Aşağıdaki aracı kod segmentini app/src/agents/emailer/agent.py içine kopyalayın.
import logging
import os
import markdown
import resend
from dotenv import load_dotenv
from google.adk.agents.llm_agent import Agent
from google.adk.models import Gemini
from google.genai import types
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 = markdown.markdown(body, extensions=["tables"])
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(
model="gemini-3.8-flash",
retry_options=types.HttpRetryOptions(attempts=5),
),
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
],
)
Bu ajan bağımsız bir araç olarak çalışır. Bağımsız olarak test etmek için YENİ bir terminalde aşağıdaki komutları çalıştırarak agents dizinindeki adk run ile sohbet edin:
cd $HOME/app/src/agents
uv run adk run emailer
Denemek için [user]: istemine bir mesaj yazın, ardından temsilciyi durdurmak için exit yazın.
6. Ajan kartları oluşturma
Şimdi, her biri kendi benzersiz temsilcisine sahip üç bağımsız geliştiricinin hikayesini inceledik. Bu aracıların tümü yayınlanmış ve şirketlerinin Aracı Kitaplığı'na giriş yapmıştır. Başka bir geliştirici olan Diane, bu bağımsız fikirleri tek bir çoklu ajan sisteminde bir araya getirmek istiyor.
Amacı, farklı aracı geliştirme çerçevelerindeki sorunlar ve PR'ler hakkında kendisini güncel tutan bir talep üzerine araştırma aracına sahip olmaktır. Ayrıca, ekosistemdeki tüm yeni gelişmelerin özetini e-posta olarak almak istiyor. Tüm aracıların farklı ortamlarda ayrı ayrı geliştirilmesi nedeniyle A2A, mevcut aracıları bir araya getirmek için ideal bir çözümdür.
Bir dizi uzak aracı oluşturmanın ilk adımı, gerekli uzak aracıların her birine bir aracı kartı vermektir. Bunları, temsilcinizin kartvizitleri olarak düşünebilirsiniz. Bunlar, ana makine temsilcinizin temsilcileri adlarına, açıklamalarına, özelliklerine ve giriş/çıkış şemalarına göre tanımlamasını sağlayan JSON dosyalarıdır. Bu codelab'deki kartlar, A2A v1.0 Agent Card biçimini kullanır. Bu biçimde, aracının supportedInterfaces içinde ulaşılabileceği her uç nokta listelenir.
İlk olarak, cards dizinini ve aracı kartları için gerekli tüm JSON dosyalarını oluşturmak üzere proje kökünüzden aşağıdaki komutu çalıştırın:
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
Bu dosyalara yalnızca aşağıdaki bash komutuyla erişilebilir:
cloudshell edit $HOME/app/cards/github_retrieval_agent_card.json
Aşağıdaki JSON içeriğini app/cards/github_retrieval_agent_card.json dosyasına kopyalayın:
{
"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.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8001",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"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."
]
}
]
}
Aşağıdaki içeriği app/cards/content_evaluator_agent_card.json dosyasına kopyalayın. Bu işlem, ana makine aracısına bu aracının neler yapabileceği, hangi tür veriler verebileceğiniz ve hangi verileri döndüreceği hakkında bir açıklama sağlar.
Değerlendirici aracısı kartını düzenlemek için öncekiyle aynı komutu kullanın:
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.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8002",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"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."
]
}
]
}
E-posta gönderen temsilcinin temsilci kartını aşağıda bulabilirsiniz. Aynı Cloud Shell komutunu kullanarak app/cards/emailer_agent_card.json dosyasına kopyalayın:
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.",
"version": "1.0.0",
"supportedInterfaces": [
{
"url": "http://localhost:8003",
"protocolBinding": "JSONRPC",
"protocolVersion": "1.0"
}
],
"capabilities": {
"streaming": true,
"pushNotifications": true
},
"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. Uzak aracı yürütücüler oluşturma
Uzak aracıların bir ana aracı tarafından "çağrılması" veya "konuşulması" için her birinin, A2A sunucusuna görünür bir aracı yürütücüsü olması ve aracı kartıyla tanımlanması gerekir. Yürütücü, uzak aracıyı çağıran ve ona bir görev veya mesaj sunan ya da görevi tamamen iptal eden execute() ve cancel() yöntemlerine sahip soyut bir sınıftır.
Burada, uzak temsilcilere mesaj göndermeyi ve görevleri sıralamayı kolaylaştıran, metin Part yapılarını değiştiren soyut bir ADK temsilci yürütücüsünün uygulaması yer almaktadır. Bu nedenle, temsilciler arasında paylaşılan ortak bir temsilci yürütücü kullanabiliriz. Yeni bir yürütücü dosyası oluşturun:
touch $HOME/app/src/agents/executor.py
Aşağıdaki kod segmentini yeni $HOME/app/src/agents/executor.py dosyasına kopyalayın:
# $HOME/app/src/agents/executor.py
from a2a.helpers import new_task_from_user_message, new_text_message
from a2a.server.agent_execution import AgentExecutor, RequestContext
from a2a.server.events.event_queue import EventQueue
from a2a.server.tasks import TaskUpdater
from a2a.types import Part, TaskState
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_from_user_message(context.message) # type: ignore
await event_queue.enqueue_event(task)
updater = TaskUpdater(event_queue, task.id, task.context_id)
user_id = context.call_context.user.user_name or "a2a_user"
try:
# Update status with custom message
await updater.update_status(
TaskState.TASK_STATE_WORKING,
new_text_message(
self.status_message,
context_id=task.context_id,
task_id=task.id,
),
)
# Process with ADK agent, reusing the session for this context
session = await self.runner.session_service.get_session(
app_name=self.agent.name,
user_id=user_id,
session_id=task.context_id,
)
if session is None:
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(text=response_text)],
name=self.artifact_name,
)
await updater.complete()
except Exception as e:
await updater.update_status(
TaskState.TASK_STATE_FAILED,
new_text_message(
f"Error: {e!s}",
context_id=task.context_id,
task_id=task.id,
),
)
async def cancel(
self,
context: RequestContext,
event_queue: EventQueue
) -> None:
"""Cancel the execution of a specific task."""
raise NotImplementedError("Cancel not implemented for RetrieverAgentExecutor")
Artık bu soyut aracı yürütücüsüne sahip olduğumuza göre, yürütücüyü her aracının belirli özelliklerine ve ihtiyaçlarına göre hassaslaştırmak için Python'a özgü devralma özelliğini kullanabiliriz. Bizim durumumuzda, yük ile temsilciyi çağırma, yanıt bekleme ve yanıt yükünü alma işlemleri iş akışımız için evrenseldir. Bu nedenle, herhangi bir değişiklik yapılması gerekmez. Ancak, A2A aracısı sunucularının her birinde bir aracı yürütücü olması gerekir. Üç aracının tümü için yürütücü dosyalarını oluşturmak üzere aşağıdaki komutu çalıştırın:
touch $HOME/app/src/agents/retriever/executor.py
touch $HOME/app/src/agents/evaluator/executor.py
touch $HOME/app/src/agents/emailer/executor.py
Şimdi, her dosyaya ilgili içeriği ekleyin.
# $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. Uzak aracıları A2A sunucularına sunma
Artık her temsilcinin kendi kartviziti olduğundan, A2A sunucusu aracılığıyla bir uç noktaya maruz kaldıklarında keşfedilebilirler. Bu codelab'de, tüm süreci yerel tutmak için her bir uzak aracı uç noktası için localhost'lar kullanıyoruz. Ancak uygulamada bunlar istenen herhangi bir uç noktaya sunulabilir ve aracı kartlarının url alanında supportedInterfaces olarak referans verildiği sürece bulunabilir.
Her sunucu, temsilci kartını /.well-known/agent-card.json konumunda sunar. A2A istemcileri, temsilci kartını bu konumda arar ve / konumunda A2A JSON-RPC isteklerini işler.
Uzak aracıların her biri için bir sonraki adım, bunları kendi A2A sunucularına göstermektir. Üç uzak aracı için de server.py dosyasını oluşturmak üzere bu komutu çalıştırın:
touch $HOME/app/src/agents/retriever/server.py
touch $HOME/app/src/agents/evaluator/server.py
touch $HOME/app/src/agents/emailer/server.py
Şimdi her aracı için sunucu kodunu ekleyeceksiniz. Alıcı aracısıyla başlayın:
# $HOME/app/src/agents/retriever/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
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 = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=RetrieverAgentExecutor(
agent=retriever_agent
),
task_store=InMemoryTaskStore(),
agent_card=github_retrieval_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(github_retrieval_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8001))
print(f"Starting Retriever Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
Ardından, değerlendirme aracısı için sunucuyu oluşturun:
# $HOME/app/src/agents/evaluator/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
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 = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=EvaluatorAgentExecutor(agent=evaluator_agent),
task_store=InMemoryTaskStore(),
agent_card=content_evaluator_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(content_evaluator_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8002))
print(f"Starting Evaluator Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
Son olarak, e-posta gönderen aracısı için:
# $HOME/app/src/agents/emailer/server.py
import logging
import os
import json
import uvicorn
from google.protobuf.json_format import ParseDict
from starlette.applications import Starlette
from a2a.types import AgentCard
from a2a.server.request_handlers import DefaultRequestHandler
from a2a.server.routes import create_agent_card_routes, create_jsonrpc_routes
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 = ParseDict(card_data, AgentCard())
request_handler = DefaultRequestHandler(
agent_executor=EmailerAgentExecutor(
agent=emailer_agent
),
task_store=InMemoryTaskStore(),
agent_card=emailer_agent_card,
)
server = Starlette(
routes=[
*create_agent_card_routes(emailer_agent_card),
*create_jsonrpc_routes(request_handler, rpc_url="/"),
]
)
if __name__ == "__main__":
port = int(os.getenv("PORT", 8003))
print(f"Starting Emailer Agent on Port {port}...")
uvicorn.run(server, host="0.0.0.0", port=port)
9. Ana makine aracısını oluşturma
Şimdiye kadar yaptığımız şey, her bir aracımızı ana aracının ping atabileceği bir API'ye dönüştürmek oldu. Uzak aracılarımız kendi A2A sunucularına maruz kaldığına göre, bunları keşfedip düzenleyen bir ana makine aracısı oluşturmamız gerekiyor.
Bunu gerçekleştirmek için ana makinenin birkaç farklı yönünü oluşturmamız gerekir. Öncelikle, her uzak aracımız ve sunucusu için ayrı istemciler oluşturmanın bir yolunu bulmamız gerekiyor. Bu işlem, ana makinenin aracı kartlarını keşfedebileceği, sunucuları tarafından barındırılan her bir uzak aracı uç noktasına bağlantı oluşturan bir A2A istemci fabrikası oluşturularak yapılır. Ana makine ile her bir uzak aracı arasındaki bağlantılar, RemoteAgentConnections nesnesiyle başlatılabilir.
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
Aşağıdaki uzak aracı bağlantısı uygulamasını src/host/remote_agent_connection.py dosyasına kopyalayın:
# src/host/remote_agent_connection.py
import traceback
from a2a.client import (
Client,
ClientFactory,
)
from a2a.types import (
AgentCard,
Message,
SendMessageRequest,
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(
SendMessageRequest(message=message)
):
if event.HasField('message'):
return event.message
# Build up the task from the streamed task and update events
if event.HasField('task'):
lastTask = event.task
elif lastTask and event.HasField('status_update'):
lastTask.status.CopyFrom(event.status_update.status)
elif lastTask and event.HasField('artifact_update'):
lastTask.artifacts.append(event.artifact_update.artifact)
if lastTask and self.is_terminal_or_interrupted(lastTask):
return lastTask
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.TASK_STATE_COMPLETED,
TaskState.TASK_STATE_CANCELED,
TaskState.TASK_STATE_FAILED,
TaskState.TASK_STATE_REJECTED,
TaskState.TASK_STATE_INPUT_REQUIRED,
TaskState.TASK_STATE_AUTH_REQUIRED,
]
Artık belirli bir istemci ve aracı kartı aracılığıyla istemci ile sunucu arasında bağlantı kurabiliyoruz. Bu, ana makine aracısının uzak aracı sunucularıyla bağlantı kurmak için kullanacağı bir araçtır.
Şimdi de ana makine aracısını oluşturmaya geçelim. İlk olarak aşağıdaki kod segmentini app/src/host/agent.py dosyasına kopyalayın. Bu, uzak aracı uç noktalarını ve standart bir httpx istemcisini gerektiren Pythonic sınıfının başlatılmasıdır. Bu aracı sınıfını başlatmak, kendisine sağlanan uzak adresler aracılığıyla bağlantıları oluşturur.
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,
)
from a2a.utils import TransportProtocol
from google.protobuf.json_format import MessageToDict
from google.adk import Agent
from google.adk.models import Gemini
from google.genai import types
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 = ''
Ana makine aracısının bir sonraki yönü, uzak aracılarla bağlantı kurmaktır. __init__ içinde bağlanmak yerine (bu, bir etkinlik döngüsü mevcut olmadan önce, modül içe aktarma sırasında çalışır) ensure_initialized yöntemi bu çalışmayı, temsilci gerçekten kullanılana kadar erteler. Bağlantıların kurulup kurulmadığını kontrol eder. Kurulmamışsa A2A istemcisini oluşturur ve her uzak aracıya paralel olarak bağlanır. Bu segmenti app/src/host/agent.py içinde önceki segmentin hemen altına kopyalayın.
#####################################
# --- Remote Agent Initialization ---
#####################################
async def ensure_initialized(self):
if not self.remote_agent_connections:
config = ClientConfig(
httpx_client=self.http_client,
supported_protocol_bindings=[
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()
# Use the actual address, not the hardcoded URL in the card JSON
for interface in card.supported_interfaces:
interface.url = address
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)
Sırada LLM Agent uygulaması var. Bu örnekte, düzenleyicimiz için bir ADK aracısı kullanıyoruz. Bu aracının, istem talimatları, geri aramalar ve araçlar gibi tipik eklentilere ihtiyacı var. Bu bileşenlerin tümünü app/src/host/agent.py dosyasındaki ana makine aracısı sınıfının içine kopyalayın. Aşağıdaki geri çağırmayı ana makine aracısı sınıfına yerleştirin. Ajan düzgün şekilde tetiklenmeden önce bile bu geri çağırma, ajanın durumunda henüz başlatılmamışsa ajanı başlatır.
############################
# -- 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
Ana makine aracımızın bir sonraki bileşeni istem talimatıdır. Bu, düzenleyici aracımız olduğundan, koordinatöre oturumda şu anda neler olduğu hakkında bir fikir vermek için kullanabileceği uzak aracıların yanı sıra etkin olan mevcut aracı da bilmesi gerekir. Geri çağırmanın altına aşağıdaki istemi ve yardımcı işlevi ekleyin.
# --- 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'}
Şimdi de ana makine aracısının en önemli kısmı olan araçlara geçelim. Ana makine, send_message() aracına ve list_remote_agents() aracına erişebilir. list_remote_agent() aracı daha basittir. Yalnızca sınıfın aracı kartlarını okur ve her aracının adını ve açıklamasını içeren bir sözlük listesi döndürür. send_message() biraz daha gelişmiştir. Ajan adına ve ileti dizesine göre uzak bir ajana ileti veya görev, akış ya da akış olmayan içerik gönderebilir. Bu kod segmentini, app/src/host/agent.py içindeki istem bölümünün altında aynı Host Agent Python sınıfının içine yerleştirin.
# --- 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.ROLE_USER,
parts=[Part(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
# Only reuse the task ID to continue a task that is waiting for input.
# Finished tasks can't take new messages, so the next call starts a new task.
if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
state['task_id'] = task.id
else:
state['task_id'] = None
if task.status.state == TaskState.TASK_STATE_INPUT_REQUIRED:
# Force user input back
tool_context.actions.skip_summarization = True
tool_context.actions.escalate = True
elif task.status.state == TaskState.TASK_STATE_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.TASK_STATE_FAILED:
# Raise error for failure
raise ValueError(f'Agent {agent_name} task {task.id} failed')
response = []
if task.status.HasField('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
Son olarak, tüm bu bileşenleri içeren ADK aracısını uygulayacağız. Bu, BeforeModelCallback çalıştırıldıktan sonra kendisine verilen istemi elindeki araçlarla gerçekleştiren düzenleyici sınıfı aracısının beynidir.
Aşağıdaki kod segmentini, app/src/host/agent.py dosyasındaki araçlar bölümünün hemen altına yerleştirin:
def create_agent(self) -> Agent:
"""Create an instance of the CoordinatorAgent."""
return Agent(
model=Gemini(
model='gemini-3.8-flash',
retry_options=types.HttpRetryOptions(attempts=5),
),
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,
],
)
Temsilcinin, A2A bölümlerinin ham metne ve verilere dönüştürülmesi için send_message() aracında birkaç yardımcı işlevi kullanması gerekir. Bu segmenti CoordinatorAgent sınıfının DIŞINDA kopyalayın:
##########################
# --- Helper Functions ---
##########################
async def convert_part(part: Part, tool_context: ToolContext):
"""Convert a part to text. Only text parts are supported."""
if part.HasField("text"):
return part.text
if part.HasField("data"):
return MessageToDict(part.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
adk web veya adk run üzerinden yerel test ve etkileşimi kolaylaştırmak için app/src/host/agent.py dosyasının en altına root_agent başlatma işlemi ekleyin:
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()
TEBRİKLER! İlk A2A ana makine aracınızı oluşturdunuz. root_agent değişkeniyle başlattığımız için diğer uzak temsilciler gibi adk run veya yerel adk web dağıtımı üzerinden etkileşim kurabiliriz. Düzenleyeninizle etkileşim kurmak için aşağıdaki komutları kullanın:
cd $HOME/app/src/
uv run adk run host
Düzenleyen, yalnızca üç uzak aracı sunucusu çalıştıktan sonra yanıt verebilir. Bu işlemi bir sonraki adımda yapacaksınız. Bu süre zarfında düzenleyen, selamlama mesajlarına yanıt verebilir ancak uzak bir temsilci gerektiren tüm istekler AgentCardResolutionError ile başarısız olur. Ana makineyi durdurmak için exit yazın.
10. Yerel olarak test etme
Çoklu temsilci sisteminin tamamını test etmek için ana temsilcinizin kullanmasını istediğiniz uzak temsilci sunucularını başlatmanız gerekir. Bu komut dosyası, tüm aracı sunucularını aynı anda başlatır. Komut dosyası oluşturma:
touch $HOME/app/start_agents.sh
cloudshell edit $HOME/app/start_agents.sh
Aşağıdaki komut dosyasını $HOME/app/start_agents.sh içine kopyalayın:
#!/bin/bash
# Exit immediately if a command exits with a non-zero status.
set -e
# Function to stop all background processes
cleanup() {
echo "Stopping agent servers..."
kill $(jobs -p) 2>/dev/null
wait
echo "All processes terminated."
}
# Call the cleanup function when the script exits or is interrupted
trap cleanup EXIT
# Start the agent servers in the background from the app directory
cd $HOME/app
echo "Starting agent servers..."
uv run python3 -m src.agents.retriever.server &
uv run python3 -m src.agents.evaluator.server &
uv run python3 -m src.agents.emailer.server &
# Start the host agent (it reads the remote agent URLs from .env)
cd $HOME/app/src/
uv run adk run host
# Wait for all background processes to finish
wait
Ardından komut dosyasını çalıştırın:
bash $HOME/app/start_agents.sh
Bu komut dosyası dört aracının (getirici, değerlendirici, e-posta gönderici ve ana makine) tümünü başlatır. Terminalinizde her sunucunun günlük çıktısını görürsünüz. Sunucular çalıştıktan sonra yeni bir terminal penceresi açıp adk web arayüzünü başlatabilirsiniz:
# In a new terminal
cd $HOME/app/src/
uv run adk web
Terminalinizde görünen localhost bağlantısını açın. Buradan, sol üstteki açılır menüden seçerek doğrudan ana makine aracısıyla etkileşim kurabilirsiniz.
11. Temizleme
Devam eden ücretleri önlemek için bu codelab sırasında oluşturulan kaynakları silin.
Çalışan tüm aracı sunucularını durdurun, ardından özellikle bu codelab için oluşturduysanız projeyi silin. Uzak aracıları başlatan terminallere gidin, Ctrl+C komutunu çalıştırın ve ardından şu Google Cloud projesini kaldırın:
gcloud projects delete <YOUR_PROJECT_ID>
12. Tebrikler
Agent2Agent ile çoklu temsilci sistemini başarıyla oluşturdunuz.
Öğrendikleriniz
- Kendi araçlarına sahip bağımsız ADK aracıları oluşturma
- Ajan kartlarıyla ajanlara bulunabilir kimlikler verme
- A2A sunucuları aracılığıyla aracıları kullanıma sunma
- Uzak aracıları düzenleyen bir ana makine aracısı oluşturma
- A2A protokolü, bağımsız olarak geliştirilen temsilciler arasında iletişimi nasıl sağlar?
Sonraki adımlar
- Sistemi üretimde kullanmak üzere Cloud Run'a dağıtma
- Sistemin özelliklerini genişletmek için daha fazla uzaktan ajan ekleme
- A2A'daki akış ve push bildirimi özelliklerini keşfetme