ra2a Rust SDK

The ra2a crate is a Rust implementation of the Agent2Agent (A2A) Protocol v1.0. It is designed for building agentic applications that require secure, type-safe communication between agents.

Key features include:

• A2A Protocol v1.0 Compliance: Full support for the latest specification.
• Async/Await: Built on tokio for high-performance asynchronous operations.
• Type-Safe Models: Uses strongly typed models with newtype IDs and serde serialization.
• Extensibility: Modular design supporting optional features like grpc, telemetry (OpenTelemetry), and server/client modes.

The ClientFactory automatically handles multi-tenancy. If an AgentInterface within an AgentCard contains a non-empty tenant field, the factory wraps the selected transport in a TenantTransportDecorator.

This decorator ensures that the tenant ID is automatically injected into the tenant field of every request (e.g., SendMessageRequest, GetTaskRequest, etc.) before it is passed to the underlying transport, preventing the need for manual injection in your application logic.

A CallContext is created by the transport layer for every incoming request. It is passed to CallInterceptors and handlers, providing access to:

• The current handler method name.
• The RequestMeta (headers/metadata).
• The authenticated User.
• Activated and requested extension URIs.

Extensions can be checked via is_extension_active(uri) or is_extension_requested(uri).

// Inside a handler or interceptor
fn handle(ctx: &CallContext) {
    let method = ctx.method();
    let user = &ctx.user;
    let is_ext = ctx.is_extension_requested("ext://my-plugin");
}

The AgentCapabilities struct defines high-level features supported by the agent. Key capabilities include:

• streaming: Indicates if the agent supports streaming responses.
• push_notifications: Indicates if the agent supports push notifications.
• extended_agent_card: Indicates if the agent provides an extended card when authenticated.
• extensions: A list of AgentExtension objects for protocol-specific features.

To enable agent-to-agent chaining (e.g., Agent A $\rightarrow$ Agent B $\rightarrow$ Agent C) where extension metadata is preserved, use the propagation components:

1. ServerPropagator: A server-side interceptor that extracts extension-related metadata and headers from incoming requests.
2. ClientPropagator: A client-side interceptor that injects that extracted metadata into outgoing requests to the next agent in the chain.

Use init_propagation to set up the propagation logic.

The A2A REST API supports Server-Sent Events (SSE) for real-time updates through two specific endpoints:

1. POST /message:stream: Streams messages using SSE.
2. GET /tasks/{id}:subscribe: Subscribes to updates for a specific task using SSE.

Both endpoints return an event stream where each event contains a JSON-serialized payload. If an error occurs within the stream, the error is serialized into the JSON data of the event.

The TaskVersion struct is used to track the version of a task stored on the server. This is critical for preventing lost updates in concurrent environments.

When calling TaskStore::save, you should provide the version you last read. If the version in the store has changed since your last read, the operation should fail with a concurrency error (e.g., A2AError::ConcurrentTaskModification).

• Use TaskVersion::MISSING (value 0) when version tracking is not active or for initial creation where no previous version exists.

pub struct TaskVersion(pub i64);

impl TaskVersion {
    pub const MISSING: Self = Self(0);
}

The A2A protocol (v1.0) uses JSON member names as discriminators for streaming events instead of a dedicated kind field. When consuming or producing events via the StreamResponse enum, expect the following JSON structures:

• {"task": {...}} (Full task snapshot)
• {"message": {...}} (Direct message)
• {"statusUpdate": {...}} (Task status change)
• {"artifactUpdate": {...}} (Artifact creation or update)

To enable the ServerPropagator to store extracted data and allow the ClientPropagator to read it later, you must wrap your request handler in init_propagation. This sets up the necessary task-local storage scope.

let result = ra2a_ext::init_propagation(async {
    // ServerPropagator.before() will store data here
    // Your handler runs
    // ClientPropagator.before() will be able to read data here
    handle_request(req).await
}).await;

let result = ra2a_ext::init_propagation(async {
    // ServerPropagator.before() stores data here
    // handler runs
    // ClientPropagator.before() reads data here
    handle_request(req).await
}).await;

For production environments requiring persistent storage, use the SQL-backed implementations provided via the sql feature. These implementations use SQLx and support different database backends.

Available Implementations

• SqliteTaskStore: For SQLite databases.
• PostgresTaskStore: For PostgreSQL databases.
• MySqlTaskStore: For MySQL databases.

Initialization

Before using a SQL store, you must call .initialize() to create the necessary tables (e.g., a2a_tasks).

Schema Details

• SQLite: Uses standard text types for history, artifacts, and metadata.
• PostgreSQL: Uses JSONB for optimized storage of history, artifacts, and metadata.

// Example pattern for initializing a SQL store
let pool = sqlx::SqlitePool::connect("sqlite:tasks.db").await?;
let store = SqliteTaskStore::new(pool);
store.initialize().await?;