Examples
This page provides ready-to-use A2A integration examples.
Prerequisites
The examples assume:
| Dependency | Version | Purpose |
|---|---|---|
| Python | 3.12 | Matches the project runtime |
a2a-sdk | >=1.0.2,<2 | A2A client and protobuf types |
httpx | >=0.27.0 | HTTP client used by the SDK and direct examples |
iac-code | current repo | Provides the iac-code a2a subcommand |
Start the server:
uv sync --extra a2a
iac-code a2a --host 127.0.0.1 --port 41242
Python SDK — Streaming Session
This example discovers the Agent Card, sends a message, prints assistant text chunks, and reports tool metadata.
"""Streaming iac-code A2A session using a2a-sdk."""
import asyncio
import uuid
from pathlib import Path
from typing import Any
import httpx
from a2a.client import ClientConfig, ClientFactory
from a2a.types import Message, Part, Role, SendMessageRequest
from google.protobuf.json_format import MessageToDict
def metadata_to_dict(value: Any) -> dict[str, Any]:
if value is None:
return {}
return MessageToDict(value, preserving_proto_field_name=False)
async def main() -> None:
headers = {}
# For authenticated servers:
# headers["Authorization"] = "Bearer YOUR_TOKEN"
async with httpx.AsyncClient(headers=headers, timeout=120.0) as httpx_client:
config = ClientConfig(httpx_client=httpx_client, streaming=True)
client = await ClientFactory(config).create_from_url("http://127.0.0.1:41242")
request = SendMessageRequest(
message=Message(
message_id=f"msg-{uuid.uuid4().hex}",
role=Role.ROLE_USER,
parts=[Part(text="Create a ROS VPC template with two vSwitches.")],
metadata={"iac_code": {"cwd": str(Path.cwd())}},
)
)
async for event in client.send_message(request):
if event.HasField("task"):
print(f"[task] {event.task.id} context={event.task.context_id}")
elif event.HasField("status_update"):
update = event.status_update
status = update.status
if status.message:
for part in status.message.parts:
if part.text:
print(part.text, end="", flush=True)
metadata = metadata_to_dict(update.metadata)
tool = metadata.get("iac_code", {}).get("tool")
if tool:
print(f"\n[tool] {tool.get('status')} {tool.get('name', '')}".rstrip())
usage = metadata.get("iac_code", {}).get("usage")
if usage:
print(f"\n[usage] {usage['totalTokens']} total tokens")
if status.state == "TASK_STATE_INPUT_REQUIRED":
print("\n[done]")
await client.close()
if __name__ == "__main__":
asyncio.run(main())
CLI — End-to-End Workflow
Start a local server with persistence, artifacts, push notification support, and a signed Agent Card:
host: 127.0.0.1
port: 41242
persistence-dir: ~/.iac-code/a2a
artifact-dir: ~/.iac-code/a2a/artifacts
signing-secret: local-card-signing-secret
push-notifications: true
iac-code a2a --config a2a-server.yml
Create a client config for the stable endpoint and card verification settings:
url: http://127.0.0.1:41242/
verify-card-secret: local-card-signing-secret
require-card-signature: true
Discover and verify the Agent Card:
iac-code a2a-client --config a2a-client.yml discover
Send a streaming request:
iac-code a2a-client --config a2a-client.yml call \
--prompt "Create a ROS VPC template with two vSwitches." \
--cwd "$PWD" \
--stream
List tasks and fetch one task as JSON:
iac-code a2a-client --config a2a-client.yml task-list --output table
iac-code a2a-client --config a2a-client.yml task-get \
--task-id task-123 \
--history-length 20
Register a push callback for a task:
iac-code a2a-client --config a2a-client.yml push-config-create \
--task-id task-123 \
--config-id webhook-1 \
--callback-url https://hooks.example.com/a2a \
--notification-token "$NOTIFICATION_TOKEN" \
--auth-scheme bearer \
--auth-credentials "$WEBHOOK_BEARER_TOKEN"
Preview route selection before calling a routed agent:
iac-code a2a-route-preview \
--route "ros=http://127.0.0.1:41242/;skills=iac_generation;tags=ros,template" \
--skill iac_generation \
--prompt "Create a ROS VPC template"
Python SDK — Follow-up Message
Follow-up messages reuse the same context_id and usually the same task ID. This keeps the internal iac-code runtime and conversation history alive.
import uuid
from pathlib import Path
from a2a.types import Message, Part, Role, SendMessageRequest
def build_follow_up(task_id: str, context_id: str, text: str) -> SendMessageRequest:
return SendMessageRequest(
message=Message(
message_id=f"msg-{uuid.uuid4().hex}",
task_id=task_id,
context_id=context_id,
role=Role.ROLE_USER,
parts=[Part(text=text)],
metadata={"iac_code": {"cwd": str(Path.cwd())}},
)
)
# Usage inside an async function:
# async for event in client.send_message(
# build_follow_up(task_id, context_id, "Add outputs for VPC and VSwitch IDs.")
# ):
# ...
The server rejects a reused contextId if the new message points at a different workspace.
Python SDK — Cancel a Task
from a2a.types import CancelTaskRequest
async def cancel_running_task(client, task_id: str) -> None:
task = await client.cancel_task(CancelTaskRequest(id=task_id))
print(f"{task.id}: {task.status.state}")
Direct HTTP — Minimal JSON-RPC Client
Use this when you do not want the SDK dependency in the caller.
"""Direct A2A JSON-RPC client using httpx."""
import asyncio
from pathlib import Path
import httpx
BASE_URL = "http://127.0.0.1:41242"
async def main() -> None:
async with httpx.AsyncClient(base_url=BASE_URL, timeout=120.0) as client:
response = await client.post(
"/",
headers={"A2A-Version": "1.0"},
json={
"jsonrpc": "2.0",
"id": "send-1",
"method": "SendMessage",
"params": {
"message": {
"messageId": "msg-1",
"role": "ROLE_USER",
"parts": [{"text": "Review this ROS template for missing parameters."}],
"metadata": {"iac_code": {"cwd": str(Path.cwd())}},
},
"configuration": {"acceptedOutputModes": ["text/plain"]},
},
},
)
response.raise_for_status()
print(response.json())
if __name__ == "__main__":
asyncio.run(main())
Direct HTTP — Streaming SSE
"""Read SendStreamingMessage events directly with httpx."""
import asyncio
from pathlib import Path
import httpx
BASE_URL = "http://127.0.0.1:41242"
async def main() -> None:
async with httpx.AsyncClient(base_url=BASE_URL, timeout=None) as client:
async with client.stream(
"POST",
"/",
headers={"A2A-Version": "1.0"},
json={
"jsonrpc": "2.0",
"id": "stream-1",
"method": "SendStreamingMessage",
"params": {
"message": {
"messageId": "msg-1",
"role": "ROLE_USER",
"parts": [{"text": "Generate a Terraform VPC example."}],
"metadata": {"iac_code": {"cwd": str(Path.cwd())}},
},
"configuration": {"acceptedOutputModes": ["text/plain"]},
},
},
) as response:
response.raise_for_status()
async for line in response.aiter_lines():
if line.startswith("data:"):
print(line.removeprefix("data:").strip())
if __name__ == "__main__":
asyncio.run(main())
Direct HTTP — Push Notification Config
The push config methods are available when the server runs with push-notifications: true.
"""Create an A2A task push notification config."""
import asyncio
import httpx
BASE_URL = "http://127.0.0.1:41242"
async def main() -> None:
async with httpx.AsyncClient(base_url=BASE_URL, timeout=30.0) as client:
response = await client.post(
"/",
headers={"A2A-Version": "1.0"},
json={
"jsonrpc": "2.0",
"id": "push-1",
"method": "CreateTaskPushNotificationConfig",
"params": {
"taskId": "task-123",
"id": "webhook-1",
"url": "https://hooks.example.com/a2a",
"token": "notification-token",
"authentication": {
"scheme": "bearer",
"credentials": "callback-token",
},
},
},
)
response.raise_for_status()
print(response.json())
if __name__ == "__main__":
asyncio.run(main())
Handling iac-code Metadata
Tool and usage events arrive in TaskStatusUpdateEvent.metadata.iac_code.
from typing import Any
def handle_iac_metadata(metadata: dict[str, Any]) -> None:
iac = metadata.get("iac_code", {})
if tool := iac.get("tool"):
status = tool.get("status")
tool_id = tool.get("toolUseId")
name = tool.get("name", "<unknown>")
print(f"tool={name} id={tool_id} status={status}")
if permission := iac.get("permission"):
if permission.get("autoApproved") is False:
print(f"permission rejected for {permission.get('toolName')}")
if usage := iac.get("usage"):
print(
"tokens="
f"{usage.get('inputTokens', 0)}+{usage.get('outputTokens', 0)}"
f"={usage.get('totalTokens', 0)}"
)
Common Pitfalls
| Symptom | Fix |
|---|---|
HTTP 401 | Include a configured auth scheme, such as Authorization: Bearer <token>, Basic auth, or X-API-Key: <key>, on both Agent Card and JSON-RPC requests |
Invalid A2A workspace metadata. | Use an existing absolute path in metadata.iac_code.cwd |
A2A server currently accepts text input only. | Send at least one non-empty text part |
Task is already working. | Wait for the current turn to finish before sending another message in the same context |
| Follow-up rejected as different workspace | Keep metadata.iac_code.cwd unchanged for a reused contextId |
| Local file URL rejected | Keep file:// parts inside metadata.iac_code.cwd and inside IACCODE_A2A_ALLOWED_CWDS |
| Push callback rejected | Use an HTTP(S) callback URL that is not localhost or a literal private/local IP address |
| Redis push queue fails to start | Install the a2a-redis extra and provide push-redis-url in the A2A config |