Ecosystem & Marketplaces›References›A2A protocol

A2A protocol

The Agent-to-Agent (A2A) Protocol is an open standard for secure, interoperable communication across distributed AI agents. TalkOps uses A2A to let top-level supervisors, domain orchestrators, and external enterprise agents discover capabilities, negotiate tasks, and synchronize state across organizational trust boundaries.

Agent Cards Discovery

Advertise structured agent capabilities, input/output schemas, and supported authentication methods via /.well-known/agent.json.

JSON-RPC 2.0 Transport

Exchange deterministic request, response, and error payloads over HTTPS with TLS 1.3 and mutual authentication.

Stateful Task Lifecycles

Track long-running infrastructure operations across pending, running, awaiting_approval, and completed states.

Python Client & Server

Reference httpx and FastAPI implementations for integrating custom agents into the TalkOps mesh.

Core characteristics

Characteristic	Specification
Transport	JSON-RPC 2.0 over HTTPS with TLS 1.3 encryption
Execution Model	Stateful, multi-turn dialogue supporting synchronous and long-running asynchronous tasks
Authentication	OAuth 2.0 Bearer tokens, scoped API keys, or mutual TLS (mTLS)
Capability Discovery	Standardized Agent Cards advertising inputs, outputs, and endpoint metadata
Multi-Modal Payloads	Natural language text, schema-validated JSON objects, diffs, and external resource URIs

Protocol flow

When a supervisor delegates a task to a specialized agent over A2A, it discovers the target agent's schema and invokes execute_task via JSON-RPC 2.0:

Five pillars of A2A

1. Unified transport

All A2A communication uses JSON-RPC 2.0 envelopes over HTTPS.

Request payload:

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "method": "execute_task",
  "params": {
    "task": "provision_cluster",
    "config": {
      "provider": "aws",
      "region": "us-east-1"
    }
  }
}

Success response:

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "result": {
    "status": "success",
    "cluster_id": "eks-prod-01"
  }
}

Error response:

{
  "jsonrpc": "2.0",
  "id": "req-001",
  "error": {
    "code": -32600,
    "message": "Invalid Request"
  }
}

2. Agent discovery

Every A2A-compliant agent publishes an Agent Card describing its identity, version, supported authentication schemes, and callable capabilities:

{
  "name": "Cloud Orchestration Agent",
  "version": "1.0.0",
  "description": "Provisions and manages multi-cloud infrastructure via Terraform",
  "capabilities": [
    {
      "name": "provision_vpc",
      "description": "Create a production Virtual Private Cloud",
      "inputs": {
        "provider": "aws|azure|gcp",
        "cidr_block": "string",
        "region": "string"
      },
      "outputs": {
        "vpc_id": "string",
        "status": "success|failure"
      }
    },
    {
      "name": "provision_cluster",
      "description": "Create a managed Kubernetes cluster",
      "inputs": {
        "provider": "aws|azure|gcp",
        "node_count": "integer",
        "instance_type": "string"
      },
      "outputs": {
        "cluster_arn": "string",
        "endpoint": "string"
      }
    }
  ],
  "authentication": ["oauth2", "api_key"],
  "endpoint": "https://cloud-agent.talkops.local/a2a"
}

3. Task management

Infrastructure operations frequently span multiple minutes and require human approval checkpoints. A2A tracks multi-turn task state explicitly:

4. Multi-modal data payloads

A2A messages carry structured payloads alongside natural language explanations:

Payload Type	Description
Text	Natural language summaries and reasoning traces
Structured Data	Schema-validated JSON objects, Terraform plans, and metric series
Files & Diffs	Unified Git diffs, YAML manifests, or presigned artifact URLs
References	External MCP resource URIs (for example, tempo://trace/{id})

5. Enterprise security controls

Control	Implementation
Authentication	OAuth 2.0 JWTs, scoped API keys, or mutual TLS (mTLS)
Authorization	Role-Based Access Control (RBAC) and capability-level scopes
Encryption	Mandatory TLS 1.3 in transit
Audit Logging	Immutable request/response correlation via id and thread_id
Rate Limiting	Per-caller token bucket quotas

Standard methods and error codes

Standard methods

Method	Description
discover	Retrieve the agent's capability card (AgentCard)
execute_task	Invoke a named capability with structured parameters
get_status	Poll the execution state of an asynchronous task
cancel_task	Gracefully abort an in-progress task
list_tasks	Enumerate active tasks scoped to the caller

Error codes

Code	Name	Description
-32700	Parse error	Invalid JSON payload received by the server
-32600	Invalid Request	JSON payload is not a valid JSON-RPC 2.0 request
-32601	Method not found	Requested method or capability does not exist
-32602	Invalid params	Input parameters failed JSON Schema validation
-32603	Internal error	Unhandled internal agent execution error
-32000	Authentication error	Missing, expired, or invalid credentials
-32001	Authorization error	Caller lacks permission for the target capability
-32002	Rate limited	Request quota exceeded for caller identity

Implementation example

Python A2A client

import uuid
import httpx


class A2AClient:
    def __init__(self, endpoint: str, api_key: str):
        self.endpoint = endpoint
        self.headers = {
            "Authorization": f"Bearer {api_key}",
            "Content-Type": "application/json",
        }

    async def discover(self) -> dict:
        """Retrieve the remote agent's Agent Card."""
        return await self._request("discover", {})

    async def execute_task(self, task: str, params: dict) -> dict:
        """Execute a named task on the remote agent."""
        return await self._request("execute_task", {"task": task, "params": params})

    async def _request(self, method: str, params: dict) -> dict:
        payload = {
            "jsonrpc": "2.0",
            "id": str(uuid.uuid4()),
            "method": method,
            "params": params,
        }
        async with httpx.AsyncClient(timeout=60.0) as client:
            response = await client.post(
                self.endpoint,
                headers=self.headers,
                json=payload,
            )
            response.raise_for_status()
            return response.json()

Python A2A server

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()


class A2ARequest(BaseModel):
    jsonrpc: str = "2.0"
    id: str
    method: str
    params: dict = {}


@app.post("/a2a")
async def handle_a2a(request: A2ARequest):
    if request.method == "discover":
        return {
            "jsonrpc": "2.0",
            "id": request.id,
            "result": AGENT_CARD,
        }
    if request.method == "execute_task":
        result = await execute_task(request.params)
        return {
            "jsonrpc": "2.0",
            "id": request.id,
            "result": result,
        }
    return {
        "jsonrpc": "2.0",
        "id": request.id,
        "error": {"code": -32601, "message": "Method not found"},
    }

Production best practices

• Enforce TLS 1.3 and mTLS for all inter-agent traffic crossing cluster or VPC boundaries.
• Validate all inputs against the capability's JSON Schema before invoking underlying MCP tools.
• Implement exponential backoff with jitter on network timeouts (-32603) while avoiding automatic retries on mutating tasks unless idempotency keys are present.
• Propagate trace context (traceparent) in HTTP headers so distributed agent calls appear as a single trace in Grafana Tempo.