I registered an A2A agent in the Agent Registry and tried calling it dynamically from Strands Agents
This page has been translated by machine translation. View original
Introduction
Hello, I'm Jinno from the consulting department who loves supermarkets.
In my previous article, I introduced how to register Agent Skills in Agent Registry and dynamically load them from Strands Agents.
This time, as a continuation, I'll cover the A2A version. We'll quickly create two Strands Agents A2A servers deployed to AgentCore Runtime using Terraform, register them with Agent Registry, and call them dynamically from an orchestrator agent via the Registry.
Here's a visual representation:
The pleasant point is that the orchestrator doesn't need to hold any URLs or tool definitions for A2A interactions, as it can dynamically reference other agents' information from the Registry. This means the code can be used as is even when the number of agents you want to support increases or decreases.
As the number of A2A agents increases throughout the organization, what the orchestrator can do automatically increases as well. Of course, there's a trade-off with the latency at execution time and the unpredictability of what gets called, so it depends on the use case. It seems like orchestrator tuning will become more challenging as capabilities increase...
Prerequisites
Here are the versions used in this article:
| Item | Version/Setting |
|---|---|
| Python | 3.12 |
| Package Management | uv |
| Terraform | >= 1.5.0 |
| AWS Provider | >= 6.25.0 |
| strands-agents[a2a] | >= 1.36.0 |
| strands-agents-tools | >= 0.5.0 |
| Region | us-east-1 |
| Model | us.anthropic.claude-haiku-4-5-20251001-v1:0 |
We'll authenticate to the AgentCore Runtime A2A server using IAM (SigV4).
We'll continue using the auto-approval Registry created in the previous article. Please keep note of your Registry ID. For details on creating a Registry, refer to my previous article.
The complete code used in this article is available in this repository. If you want to try it locally, please clone it and proceed:
git clone https://github.com/yuu551/a2a-agentcore-strands.git
cd a2a-agentcore-strands
uv sync
Overall Flow
Here's the implementation flow:
- Deploy A2A servers to ECR & AgentCore Runtime using Terraform (Calculator / Weather)
- Register A2A Agent Cards to Agent Registry
- Local orchestrator agent searches the Registry and connects to A2A servers
- Put the orchestrator on AgentCore Runtime for Runtime→Runtime calls
The file structure is as follows:
.
├── register_a2a.py # Register A2A Agent Cards to Registry
├── registry_a2a_loader.py # Build A2AClientToolProvider from Registry
├── a2a_client.py # Run orchestrator locally
├── invoke_orchestrator.py # Call Runtime orchestrator using SSE
├── sigv4_auth.py # SigV4 Auth for httpx
└── terraform/
├── versions.tf
├── variables.tf
├── iam.tf
├── runtime.tf
├── outputs.tf
└── agent/
├── calculator/ # A2A: Calculator agent
│ ├── server.py
│ ├── Dockerfile
│ └── requirements.txt
├── weather/ # A2A: Weather agent
│ ├── server.py
│ ├── Dockerfile
│ └── requirements.txt
└── orchestrator/ # HTTP: Orchestrator
├── server.py
├── registry_a2a_loader.py # Copied from root
├── sigv4_auth.py # Copied from root
├── Dockerfile
└── requirements.txt
A2A Server Implementation
Calculator Agent
We wrap the Agent with Strands Agents' A2AServer and expose it as a FastAPI app. For the calculation tool, we'll use strands_tools.calculator provided by Strands as is.
AgentCore Runtime injects the environment variable AGENTCORE_RUNTIME_URL for A2A, which we pass to A2AServer's http_url. This URL goes into the url field of the Agent Card, and the client sends JSON-RPC requests there.
from __future__ import annotations
import os
import uvicorn
from strands import Agent
from strands.models import BedrockModel
from strands.multiagent.a2a import A2AServer
from strands_tools.calculator import calculator
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
REGION = os.environ.get("AWS_REGION", "us-east-1")
RUNTIME_URL = os.environ.get("AGENTCORE_RUNTIME_URL", "http://127.0.0.1:9000/")
model = BedrockModel(model_id=MODEL_ID, region_name=REGION)
agent = Agent(
name="計算エージェント",
description="四則演算・代数・微積分・記号計算を行うエージェントです。",
model=model,
tools=[calculator],
callback_handler=None,
)
server = A2AServer(
agent=agent,
host="0.0.0.0",
port=9000,
http_url=RUNTIME_URL,
serve_at_root=True,
)
app = server.to_fastapi_app()
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=9000)
To run A2A on AgentCore Runtime, port 9000 and root path are required. The endpoints /ping and /.well-known/agent-card.json are automatically created by A2AServer, so you don't need to write them yourself.
Weather Agent
The second agent returns weather information. For this example, we'll create a tool with a mock function that returns fixed data.
from __future__ import annotations
import os
import uvicorn
from strands import Agent, tool
from strands.models import BedrockModel
from strands.multiagent.a2a import A2AServer
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
REGION = os.environ.get("AWS_REGION", "us-east-1")
RUNTIME_URL = os.environ.get("AGENTCORE_RUNTIME_URL", "http://127.0.0.1:9000/")
_MOCK_WEATHER = {
"tokyo": {"condition": "Sunny", "temperature_c": 22, "humidity": 55},
"osaka": {"condition": "Cloudy", "temperature_c": 20, "humidity": 65},
"sapporo": {"condition": "Snowy", "temperature_c": -2, "humidity": 80},
"fukuoka": {"condition": "Rainy", "temperature_c": 18, "humidity": 75},
"naha": {"condition": "Sunny", "temperature_c": 27, "humidity": 70},
}
@tool
def get_weather(city: str) -> dict:
"""日本の都市の現在の天気情報を返します。対応都市: Tokyo, Osaka, Sapporo, Fukuoka, Naha。"""
data = _MOCK_WEATHER.get(city.strip().lower())
if data is None:
return {"error": f"Unknown city: {city}. Supported: {list(_MOCK_WEATHER)}"}
return {"city": city, **data}
model = BedrockModel(model_id=MODEL_ID, region_name=REGION)
agent = Agent(
name="天気エージェント",
description="日本の主要都市の現在の天気情報を返すエージェントです。",
model=model,
tools=[get_weather],
callback_handler=None,
)
server = A2AServer(
agent=agent,
host="0.0.0.0",
port=9000,
http_url=RUNTIME_URL,
serve_at_root=True,
)
app = server.to_fastapi_app()
if __name__ == "__main__":
uvicorn.run(app, host="0.0.0.0", port=9000)
Common Dockerfile
Both agents use the same Dockerfile structure. Don't forget --platform=linux/arm64 as AgentCore Runtime requires ARM64.
FROM --platform=linux/arm64 public.ecr.aws/docker/library/python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
COPY server.py .
EXPOSE 9000
CMD ["python", "server.py"]
strands-agents[a2a]>=1.36.0
strands-agents-tools>=0.5.0
fastapi>=0.115.0
uvicorn>=0.32.0
Terraform Deployment
We'll create two ECRs with Docker build & push, and two agents running on AgentCore Runtime. We specify server_protocol = "A2A" in Runtime's protocol_configuration.
locals {
agents = {
calculator = { source_dir = "${path.module}/agent/calculator" }
weather = { source_dir = "${path.module}/agent/weather" }
}
}
resource "aws_ecr_repository" "calculator" {
name = "${var.project_name}-calculator-${random_string.suffix.result}"
image_tag_mutability = "MUTABLE"
force_delete = true
}
resource "aws_ecr_repository" "weather" {
name = "${var.project_name}-weather-${random_string.suffix.result}"
image_tag_mutability = "MUTABLE"
force_delete = true
}
resource "null_resource" "calculator_build_push" {
triggers = {
dockerfile = filemd5("${local.agents.calculator.source_dir}/Dockerfile")
server = filemd5("${local.agents.calculator.source_dir}/server.py")
requirements = filemd5("${local.agents.calculator.source_dir}/requirements.txt")
}
provisioner "local-exec" {
working_dir = local.agents.calculator.source_dir
command = <<-EOT
aws ecr get-login-password --region ${data.aws_region.current.region} | \
docker login --username AWS --password-stdin ${aws_ecr_repository.calculator.repository_url}
docker buildx build --platform linux/arm64 \
-t ${aws_ecr_repository.calculator.repository_url}:latest --push .
EOT
}
depends_on = [aws_ecr_repository.calculator]
}
# weather_build_push has the same structure, so it's omitted
resource "aws_bedrockagentcore_agent_runtime" "calculator" {
agent_runtime_name = replace("${var.project_name}_calculator_${random_string.suffix.result}", "-", "_")
description = "Calculator A2A server"
role_arn = aws_iam_role.agent_runtime.arn
agent_runtime_artifact {
container_configuration {
container_uri = "${aws_ecr_repository.calculator.repository_url}:latest"
}
}
network_configuration {
network_mode = "PUBLIC"
}
protocol_configuration {
server_protocol = "A2A"
}
depends_on = [
null_resource.calculator_build_push,
aws_iam_role_policy.runtime_ecr,
aws_iam_role_policy.runtime_logs,
aws_iam_role_policy.runtime_bedrock,
]
}
# aws_bedrockagentcore_agent_runtime.weather has the same structure, so it's omitted
The Runtime execution role is granted three types of permissions: ECR pull, CloudWatch Logs write, and Bedrock InvokeModel.
resource "aws_iam_role" "agent_runtime" {
name = "${var.project_name}-runtime-${random_string.suffix.result}"
assume_role_policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = "sts:AssumeRole"
Principal = { Service = "bedrock-agentcore.amazonaws.com" }
Condition = {
StringEquals = { "aws:SourceAccount" = data.aws_caller_identity.current.account_id }
}
}]
})
}
resource "aws_iam_role_policy" "runtime_bedrock" {
name = "bedrock-invoke"
role = aws_iam_role.agent_runtime.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = ["bedrock:InvokeModel", "bedrock:InvokeModelWithResponseStream"]
Resource = "*"
}]
})
}
# runtime_ecr / runtime_logs are omitted. See the GitHub repository for details
In outputs, we set the Runtime ARN for A2A:
output "calculator_runtime_arn" {
value = aws_bedrockagentcore_agent_runtime.calculator.agent_runtime_arn
}
output "weather_runtime_arn" {
value = aws_bedrockagentcore_agent_runtime.weather.agent_runtime_arn
}
The client URL is constructed in the form https://bedrock-agentcore.<region>.amazonaws.com/runtimes/<arn>/invocations (with ARN URL-encoded using urllib.parse.quote).
Let's deploy:
cd terraform
terraform init
terraform apply
After deployment, let's register these agents in the Registry.
Register A2A Agent Cards in Registry
A2A descriptor structure
Agent Registry provides A2A as a descriptorType. The A2A descriptor puts the Agent Card JSON in a2a.agentCard.inlineContent.
Before registration, we'll fetch the actual Agent Card from the Runtime and register its contents in the Registry. This registration allows clients to understand agent names, skills, connection URLs, and so on.
For details on A2A, please refer to my previous blog:
SigV4 Authentication Helper
Since Runtime uses IAM authentication for bedrock-agentcore, both Agent Card fetching and JSON-RPC calls require SigV4 signatures. We'll create a custom httpx Auth:
from __future__ import annotations
import boto3
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
class SigV4HttpxAuth(httpx.Auth):
requires_request_body = True
def __init__(self, region: str, service: str = "bedrock-agentcore") -> None:
credentials = boto3.Session().get_credentials()
if credentials is None:
raise RuntimeError("No AWS credentials found.")
self._signer = SigV4Auth(credentials, service, region)
def auth_flow(self, request: httpx.Request):
aws_request = AWSRequest(
method=request.method,
url=str(request.url),
data=request.content,
headers=dict(request.headers),
)
self._signer.add_auth(aws_request)
for key, value in aws_request.headers.items():
request.headers[key] = value
yield request
Note that requires_request_body = True is necessary; otherwise, the JSON-RPC body won't be included in the signature, resulting in a 4xx error on POST requests.
Registration Script
"""Register A2A agents (Calculator / Weather) to AWS Agent Registry."""
from __future__ import annotations
import json
import logging
import os
import time
from urllib.parse import quote
import boto3
import httpx
from sigv4_auth import SigV4HttpxAuth
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
REGISTRY_ID = os.environ["AGENT_REGISTRY_ID"]
REGION = os.environ.get("AWS_REGION", "us-east-1")
def runtime_base_url(arn: str) -> str:
return f"https://bedrock-agentcore.{REGION}.amazonaws.com/runtimes/{quote(arn, safe='')}/invocations"
def fetch_agent_card(base_url: str, auth: SigV4HttpxAuth) -> dict:
card_url = f"{base_url}/.well-known/agent-card.json"
response = httpx.get(card_url, auth=auth, timeout=30.0)
response.raise_for_status()
return response.json()
def wait_until_not(control, registry_id: str, record_id: str, transient: str, *, timeout: float = 30.0) -> str:
deadline = time.monotonic() + timeout
while True:
status = control.get_registry_record(registryId=registry_id, recordId=record_id)["status"]
if status != transient:
return status
if time.monotonic() > deadline:
raise TimeoutError(f"record {record_id} stuck in {transient}")
time.sleep(1.0)
def register(control, name: str, description: str, card: dict) -> str:
created = control.create_registry_record(
registryId=REGISTRY_ID,
name=name,
description=description,
recordVersion="1.0",
descriptorType="A2A",
descriptors={"a2a": {"agentCard": {"inlineContent": json.dumps(card)}}},
)
record_id = created["recordArn"].rsplit("/", 1)[-1]
logger.info("Created record %s (status=%s)", record_id, created["status"])
wait_until_not(control, REGISTRY_ID, record_id, "CREATING")
control.submit_registry_record_for_approval(registryId=REGISTRY_ID, recordId=record_id)
status = wait_until_not(control, REGISTRY_ID, record_id, "SUBMITTING")
logger.info("Record %s reached status %s", record_id, status)
return record_id
def main() -> None:
control = boto3.client("bedrock-agentcore-control", region_name=REGION)
auth = SigV4HttpxAuth(region=REGION)
targets = [
("calculator-a2a", "Calculator A2A agent (SymPy)", os.environ["CALCULATOR_RUNTIME_ARN"]),
("weather-a2a", "Weather A2A agent (mock for Japan cities)", os.environ["WEATHER_RUNTIME_ARN"]),
]
for name, description, arn in targets:
base = runtime_base_url(arn)
card = fetch_agent_card(base, auth)
logger.info("Fetched agent card for %s: %s", name, card["name"])
record_id = register(control, name, description, card)
print(f"{name}: {record_id}")
if __name__ == "__main__":
main()
We build the base URL from the Runtime ARN, fetch the Agent Card, and register it with the Registry. Since this Registry is set for auto-approval, it follows the create → submit → auto-approve flow to reach APPROVED status.
export AGENT_REGISTRY_ID="xxx"
export CALCULATOR_RUNTIME_ARN="$(cd terraform && terraform output -raw calculator_runtime_arn)"
export WEATHER_RUNTIME_ARN="$(cd terraform && terraform output -raw weather_runtime_arn)"
uv run python register_a2a.py
INFO: Fetched agent card for calculator-a2a: 計算エージェント
INFO: Created record ul4ncy9CGo0p (status=CREATING)
INFO: Record ul4ncy9CGo0p reached status APPROVED
INFO: Fetched agent card for weather-a2a: 天気エージェント
INFO: Created record vZspJcEwxhy5 (status=CREATING)
INFO: Record vZspJcEwxhy5 reached status APPROVED
calculator-a2a: ul4ncy9CGo0p
weather-a2a: vZspJcEwxhy5
Dynamically Loading A2A Agents from Registry
A2A Loader
This loader searches A2A records in the Registry, extracts the url field from Agent Cards, and builds an A2AClientToolProvider:
"""Search AWS Agent Registry for A2A records and build A2AClientToolProvider."""
from __future__ import annotations
import json
import logging
from dataclasses import dataclass
from typing import Any
import boto3
from strands_tools.a2a_client import A2AClientToolProvider
from sigv4_auth import SigV4HttpxAuth
logger = logging.getLogger(__name__)
@dataclass
class RegistryA2ALoaderConfig:
registry_id: str
region_name: str = "us-east-1"
max_results: int = 10
class RegistryA2ALoader:
def __init__(self, config: RegistryA2ALoaderConfig) -> None:
self._config = config
self._client = boto3.client("bedrock-agentcore", region_name=config.region_name)
def search(self, query: str) -> A2AClientToolProvider:
response = self._client.search_registry_records(
registryIds=[self._config.registry_id],
searchQuery=query,
maxResults=self._config.max_results,
filters={"descriptorType": {"$eq": "A2A"}},
)
records = response.get("registryRecords") or []
logger.info("Fetched %d A2A record(s) from registry", len(records))
urls: list[str] = []
for record in records:
url = self._extract_url(record)
if url:
urls.append(url)
logger.info("Loaded A2A: %s -> %s", record.get("name"), url[:80])
auth = SigV4HttpxAuth(region=self._config.region_name)
return A2AClientToolProvider(
known_agent_urls=urls,
httpx_client_args={"auth": auth},
)
@staticmethod
def _extract_url(record: dict[str, Any]) -> str | None:
descriptors = record.get("descriptors") or {}
a2a_desc = descriptors.get("a2a") or {}
card_json = (a2a_desc.get("agentCard") or {}).get("inlineContent")
if not card_json:
return None
card = json.loads(card_json)
return card.get("url")
It follows a simple flow: Registry search → extract url from Agent Card → pass to A2AClientToolProvider. Authentication is set using httpx_client_args.
Orchestrator
Let's connect A2A tools obtained from the Registry to a local Strands Agent and execute a complex task.
We'll try a request like: "Please tell me the weather in Tokyo. After that, convert the current temperature (Celsius) to Fahrenheit (Fahrenheit = Celsius × 9 / 5 + 32)."
"""Orchestrator Strands Agent that loads A2A tools from Agent Registry."""
from __future__ import annotations
import logging
import os
from strands import Agent
from strands.models import BedrockModel
from registry_a2a_loader import RegistryA2ALoader, RegistryA2ALoaderConfig
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
def main() -> None:
region = os.environ.get("AWS_REGION", "us-east-1")
config = RegistryA2ALoaderConfig(
registry_id=os.environ["AGENT_REGISTRY_ID"],
region_name=region,
)
loader = RegistryA2ALoader(config)
provider = loader.search("calculator weather")
model = BedrockModel(model_id=MODEL_ID, region_name=region)
agent = Agent(
model=model,
tools=provider.tools,
system_prompt=(
"You are an orchestrator that delegates tasks to remote A2A agents. "
"Use the A2A tools to send requests to the appropriate agent discovered from the registry."
),
)
user_input = (
"東京の天気を教えてください。その後、現在の気温（摂氏）を華氏に変換してください（華氏 = 摂氏 × 9 / 5 + 32）。"
)
logger.info("Query: %s", user_input)
result = agent(user_input)
print("\n--- result ---")
print(result.message)
if __name__ == "__main__":
main()
Functionality Check
Now that we've implemented it, let's run it. I'll specify the ID for a Registry I created beforehand.
AGENT_REGISTRY_ID="xxx" uv run python a2a_client.py
registry_a2a_loader INFO: Fetched 2 A2A record(s) from registry
registry_a2a_loader INFO: Loaded A2A: calculator-a2a -> https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/...
registry_a2a_loader INFO: Loaded A2A: weather-a2a -> https://bedrock-agentcore.us-east-1.amazonaws.com/runtimes/...
strands_tools.a2a_client INFO: Successfully discovered and cached agent card for (Weather)
strands_tools.a2a_client INFO: Sending message to (Weather)
strands_tools.a2a_client INFO: Successfully discovered and cached agent card for (Calculator)
strands_tools.a2a_client INFO: Sending message to (Calculator)
## Tokyo Weather Information
- Weather: Sunny
- Temperature: 22°C
- Humidity: 55%
## Temperature Conversion
Converting 22°C to Fahrenheit:
Formula: Fahrenheit = Celsius × 9 / 5 + 32
22 × 9 ÷ 5 + 32 = 71.6°F
Therefore, Tokyo's current temperature of 22°C equals 71.6°F!
The orchestrator successfully used A2A tools fetched from the Registry, calling the weather agent and then the calculator agent in sequence to complete the composite task!
It's great that we didn't have to write any individual tool definitions in the agent code - it can connect using only the endpoint information dynamically retrieved from the Registry.
Deploying the Orchestrator on AgentCore Runtime
So far we've been running the orchestrator locally, but the same logic can be deployed as a server on the HTTP protocol AgentCore Runtime. We just need to wrap it with BedrockAgentCoreApp from the bedrock-agentcore SDK.
from __future__ import annotations
import logging
import os
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from strands import Agent
from strands.models import BedrockModel
from registry_a2a_loader import RegistryA2ALoader, RegistryA2ALoaderConfig
logging.basicConfig(level=logging.INFO, format="%(asctime)s %(name)s %(levelname)s: %(message)s")
logger = logging.getLogger(__name__)
MODEL_ID = "us.anthropic.claude-haiku-4-5-20251001-v1:0"
REGION = os.environ.get("AWS_REGION", "us-east-1")
REGISTRY_ID = os.environ["AGENT_REGISTRY_ID"]
app = BedrockAgentCoreApp()
@app.entrypoint
async def handler(payload: dict, context):
query = payload.get("prompt")
if not query:
yield {"error": "`prompt` is required"}
return
search_query = payload.get("search_query", query)
config = RegistryA2ALoaderConfig(registry_id=REGISTRY_ID, region_name=REGION)
loader = RegistryA2ALoader(config)
provider = loader.search(search_query)
model = BedrockModel(model_id=MODEL_ID, region_name=REGION)
agent = Agent(
model=model,
tools=provider.tools,
system_prompt=(
"You are an orchestrator that delegates tasks to remote A2A agents."
),
)
async for event in agent.stream_async(query):
yield event
if __name__ == "__main__":
app.run()
We set up BedrockAgentCoreApp and implement a simple agent that connects via Registry with A2A.
Here's the Dockerfile implementation:
FROM --platform=linux/arm64 public.ecr.aws/docker/library/python:3.12-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
RUN pip install --no-cache-dir aws-opentelemetry-distro==0.12.2
COPY server.py .
COPY registry_a2a_loader.py .
COPY sigv4_auth.py .
EXPOSE 8080
CMD ["opentelemetry-instrument", "python", "server.py"]
On the Terraform side, we'll add a new Runtime with server_protocol = "HTTP". The Registry ID is set as an environment variable.
resource "aws_bedrockagentcore_agent_runtime" "orchestrator" {
agent_runtime_name = replace("${var.project_name}_orchestrator_${random_string.suffix.result}", "-", "_")
description = "Orchestrator HTTP server calling A2A agents via Registry"
role_arn = aws_iam_role.orchestrator_runtime.arn
agent_runtime_artifact {
container_configuration {
container_uri = "${aws_ecr_repository.orchestrator.repository_url}:latest"
}
}
network_configuration {
network_mode = "PUBLIC"
}
protocol_configuration {
server_protocol = "HTTP"
}
environment_variables = {
AGENT_REGISTRY_ID = var.agent_registry_id
}
}
GetAgentCard Permission Is Also Required
When connecting to other AgentCore Runtimes via A2A, bedrock-agentcore:InvokeAgentRuntime alone is not sufficient. The A2A specification first calls GET /.well-known/agent-card.json to get agent information, but this API is assigned a different action: bedrock-agentcore:GetAgentCard.
I got stuck in a pattern where Registry search worked but Agent Card retrieval resulted in 403 Forbidden, so I'm sharing this information for reference. It's something to be careful about.
resource "aws_iam_role_policy" "orchestrator_invoke_runtime" {
name = "invoke-a2a-runtime"
role = aws_iam_role.orchestrator_runtime.id
policy = jsonencode({
Version = "2012-10-17"
Statement = [{
Effect = "Allow"
Action = [
"bedrock-agentcore:InvokeAgentRuntime",
"bedrock-agentcore:InvokeAgentRuntimeForUser",
"bedrock-agentcore:GetAgentCard", # ← Required for A2A Agent Card retrieval
]
Resource = [
aws_bedrockagentcore_agent_runtime.calculator.agent_runtime_arn,
"${aws_bedrockagentcore_agent_runtime.calculator.agent_runtime_arn}/runtime-endpoint/*",
aws_bedrockagentcore_agent_runtime.weather.agent_runtime_arn,
"${aws_bedrockagentcore_agent_runtime.weather.agent_runtime_arn}/runtime-endpoint/*",
]
}]
})
}
Calling the Orchestrator (via Runtime)
Let's call the orchestrator deployed on AgentCore.
We'll prepare a simple client invoke_orchestrator.py to format the response:
"""Invoke the Orchestrator Runtime and pretty-print the SSE stream."""
from __future__ import annotations
import json
import os
import sys
import boto3
REGION = os.environ.get("AWS_REGION", "us-east-1")
def stream_events(body: bytes):
for chunk in body.decode("utf-8", errors="replace").split("\n\n"):
chunk = chunk.strip()
if not chunk.startswith("data:"):
continue
try:
yield json.loads(chunk[5:].strip())
except json.JSONDecodeError:
continue
def pretty_print(body: bytes) -> None:
tool_use: dict[str, dict] = {}
for event in stream_events(body):
if not isinstance(event, dict):
continue
ev = event.get("event") or {}
if "contentBlockStart" in ev:
tu = (ev["contentBlockStart"].get("start") or {}).get("toolUse")
if tu:
tool_use[tu["toolUseId"]] = {"name": tu.get("name"), "input": ""}
print(f"\n[tool_call] {tu.get('name')}", flush=True)
if "contentBlockDelta" in ev:
delta = ev["contentBlockDelta"].get("delta") or {}
if "text" in delta:
sys.stdout.write(delta["text"])
sys.stdout.flush()
if "toolUse" in delta:
frag = delta["toolUse"].get("input", "")
for tu in tool_use.values():
tu["input"] += frag
break
if "contentBlockStop" in ev:
for tu in list(tool_use.values()):
if tu["input"]:
try:
args = json.loads(tu["input"])
print(f" args: {json.dumps(args, ensure_ascii=False)[:200]}", flush=True)
except json.JSONDecodeError:
pass
tu["input"] = ""
if "messageStop" in ev:
print(flush=True)
def main() -> None:
arn = os.environ["ORCHESTRATOR_RUNTIME_ARN"]
prompt = " ".join(sys.argv[1:]) or "東京の天気を教えて、気温を華氏に変換して。"
search_query = os.environ.get("SEARCH_QUERY", "calculator weather")
client = boto3.client("bedrock-agentcore", region_name=REGION)
response = client.invoke_agent_runtime(
agentRuntimeArn=arn,
qualifier="DEFAULT",
payload=json.dumps({"prompt": prompt, "search_query": search_query}).encode(),
contentType="application/json",
)
pretty_print(response["response"].read())
if __name__ == "__main__":
main()
Let's check the operation:
export ORCHESTRATOR_RUNTIME_ARN="$(cd terraform && terraform output -raw orchestrator_runtime_arn)"
uv run python invoke_orchestrator.py "東京の天気を教えて、気温を華氏に変換して。"
I'll help you with Tokyo's weather information and temperature conversion to Fahrenheit. First, let me check the available agents.
[tool_call] a2a_list_discovered_agents
Perfect! Weather and calculator agents are available. I'll get Tokyo's weather information and then convert the temperature to Fahrenheit.
[tool_call] a2a_send_message
args: {"message_text": "東京の現在の天気情報を教えてください。", "target_agent_url": "https://.../a2a_demo_weather_.../invocations/"}
Great! I've got Tokyo's weather information. Now I'll convert the temperature of 22°C to Fahrenheit.
[tool_call] a2a_send_message
args: {"message_text": "22度セルシウスを華氏に変換してください。計算式は (22 * 9/5) + 32 です。", "target_agent_url": "https://.../a2a_demo_calculator_.../invocations/"}
Perfect! I've completed the Tokyo weather information and temperature conversion to Fahrenheit.
## 📍 Tokyo Weather Information
**Current Weather:**
- **Weather:** Sunny
- **Temperature:** 22°C → **71.6°F**
- **Humidity:** 55%
The weather continues to be nice with comfortable temperatures. 22 degrees Celsius equals 71.6 degrees Fahrenheit, which is a comfortable room temperature.
We can confirm the flow: retrieve agent list from Registry via a2a_list_discovered_agents → message to weather agent → message to calculator agent!
Looking at the CloudWatch Logs on the Runtime side, we can also confirm the internal flow of Registry search → Agent Card retrieval → JSON-RPC sending:
Fetched 2 A2A record(s) from registry
Loaded A2A: calculator-a2a -> https://bedrock-agentcore.us-east-1.amazonaws.com/...
Loaded A2A: weather-a2a -> https://bedrock-agentcore.us-east-1.amazonaws.com/...
Successfully discovered and cached agent card for ...weather...
Successfully discovered and cached agent card for ...calculator...
Sending message to ...weather...
Sending message to ...calculator...
Conclusion
Following the previous Skills version, this time we tried dynamic loading with A2A.
It's interesting how an orchestrator agent can connect to necessary components dynamically through Registry lookup, without directly holding information about its A2A counterparts. This seems to make management easier when the number of connections increases. It appears to be a useful foundation for centrally managing organization-approved agents. (It's unfortunate that Registry doesn't support HTTPS yet, so hopefully that will be supported soon...)
I hope this article has been helpful. Thank you for reading to the end!