Expose your agents as callable services and invoke other agents via the A2A JSON-RPC protocol.

A2A Protocol — Agent-to-Agent Communication

AgentBreeder implements the Google A2A specification — a JSON-RPC 2.0 protocol for agents to discover and call each other. Every deployed agent can expose an A2A endpoint, be discovered by other agents, and call remote agents without knowing their internal implementation.

Core Concepts

Concept	Meaning
Agent Card	A JSON document at `/.well-known/agent.json` that describes the agent — name, capabilities, endpoint, and auth requirements
A2A endpoint	The JSON-RPC endpoint at `/a2a` that accepts `tasks/send` and `tasks/get` calls
Service token	A short-lived JWT issued by one agent to authenticate its calls to another
A2A client	`AgentInvocationClient` — the async HTTP client for calling remote agents
A2A server	`create_a2a_app()` — the FastAPI sub-app that serves the protocol

Exposing Your Agent via A2A

When you deploy an agent with AgentBreeder, the A2A endpoint is mounted automatically. To customise the Agent Card, set metadata in agent.yaml:

name: summarizer
version: 1.0.0
team: engineering
owner: alice@company.com
description: "Summarises long documents into bullet-point reports."
tags: [summarization, documents]

model:
  primary: claude-sonnet-4

framework: claude_sdk

deploy:
  cloud: gcp
  runtime: cloud-run

The Agent Card served at https://your-agent.run.app/.well-known/agent.json:

{
  "name": "summarizer",
  "version": "1.0.0",
  "description": "Summarises long documents into bullet-point reports.",
  "tags": ["summarization", "documents"],
  "endpoint": "https://your-agent.run.app/a2a",
  "auth": {
    "type": "bearer",
    "scheme": "a2a_service"
  },
  "capabilities": ["tasks/send", "tasks/get"]
}

Mounting the A2A server manually (Full Code tier)

from engine.a2a.server import create_a2a_app
from engine.a2a.protocol import AgentCardInfo

agent_card = AgentCardInfo(
    name="summarizer",
    version="1.0.0",
    description="Summarises long documents.",
    endpoint="https://summarizer.run.app/a2a",
    tags=["summarization"],
)

async def handle_task(message: str, context: dict) -> dict:
    result = await your_llm_call(message)
    return {"output": result, "tokens": 420}

a2a_app = create_a2a_app(agent_card=agent_card, invoke_handler=handle_task)

# Mount onto your main FastAPI app
from fastapi import FastAPI
app = FastAPI()
app.mount("/", a2a_app)

Calling Another Agent

Via `agent.yaml` (Low Code)

Reference another agent as a tool in agent.yaml:

tools:
  - name: summarizer
    type: agent
    ref: agent://summarizer@v1.0
    endpoint: https://summarizer.run.app/a2a

The AgentBreeder runtime issues a service token and calls the remote agent on behalf of your agent.

Via Python SDK (Full Code)

from engine.a2a.client import AgentInvocationClient
from engine.a2a.auth import create_service_token

# Create a service token for agent-to-agent auth
token = create_service_token(agent_name="triage-agent", team="engineering")

client = AgentInvocationClient(auth_token=token, timeout=30.0)

result = await client.invoke(
    endpoint_url="https://summarizer.run.app",
    input_message="Summarise this contract: ...",
    context={"session_id": "sess-abc123", "user": "alice"},
)

print(result.output)      # "Key points: ..."
print(result.tokens)      # 312
print(result.latency_ms)  # 1840

Via JSON-RPC directly

curl -X POST https://summarizer.run.app/a2a \
  -H "Authorization: Bearer $SERVICE_TOKEN" \
  -H "Content-Type: application/json" \
  -d '{
    "jsonrpc": "2.0",
    "id": "req-001",
    "method": "tasks/send",
    "params": {
      "message": "Summarise this contract: ...",
      "context": {"session_id": "sess-abc123"}
    }
  }'

