A2A Protocol

The A2A (Agent-to-Agent) protocol is the communication backbone of Solace Agent Mesh. It defines how agents, gateways, and orchestrators exchange messages over a Solace PubSub+ message broker using JSON-RPC 2.0.

Protocol Overview

A2A builds on three foundational standards:

JSON-RPC 2.0

Request/response envelope structure

Solace Topics

Hierarchical publish/subscribe routing

A2A Spec

Message types, content format, and lifecycle

Key Characteristics

Asynchronous: Non-blocking request/response via pub/sub
Streaming: Support for incremental results (Server-Sent Events style)
Routable: Topic-based routing enables dynamic discovery
Typed: Strong schema validation via Pydantic models
Extensible: Metadata and extensions for custom data

Message Structure

JSON-RPC Envelope

All A2A messages follow JSON-RPC 2.0:

{
  "jsonrpc": "2.0",
  "id": "task-abc123",
  "method": "agent/sendStreamingMessage",
  "params": {
    "message": { /* A2A Message */ },
    "metadata": { /* Optional metadata */ }
  }
}

jsonrpc

string

required

Always “2.0” per JSON-RPC specification

string | number

required

Unique request identifier for correlation (typically task ID)

method

string

required

RPC method name (e.g., “agent/sendStreamingMessage”)

params

object

Method-specific parameters

A2A Message Format

From the A2A specification:

interface Message {
  role: "user" | "agent";
  parts: Part[];
  context_id?: string;
  metadata?: Record<string, any>;
}

type Part = TextPart | DataPart | FilePart;

interface TextPart {
  text: string;
  metadata?: Record<string, any>;
}

interface DataPart {
  data: Record<string, any>;
  metadata?: Record<string, any>;
}

interface FilePart {
  file: FileWithUri | FileWithBytes;
  metadata?: Record<string, any>;
}

Message Types

Request Messages

SendStreamingMessageRequest
SendMessageRequest
CancelTaskRequest

Request streaming execution (incremental results):

{
  "jsonrpc": "2.0",
  "id": "task-123",
  "method": "agent/sendStreamingMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {
          "text": "Analyze the quarterly sales data"
        }
      ],
      "context_id": "session-abc",
      "metadata": {
        "agent_name": "analysis_agent",
        "priority": "high"
      }
    }
  }
}

From src/solace_agent_mesh/common/a2a/protocol.py:554-571:

def create_send_streaming_message_request(
    message: Message,
    task_id: str,
    metadata: Optional[Dict[str, Any]] = None,
) -> SendStreamingMessageRequest:
    """Creates a SendStreamingMessageRequest object."""
    send_params = MessageSendParams(
        message=message, 
        metadata=metadata
    )
    return SendStreamingMessageRequest(
        id=task_id, 
        params=send_params
    )

Request non-streaming execution (single final result):

{
  "jsonrpc": "2.0",
  "id": "task-456",
  "method": "agent/sendMessage",
  "params": {
    "message": {
      "role": "user",
      "parts": [
        {"text": "What is 2+2?"}
      ]
    }
  }
}

Use when: Simple request/response with no intermediate updates needed.

Cancel an in-progress task:

{
  "jsonrpc": "2.0",
  "id": "cancel-req-789",
  "method": "agent/cancelTask",
  "params": {
    "id": "task-123"
  }
}

From src/solace_agent_mesh/common/a2a/protocol.py:520-531:

def create_cancel_task_request(task_id: str) -> CancelTaskRequest:
    """Creates a CancelTaskRequest object."""
    params = TaskIdParams(id=task_id)
    return CancelTaskRequest(
        id=uuid.uuid4().hex, 
        params=params
    )

Response Messages

TaskStatusUpdateEvent
TaskArtifactUpdateEvent
Task (Final Result)
Error Response

Incremental update during streaming execution:

