A2A Guide

This guide explains how to use isa_agent_sdk for Agent-to-Agent (A2A) communication, based on a verified local test between two isA_Agent instances.

What Was Verified

• auth_service running on http://localhost:8201
• Agent instance A on http://localhost:8101
• Agent instance B on http://localhost:8102
• A2A server endpoint on each instance:
• GET /.well-known/agent-card.json
• POST /a2a
• End-to-end client call from 8101 to 8102 succeeded via:
• POST /api/v1/a2a/test-client

SDK Components

Use these exports from isa_agent_sdk:

• A2AAgentCard
• A2AClient
• A2AServerAdapter
• register_a2a_fastapi_routes
• build_auth_service_token_validator

Server Integration (FastAPI)

from fastapi import FastAPI
from isa_agent_sdk import (
    A2AAgentCard,
    A2AServerAdapter,
    register_a2a_fastapi_routes,
    build_auth_service_token_validator,
)
 
app = FastAPI()
adapter = A2AServerAdapter()
 
card = A2AAgentCard(
    name="isA Agent",
    url="http://localhost:8102/a2a",
    token_url="http://localhost:8201/oauth/token",
).to_dict()
 
register_a2a_fastapi_routes(
    app,
    adapter=adapter,
    agent_card=card,
    rpc_path="/a2a",
    card_path="/.well-known/agent-card.json",
    auth_validator=build_auth_service_token_validator(
        "http://localhost:8201",
        required_scopes=["a2a.invoke"],
    ),
)

Client Call Example

from isa_agent_sdk import A2AClient
 
client = A2AClient("http://localhost:8102", auth_token="<bearer_token>")
card = await client.get_agent_card()
resp = await client.send_message("http://localhost:8102/a2a", "Reply exactly: A2A_OK")

Local Test Command

curl -X POST http://localhost:8101/api/v1/a2a/test-client \
  -H "Content-Type: application/json" \
  -d '{
    "target_base_url":"http://localhost:8102",
    "target_rpc_url":"http://localhost:8102/a2a",
    "message":"Reply exactly: A2A_OK",
    "timeout":180
  }'

Expected shape:

{
  "status": "success",
  "agent_card_name": "isA Agent",
  "rpc_response": { "jsonrpc": "2.0", "result": { "kind": "message" } }
}

Heartbeat Streaming

Long-running A2A tasks risk client-side HTTP read timeouts when the server takes more than a few seconds to respond. The SDK solves this with heartbeat streaming — periodic keepalive events sent over SSE while the runner is processing.

How It Works

When a client sends a message/stream (or SendStreamingMessage) request, the server adapter uses stream_rpc_events() to manage the full lifecycle:

1. Submitted — an initial event is yielded immediately, confirming the task was received.
2. Working — a second event confirms processing has started.
3. Heartbeat keepalives — while the runner executes, a working event is emitted every heartbeat_interval seconds (default: 15 seconds). Each heartbeat includes elapsed time:

   {
     "jsonrpc": "2.0",
     "id": "rpc_abc123",
     "result": {
       "kind": "task",
       "id": "task_def456",
       "status": {
         "state": "working",
         "message": "Processing... (30s elapsed)",
         "timestamp": "2026-03-12T10:00:30+00:00"
       }
     }
   }

4. Completed (or Failed) — the final event carries the result artifacts or error.

This sequence keeps the HTTP connection alive regardless of how long the underlying task takes.

Server Configuration

The heartbeat interval is set when calling stream_rpc_events(). The default route registered by register_a2a_fastapi_routes uses the adapter’s default of 15 seconds. To customize it in a manual handler:

from fastapi import Request
from fastapi.responses import StreamingResponse
import json
 
@app.post("/a2a")
async def custom_a2a_handler(request: Request):
    body = await request.json()
    method = body.get("method")
 
    if method in {"message/stream", "SendStreamingMessage"}:
        async def event_gen():
            async for payload in adapter.stream_rpc_events(
                body, heartbeat_interval=20  # seconds
            ):
                yield f"data: {json.dumps(payload)}\n\n"
 
        return StreamingResponse(event_gen(), media_type="text/event-stream")
 
    result = await adapter.handle_rpc(body)
    return result

Client-Side Handling

When consuming an SSE stream, heartbeat events arrive as working status updates with no artifacts. Filter them out to collect only the final result:

from isa_agent_sdk import A2AClient
 
client = A2AClient("http://localhost:8102", auth_token="<token>")
 
async for event in client.stream_message(
    "http://localhost:8102/a2a",
    "Long-running analysis task",
):
    result = event.get("result", {})
    status = result.get("status", {})
    state = status.get("state", "") if isinstance(status, dict) else str(status)
 
    # Skip heartbeat/progress events (working state with no artifacts)
    if state in ("submitted", "working") and not result.get("artifacts"):
        continue
 
    if state == "completed":
        for artifact in result.get("artifacts", []):
            for part in artifact.get("parts", []):
                if part.get("kind") == "text":
                    print(part["text"])
 
    if state == "failed":
        error = result.get("error", {})
        print(f"Task failed: {error.get('message', 'unknown error')}")

Auto-Streaming via Delegation

The SDK’s delegate_to_team tool automatically switches to streaming delegation for teams whose timeout >= 180s. This means long-running inter-agent calls get heartbeat protection without any client code changes. The delegation client skips heartbeat events internally and returns only the final result.

Configure team timeout to enable auto-streaming:

teams:
  entries:
    - name: isa_vibe
      url: "http://127.0.0.1:18793"
      rpc_path: "/a2a"
      timeout: 600.0  # >= 180s triggers auto-streaming with heartbeats
    - name: isa_trade
      url: "http://127.0.0.1:18790"
      rpc_path: "/a2a"
      timeout: 120.0  # < 180s uses synchronous send_message

Connection Drop Prevention

Without heartbeats, three layers can terminate idle connections:

By emitting a heartbeat every 15 seconds, the SDK stays well within all common timeout windows. For environments with aggressive timeouts (e.g., 10s), reduce the interval accordingly:

adapter.stream_rpc_events(body, heartbeat_interval=5)

Delegation Result Status

DelegationResult.status supports 4 values:

When status="degraded", DelegationResult.degradation_reason explains why (e.g. rate_limited, partial_content, fallback_model).

Resubscribe

tasks/resubscribe with an existing taskId returns the stored task state without creating new work:

resp = await client.send_rpc("tasks/resubscribe", {"taskId": "task_abc123"})
# Returns existing completed/working task — no duplicate execution

Lightweight Delegation

For simple delegated tasks, ISAAgentOptions(lightweight_context=True) skips memory, file context, and semantic tool search, reducing context initialization from ~60s to ~2s.

Notes

• For quick local functional testing, A2A_AUTH_REQUIRED=false can be used.
• For real interoperability testing, enable auth and pass OAuth token from auth_service /oauth/token.