Response:

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "task_id": "task-xyz789",
    "status": "completed",
    "output": "Key points:\n- Payment: net-30\n- Term: 24 months\n- Auto-renews unless cancelled 60 days prior",
    "tokens": 312,
    "latency_ms": 1840
  }
}

Authentication

All A2A calls use JWT service tokens. Tokens are scoped to agent identity and expire after 1 hour.

Issuing a service token

from engine.a2a.auth import create_service_token

token = create_service_token(
    agent_name="triage-agent",
    team="engineering",
    extra_claims={"env": "production"},
)

Validating an incoming token

from engine.a2a.auth import validate_service_token
from jwt.exceptions import InvalidTokenError

try:
    claims = validate_service_token(token)
    # claims → {"sub": "agent:triage-agent", "team": "engineering", ...}
except InvalidTokenError:
    # Reject the request
    ...

Token validation happens automatically inside create_a2a_app() — you don't need to add it manually.

Environment variable:

Variable	Default	Description
`A2A_JWT_SECRET`	`$SECRET_KEY`	Secret used to sign and verify service tokens

Agent Discovery

Look up an agent by name

agentbreeder describe agent summarizer

Name:        summarizer
Version:     1.0.0
Team:        engineering
Framework:   claude_sdk
Endpoint:    https://summarizer.run.app
A2A:         https://summarizer.run.app/.well-known/agent.json
Status:      running

Registry API

# List all A2A-enabled agents
curl https://your-agentbreeder/api/v1/a2a/agents \
  -H "Authorization: Bearer $TOKEN"

{
  "data": [
    {
      "name": "summarizer",
      "endpoint": "https://summarizer.run.app/a2a",
      "agent_card_url": "https://summarizer.run.app/.well-known/agent.json",
      "capabilities": ["tasks/send", "tasks/get"],
      "team": "engineering"
    }
  ]
}

Long-Running Tasks

For tasks that take more than a few seconds, poll via tasks/get:

# Send the task
result = await client.invoke(endpoint_url=url, input_message="Analyse this 200-page report...")

# If status is "in_progress", poll
if result.status == "in_progress":
    import asyncio
    task_id = result.task_id  # returned when status is in_progress
    for _ in range(30):       # poll up to 30 times
        await asyncio.sleep(5)
        poll = await client.poll(endpoint_url=url, task_id=task_id)
        if poll.status == "completed":
            break

Building a Multi-Agent Pipeline

A common pattern: a triage agent routes to specialised agents.

# triage-agent/agent.yaml
name: triage-agent
framework: langgraph
model:
  primary: claude-haiku-4-5
tools:
  - name: summarizer
    type: agent
    ref: agent://summarizer@v1.0
    endpoint: https://summarizer.run.app/a2a
  - name: sentiment-analyzer
    type: agent
    ref: agent://sentiment@v2.1
    endpoint: https://sentiment.run.app/a2a

Or via the orchestration SDK:

from sdk.python.agenthub.orchestration import Orchestration

pipeline = (
    Orchestration("support-pipeline", strategy="router", team="engineering")
    .add_agent("triage",    ref="agents/triage")
    .add_agent("billing",   ref="agents/billing")
    .add_agent("technical", ref="agents/technical")
    .with_route("triage", condition="billing",   target="billing")
    .with_route("triage", condition="technical", target="technical")
    .with_route("triage", condition="default",   target="billing")
)

See Orchestration SDK → for full multi-agent pipeline docs.

A2A vs Direct Invocation

	A2A	Direct `/invoke`
Discovery	Via Agent Card (`/.well-known/agent.json`)	Manual — you must know the URL
Auth	JWT service tokens, auto-issued	Bring your own
Protocol	JSON-RPC 2.0 (standard)	Custom per-agent
Async tasks	`tasks/get` polling built in	Not standardised
Use when	Calling agents you don't control	Internal service mesh

Use A2A when calling agents across team boundaries or from agents you didn't write. For agents within your own stack on the same deploy, direct HTTP is fine.

A2A Protocol — Agent-to-Agent Communication

On this page