{
  "jsonrpc": "2.0",
  "id": "task-123",
  "result": {
    "type": "task.status.update",
    "task_id": "task-123",
    "context_id": "session-abc",
    "status": {
      "state": "working",
      "message": {
        "role": "agent",
        "parts": [
          {
            "text": "Analyzing data... Found 1,234 records."
          }
        ]
      },
      "is_final": false
    }
  }
}

Published to: Status topic for streaming updates

Artifact (file) update during execution:

{
  "jsonrpc": "2.0",
  "id": "task-123",
  "result": {
    "type": "task.artifact.update",
    "task_id": "task-123",
    "context_id": "session-abc",
    "artifact": {
      "artifact_id": "artifact-xyz",
      "name": "sales_chart.png",
      "description": "Q4 sales visualization",
      "parts": [
        {
          "file": {
            "name": "sales_chart.png",
            "mime_type": "image/png",
            "uri": "artifact://session-abc/sales_chart.png"
          }
        }
      ]
    }
  }
}

Published to: Status topic for streaming updates

Final task result:

{
  "jsonrpc": "2.0",
  "id": "task-123",
  "result": {
    "id": "task-123",
    "context_id": "session-abc",
    "status": {
      "state": "completed",
      "message": {
        "role": "agent",
        "parts": [
          {
            "text": "Analysis complete. See attached report."
          }
        ]
      },
      "is_final": true
    },
    "artifacts": [
      {
        "artifact_id": "report-123",
        "name": "analysis_report.pdf",
        "parts": [...]
      }
    ]
  }
}

Published to: Response topic (final result)

Error result:

{
  "jsonrpc": "2.0",
  "id": "task-123",
  "error": {
    "code": -32000,
    "message": "Agent execution failed",
    "data": {
      "error_type": "timeout",
      "details": "Task exceeded 5 minute timeout"
    }
  }
}

From src/solace_agent_mesh/common/a2a/protocol.py:412-430:

def create_internal_error_response(
    message: str,
    request_id: Optional[Union[str, int]],
    data: Optional[Dict[str, Any]] = None,
) -> JSONRPCResponse:
    """Creates a JSON-RPC response for an InternalError."""
    error = InternalError(message=message, data=data)
    return JSONRPCResponse(id=request_id, error=error)

Topic Structure

A2A uses hierarchical Solace topics for routing:

Topic Hierarchy

{namespace}/a2a/v1/{category}/{type}/{identifier}

namespace

string

required

Organization/tenant namespace (e.g., “acme/ai”)

Topic Patterns

From src/solace_agent_mesh/common/a2a/protocol.py:34-215:

Agent Topics
Gateway Topics
Discovery Topics
Client Topics

Agent Request Topic:

def get_agent_request_topic(namespace: str, agent_name: str) -> str:
    return f"{namespace}/a2a/v1/agent/request/{agent_name}"

# Example: "acme/ai/a2a/v1/agent/request/research_agent"

Agent Response Topic (for peer agents):

def get_agent_response_topic(
    namespace: str, 
    delegating_agent_name: str, 
    sub_task_id: str
) -> str:
    return f"{namespace}/a2a/v1/agent/response/{delegating_agent_name}/{sub_task_id}"

# Example: "acme/ai/a2a/v1/agent/response/orchestrator/subtask-xyz"

Agent Status Topic:

def get_peer_agent_status_topic(
    namespace: str,
    delegating_agent_name: str,
    sub_task_id: str
) -> str:
    return f"{namespace}/a2a/v1/agent/status/{delegating_agent_name}/{sub_task_id}"

Subscription Patterns:

# Subscribe to all responses for this agent
get_agent_response_subscription_topic(namespace, agent_name)
# Returns: "acme/ai/a2a/v1/agent/response/{agent_name}/>"

# Subscribe to all status updates for this agent
get_agent_status_subscription_topic(namespace, agent_name)
# Returns: "acme/ai/a2a/v1/agent/status/{agent_name}/>"

Gateway Response Topic:

def get_gateway_response_topic(
    namespace: str,
    gateway_id: str,
    task_id: str
) -> str:
    return f"{namespace}/a2a/v1/gateway/response/{gateway_id}/{task_id}"

Gateway Status Topic:

def get_gateway_status_topic(
    namespace: str,
    gateway_id: str,
    task_id: str
) -> str:
    return f"{namespace}/a2a/v1/gateway/status/{gateway_id}/{task_id}"

Subscription Patterns:

# Subscribe to all responses for this gateway
get_gateway_response_subscription_topic(namespace, gateway_id)
# Returns: "acme/ai/a2a/v1/gateway/response/{gateway_id}/>"

# Subscribe to all status updates for this gateway
get_gateway_status_subscription_topic(namespace, gateway_id)
# Returns: "acme/ai/a2a/v1/gateway/status/{gateway_id}/>"

Agent Card Publishing:

def get_agent_discovery_topic(namespace: str) -> str:
    return f"{namespace}/a2a/v1/discovery/agentcards"

Gateway Card Publishing:

def get_gateway_discovery_topic(namespace: str) -> str:
    return f"{namespace}/a2a/v1/discovery/gatewaycards"

Discovery Subscription (all):

def get_discovery_subscription_topic(namespace: str) -> str:
    return f"{namespace}/a2a/v1/discovery/>"
# Receives both agent and gateway cards

Client Response Topic:

def get_client_response_topic(namespace: str, client_id: str) -> str:
    return f"{namespace}/a2a/v1/client/response/{client_id}"

Client Status Topic:

def get_client_status_topic(
    namespace: str,
    client_id: str,
    task_id: str
) -> str:
    return f"{namespace}/a2a/v1/client/status/{client_id}/{task_id}"

Client Status Subscription:

def get_client_status_subscription_topic(
    namespace: str,
    client_id: str
) -> str:
    return f"{namespace}/a2a/v1/client/status/{client_id}/>"

Message Flow Patterns

Pattern 1: Gateway → Agent (Streaming)

Pattern 2: Agent → Agent (Delegation)

Pattern 3: Task Cancellation

User Properties (Metadata)

Solace messages can include user properties for routing and context:

user_properties = {
    "clientId": "gateway_123",          # Sender identifier
    "userId": "user_abc",               # End-user identifier
    "replyTo": response_topic,          # Where to send response
    "a2aStatusTopic": status_topic,     # Where to send status updates
    "a2aUserConfig": user_config_json,  # User-specific configuration
    "authToken": signed_jwt_token,      # Optional: Enterprise auth
}

component.publish_a2a_message(
    payload=request.model_dump(exclude_none=True),
    topic=target_topic,
    user_properties=user_properties
)

clientId

string

required

Identifier of the sending component (gateway ID or agent name)

userId

string

End-user identifier for access control and personalization

replyTo

string

required

Topic where the agent should publish the final Task result

a2aStatusTopic

string

Topic where the agent should publish streaming status updates (SendStreamingMessage only)

a2aUserConfig

string (JSON)

User-specific configuration resolved by middleware (scopes, preferences, etc.)

Content Part Types

TextPart

Plain text content:

{
  "text": "The analysis shows a 15% increase in sales.",
  "metadata": {
    "language": "en",
    "confidence": 0.95
  }
}

DataPart

Structured data (JSON):

{
  "data": {
    "type": "tool_call",
    "function_name": "web_search",
    "arguments": {
      "query": "latest AI news"
    }
  },
  "metadata": {
    "tool_call_id": "call_xyz"
  }
}

Common data part types:

tool_call: Function call from LLM
tool_result: Function execution result
structured_output: Validated JSON response
agent_progress_update: Custom status information

FilePart

File/artifact with URI or bytes:

// With URI (recommended for large files)
{
  "file": {
    "name": "report.pdf",
    "mime_type": "application/pdf",
    "uri": "artifact://session-abc/report.pdf"
  },
  "metadata": {
    "size_bytes": 1048576,
    "version": 2
  }
}

// With inline bytes (for small files)
{
  "file": {
    "name": "config.json",
    "mime_type": "application/json",
    "bytes": "eyJrZXkiOiJ2YWx1ZSJ9"  // base64 encoded
  }
}

Helper Functions

The A2A SDK provides helper functions for common operations:

Creating Messages

from solace_agent_mesh.common import a2a

# Create user message
message = a2a.create_user_message(
    parts=[
        a2a.create_text_part("Hello, agent!"),
        a2a.create_file_part_from_uri(
            uri="artifact://session-123/data.csv",
            name="data.csv",
            mime_type="text/csv"
        )
    ],
    metadata={"priority": "high"},
    context_id="session-123"
)

# Create agent message
response_msg = a2a.create_agent_message(
    parts=[a2a.create_text_part("Analysis complete!")]
)

# Create data message
data_msg = a2a.create_agent_data_message(
    data={"status": "processing", "progress": 0.5}
)

Creating Events

# Create status update
status_event = a2a.create_status_update(
    task_id="task-123",
    context_id="session-abc",
    message=response_msg,
    is_final=False
)

# Create artifact update  
artifact_event = a2a.create_artifact_update(
    task_id="task-123",
    context_id="session-abc",
    artifact=artifact_obj
)

Extracting Data

# Get message from status update
message = a2a.get_message_from_status_update(status_event)

# Get parts from message
parts = a2a.get_parts_from_message(message)

# Extract task ID from topic
task_id = a2a.extract_task_id_from_topic(
    topic="acme/ai/a2a/v1/gateway/response/web_ui/task-123",
    subscription_pattern="acme/ai/a2a/v1/gateway/response/web_ui/>",
    log_identifier="[Gateway]"
)

Best Practices

Topic Design

Use clear, consistent namespaces
Include version in topic path (e.g., /a2a/v1/)
Use wildcards (>) for subscription patterns
Avoid overly deep topic hierarchies

Message Size

Use artifact URIs for large files (> 1MB)
Compress large text content when possible
Stream large datasets in chunks
Monitor broker message size limits

Error Handling

Always include meaningful error messages
Use JSON-RPC error codes consistently
Include context in error data field
Log errors with correlation IDs

Metadata Usage

Use metadata for non-essential context
Keep metadata JSON-serializable
Document custom metadata fields
Avoid sensitive data in metadata

Next Steps

Agents

Learn how agents implement A2A communication

Gateways

See how gateways translate external protocols to A2A

Workflows

Orchestrate agents with workflow DAGs

Message Broker Setup

Configure Solace PubSub+ for A2A

Getting Started

Installation & Configuration

Core Concepts

Components

Built-in Tools

Developer Guides

Deployment

Enterprise Features

​A2A Protocol

​Protocol Overview

JSON-RPC 2.0

Solace Topics

A2A Spec

​Key Characteristics

​Message Structure

​JSON-RPC Envelope

​A2A Message Format

​Message Types

​Request Messages

​Response Messages

​Topic Structure

​Topic Hierarchy

​Topic Patterns

​Message Flow Patterns

​Pattern 1: Gateway → Agent (Streaming)

​Pattern 2: Agent → Agent (Delegation)

​Pattern 3: Task Cancellation

​User Properties (Metadata)

​Content Part Types

​TextPart

​DataPart

​FilePart

​Helper Functions

​Creating Messages

​Creating Events

​Extracting Data

​Best Practices

​Next Steps

Agents

Gateways

Workflows

Message Broker Setup

Build docs developers (and LLMs) love

A2A Protocol

Protocol Overview

Key Characteristics

Message Structure

JSON-RPC Envelope

A2A Message Format

Message Types

Request Messages

Response Messages

Topic Structure

Topic Hierarchy

Topic Patterns

Message Flow Patterns

Pattern 1: Gateway → Agent (Streaming)

Pattern 2: Agent → Agent (Delegation)

Pattern 3: Task Cancellation

User Properties (Metadata)

Content Part Types

TextPart

DataPart

FilePart

Helper Functions

Creating Messages

Creating Events

Extracting Data

Best Practices

Next Steps