Task & Config Stores

a2a-rust uses pluggable storage backends for tasks and push notification configs. The built-in in-memory stores work for development and testing. For production, implement the traits for your database.

TaskStore Trait

The TaskStore trait defines how tasks are persisted:

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use a2a_protocol_sdk::types::error::A2aResult;
use a2a_protocol_sdk::types::params::ListTasksParams;
use a2a_protocol_sdk::types::responses::TaskListResponse;
use a2a_protocol_sdk::types::task::{Task, TaskId};
pub trait TaskStore: Send + Sync + 'static {
    fn save<'a>(&'a self, task: &'a Task)
        -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>;

    fn get<'a>(&'a self, id: &'a TaskId)
        -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>>;

    fn list<'a>(&'a self, params: &'a ListTasksParams)
        -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>>;

    fn insert_if_absent<'a>(&'a self, task: &'a Task)
        -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>>;

    fn delete<'a>(&'a self, id: &'a TaskId)
        -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>;

    /// Returns the total number of tasks. Default returns 0.
    fn count<'a>(&'a self)
        -> Pin<Box<dyn Future<Output = A2aResult<u64>> + Send + 'a>>;

    // Further provided methods — idempotency keys, delta saves, an event
    // log for stream resumption — have defaults; see the API docs.
}
// The signatures above are the real trait's, and they are all it requires:
// this impl of it defines exactly these methods.
struct Probe;
impl a2a_protocol_sdk::server::store::TaskStore for Probe {
    fn save<'a>(&'a self, _: &'a Task) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { Box::pin(async { Ok(()) }) }
    fn get<'a>(&'a self, _: &'a TaskId) -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>> { Box::pin(async { Ok(None) }) }
    fn list<'a>(&'a self, _: &'a ListTasksParams) -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>> { Box::pin(async { Ok(TaskListResponse::new(vec![])) }) }
    fn insert_if_absent<'a>(&'a self, _: &'a Task) -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>> { Box::pin(async { Ok(true) }) }
    fn delete<'a>(&'a self, _: &'a TaskId) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { Box::pin(async { Ok(()) }) }
    fn count<'a>(&'a self) -> Pin<Box<dyn Future<Output = A2aResult<u64>> + Send + 'a>> { Box::pin(async { Ok(0) }) }
}
}

InMemoryTaskStore

The default implementation with optional TTL and capacity limits:

#![allow(unused)]
fn main() {
use a2a_protocol_sdk::server::{InMemoryTaskStore, TaskStoreConfig};
use std::time::Duration;

// Default: 1hr TTL, 10k capacity
let store = InMemoryTaskStore::new();

// With custom limits
let store = InMemoryTaskStore::with_config(
    TaskStoreConfig::default()
        .with_task_ttl(Some(Duration::from_secs(7200))) // 2 hour TTL
        .with_max_capacity(Some(50_000)),
);
}

Features:

• Thread-safe (RwLock — concurrent readers, exclusive writers)
• Pre-allocated HashMap::with_capacity(max_capacity) — eliminates resize-induced latency spikes under load
• O(1) amortized save()/get()/delete() via HashMap (no log(n) tree traversal overhead)
• O(log n + page_size) list() with secondary indexes:
  • BTreeMap<u64, TaskId> update-order index keyed by a monotonic per-write sequence — iterated in reverse to return tasks most-recently-updated first (spec §3.1.4), with no O(n log n) per-call sort
  • HashMap<String, BTreeMap<u64, TaskId>> context index — O(log m + page_size) filtered queries
  • The sequence is a collision-free integer pagination cursor; every write re-positions its task to the front of the update order
• Automatic TTL eviction on access (maintains all indexes)
• Capacity eviction (oldest terminal tasks first; falls back to non-terminal tasks when needed) when limit exceeded — save does not return with the store over capacity. Measured with 16 concurrent writers against a capacity of 100: a concurrent reader saw a peak of 102 and a final count of exactly 100, so the transient overshoot is bounded by the number of writes in flight
• Cursor-based pagination, most-recently-updated first
• Filtering by context_id (index-accelerated) and status

SqliteTaskStore (feature-gated)

Enable the sqlite feature for a production-ready persistent store:

[dependencies]
a2a-protocol-server = { version = "0.14", features = ["sqlite"] }

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
use a2a_protocol_server::store::SqliteTaskStore;

let store = SqliteTaskStore::new("sqlite:tasks.db").await?;
// Or use an in-memory database for testing:
let store = SqliteTaskStore::new("sqlite::memory:").await?;
Ok(())
}

Features:

• Auto-creates schema on first use
• Stores tasks as JSON blobs with indexed context_id and state columns
• Cursor-based pagination ordered by (updated_at DESC, id DESC) — tasks most-recently-updated first (spec §3.1.4), with a composite row-value cursor that never drops or repeats a row even when timestamps tie
• Atomic insert_if_absent via INSERT OR IGNORE
• Upsert via ON CONFLICT DO UPDATE
• Production-ready defaults: WAL journal mode, busy_timeout=5000ms, synchronous=NORMAL, foreign_keys=ON, pool size of 8

TenantAwareInMemoryTaskStore

For multi-tenant deployments, use TenantAwareInMemoryTaskStore which provides full tenant isolation using tokio::task_local!:

use a2a_protocol_sdk::types::task::{ContextId, Task, TaskId, TaskState, TaskStatus};
#[tokio::main]
async fn main() -> () {
let task_id = TaskId::new("task-1");
let task = Task { id: task_id.clone(), context_id: ContextId::new("ctx-1"),
    status: TaskStatus::new(TaskState::Submitted), history: None, artifacts: None, metadata: None };
use a2a_protocol_server::store::{TaskStore, TenantAwareInMemoryTaskStore, TenantContext};
use std::sync::Arc;

let store = Arc::new(TenantAwareInMemoryTaskStore::new());

// Each tenant gets an independent store instance.
// Use TenantContext::scope() to set the active tenant:
TenantContext::scope("tenant-alpha".to_string(), {
    let store = store.clone();
    async move {
        store.save(&task).await.unwrap();
    }
}).await;

// Tasks saved under one tenant are invisible to others:
TenantContext::scope("tenant-beta".to_string(), {
    let store = store.clone();
    async move {
        let result = store.get(&task_id).await.unwrap();
        assert!(result.is_none()); // tenant-beta can't see tenant-alpha's task
    }
}).await;

// Track tenant count for capacity monitoring:
let count = store.tenant_count().await;
}

The TenantContext::scope() pattern uses tokio::task_local! to thread the tenant ID through the async call stack without passing it as a parameter. The RequestHandler automatically sets the tenant scope when params.tenant is populated.

TenantAwareSqliteTaskStore (feature-gated)

For persistent multi-tenant storage, enable the sqlite feature:

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
use a2a_protocol_server::store::TenantAwareSqliteTaskStore;

let store = TenantAwareSqliteTaskStore::new("sqlite:tasks.db").await?;
Ok(())
}

This variant partitions data by a tenant_id column instead of using task-local storage, making it suitable for production deployments where tenants may span multiple server instances.

Note: Corresponding TenantAwareInMemoryPushConfigStore and TenantAwareSqlitePushConfigStore variants exist for push notification config storage.

PostgresTaskStore (postgres feature)

PostgreSQL-backed stores ship with the crate — PostgresTaskStore, PostgresPushConfigStore, tenant-aware variants, and a forward-only migration runner — and are exercised against a live PostgreSQL 16 service in CI (postgres_store_tests.rs):

use a2a_protocol_sdk::prelude::*;
struct MyExecutor;
agent_executor!(MyExecutor, |_ctx, _queue| async { Ok(()) });
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
use a2a_protocol_server::store::PostgresTaskStore;

let store = PostgresTaskStore::with_migrations("postgres://user:pass@localhost/a2a").await?;
let handler = RequestHandlerBuilder::new(MyExecutor)
    .with_task_store(store)
    .build()?;
Ok(())
}

Custom Implementation

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use a2a_protocol_sdk::types::error::A2aResult;
use a2a_protocol_sdk::types::params::ListTasksParams;
use a2a_protocol_sdk::types::responses::TaskListResponse;
use a2a_protocol_sdk::types::task::{Task, TaskId};
use a2a_protocol_sdk::types::error::A2aError;
use a2a_protocol_sdk::server::store::TaskStore;
struct PgJsonTaskStore {
    pool: sqlx::PgPool,
}

impl TaskStore for PgJsonTaskStore {
    fn get<'a>(&'a self, id: &'a TaskId)
        -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>>
    {
        Box::pin(async move {
            let data: Option<serde_json::Value> =
                sqlx::query_scalar("SELECT data FROM tasks WHERE id = $1")
                    .bind(id.as_ref())
                    .fetch_optional(&self.pool)
                    .await
                    .map_err(|e| A2aError::internal(e.to_string()))?;

            data.map(serde_json::from_value)
                .transpose()
                .map_err(|e| A2aError::internal(format!("stored task does not parse: {e}")))
        })
    }

    // ... implement save, list, delete, insert_if_absent similarly
    fn save<'a>(&'a self, _: &'a Task) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { unimplemented!() }
    fn list<'a>(&'a self, _: &'a ListTasksParams) -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>> { unimplemented!() }
    fn insert_if_absent<'a>(&'a self, _: &'a Task) -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>> { unimplemented!() }
    fn delete<'a>(&'a self, _: &'a TaskId) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { unimplemented!() }
}
}

PushConfigStore Trait

The PushConfigStore trait manages push notification configurations:

#![allow(unused)]
fn main() {
use std::future::Future;
use std::pin::Pin;
use a2a_protocol_sdk::types::error::A2aResult;
use a2a_protocol_sdk::types::push::TaskPushNotificationConfig;
pub trait PushConfigStore: Send + Sync + 'static {
    fn set<'a>(&'a self, config: TaskPushNotificationConfig)
        -> Pin<Box<dyn Future<Output = A2aResult<TaskPushNotificationConfig>> + Send + 'a>>;

    fn get<'a>(&'a self, task_id: &'a str, id: &'a str)
        -> Pin<Box<dyn Future<Output = A2aResult<Option<TaskPushNotificationConfig>>> + Send + 'a>>;

    fn list<'a>(&'a self, task_id: &'a str)
        -> Pin<Box<dyn Future<Output = A2aResult<Vec<TaskPushNotificationConfig>>> + Send + 'a>>;

    fn delete<'a>(&'a self, task_id: &'a str, id: &'a str)
        -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>>;

    // Optional: total stored configs, counted against the handler's global
    // push-config ceiling. The default implementation returns `None`
    // (only the per-task cap is then enforced), so existing custom
    // implementations keep compiling unchanged.
    fn count(&self)
        -> Pin<Box<dyn Future<Output = A2aResult<Option<usize>>> + Send + '_>> {
        Box::pin(async { Ok(None) })
    }
}
// The real trait requires exactly these four; this impl of it compiles.
struct Probe;
impl a2a_protocol_sdk::server::PushConfigStore for Probe {
    fn set<'a>(&'a self, c: TaskPushNotificationConfig) -> Pin<Box<dyn Future<Output = A2aResult<TaskPushNotificationConfig>> + Send + 'a>> { Box::pin(async { Ok(c) }) }
    fn get<'a>(&'a self, _: &'a str, _: &'a str) -> Pin<Box<dyn Future<Output = A2aResult<Option<TaskPushNotificationConfig>>> + Send + 'a>> { Box::pin(async { Ok(None) }) }
    fn list<'a>(&'a self, _: &'a str) -> Pin<Box<dyn Future<Output = A2aResult<Vec<TaskPushNotificationConfig>>> + Send + 'a>> { Box::pin(async { Ok(vec![]) }) }
    fn delete<'a>(&'a self, _: &'a str, _: &'a str) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { Box::pin(async { Ok(()) }) }
}
}

InMemoryPushConfigStore

The default implementation stores configs in a HashMap with a secondary index for efficient per-task counting:

#![allow(unused)]
fn main() {
use a2a_protocol_sdk::server::InMemoryPushConfigStore;

let store = InMemoryPushConfigStore::new();
}

Features:

• Server-assigned config IDs (UUIDs)
• Per-task config limits (prevents abuse) — uses a secondary index (task_counts) for O(1) per-task count lookups instead of scanning all keys
• Global config limit (max_total_configs, default 100,000) — prevents unbounded memory growth across all tasks
• Thread-safe access

Wiring Custom Stores

use std::future::Future;
use std::pin::Pin;
use a2a_protocol_sdk::prelude::*;
use a2a_protocol_sdk::types::task::TaskId;
use a2a_protocol_sdk::server::store::TaskStore;
struct MyExecutor;
agent_executor!(MyExecutor, |_ctx, _queue| async { Ok(()) });
struct PgJsonTaskStore { pool: sqlx::PgPool }
impl TaskStore for PgJsonTaskStore {
    fn save<'a>(&'a self, _: &'a Task) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { unimplemented!() }
    fn get<'a>(&'a self, _: &'a TaskId) -> Pin<Box<dyn Future<Output = A2aResult<Option<Task>>> + Send + 'a>> { unimplemented!() }
    fn list<'a>(&'a self, _: &'a ListTasksParams) -> Pin<Box<dyn Future<Output = A2aResult<TaskListResponse>> + Send + 'a>> { unimplemented!() }
    fn insert_if_absent<'a>(&'a self, _: &'a Task) -> Pin<Box<dyn Future<Output = A2aResult<bool>> + Send + 'a>> { unimplemented!() }
    fn delete<'a>(&'a self, _: &'a TaskId) -> Pin<Box<dyn Future<Output = A2aResult<()>> + Send + 'a>> { unimplemented!() }
}
#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
let executor = MyExecutor;
let database_url = "postgres://user:pass@localhost/a2a";
use a2a_protocol_server::PostgresPushConfigStore;

let pool = sqlx::PgPool::connect(database_url).await?;
let handler = RequestHandlerBuilder::new(executor)
    .with_task_store(PgJsonTaskStore { pool })
    .with_push_config_store(PostgresPushConfigStore::new(database_url).await?)
    .build()?;
Ok(())
}

Design Considerations

Object Safety

Both traits use Pin<Box<dyn Future>> return types for object safety. This allows the handler to store them as Arc<dyn TaskStore>.

Tenant Isolation

Tenant isolation uses tokio::task_local! via TenantContext::scope(), not method parameters. For in-memory stores, TenantAwareInMemoryTaskStore automatically partitions data by tenant. For SQL stores, TenantAwareSqliteTaskStore partitions by a tenant_id column. If you implement a custom store, use TenantContext::current() to retrieve the active tenant within the async call stack.

TaskStoreConfig Fields

TaskStoreConfig configures InMemoryTaskStore only. The SQL stores take a connection URL, not a config — max_capacity, task_ttl and eviction_interval have no meaning for them (the database holds the rows, and retention is purge_expired), and each carries its own page-size cap via with_max_page_size. Until 2026-08-19 that cap was a hardcoded 1000 that happened to equal this table's default, so setting max_page_size to something tighter changed the in-memory store and nothing else: measured with the cap at 10 and a client asking for 100, the in-memory store returned 10 and SQLite returned all 60 rows it held.

eviction_interval is a tail-latency knob

The TTL sweep is not spawned: it is awaited inside save, and it holds the store's write lock for its whole duration, so the write that triggers it pays for it and every concurrent writer waits. Measured (debug profile, 50,000 terminal tasks, eviction_interval 1000): the quietest of 1,000 consecutive saves took 3.99 µs and the one that swept took 4.54 ms — about 1,100×.

The scan is O(n) unavoidably, which is why it is amortized behind the interval rather than run per write. If p99.9 write latency matters to you, raise eviction_interval and drive cleanup yourself with run_eviction() on your own schedule, so the stall lands where you chose it. The capacity pass is far cheaper (measured 3.5× and 3.7× on two runs at 10,000 entries) because it removes the overflow instead of scanning for it.

Pagination

The list method receives ListTasksParams with:

• page_size — Number of results per page. Capped by TaskStoreConfig::max_page_size on InMemoryTaskStore, and by with_max_page_size on each SQL store; both default to DEFAULT_MAX_PAGE_SIZE (1,000)
• page_token — Opaque cursor for the next page
• Various filter fields

Your implementation should return a TaskListResponse with a next_page_token if more results exist. Per spec §3.1.4, tasks must be returned most-recently-updated first; the built-in stores order by last-update time (descending) with a stable cursor, and a custom store should do the same so pagination is deterministic across updates. The page_token is opaque — encode whatever your ordering needs into it; treat a token you did not issue as an empty page rather than scanning from the top.

Concurrency

Both traits require Send + Sync, and every method takes &self. Use a connection pool rather than a single connection: a query needs &mut access to a connection, so a store holding one has to put it behind a lock, and every call then waits for the one before it.

#![allow(unused)]
fn main() {
// Good — the pool hands each call its own connection
struct MyStore { pool: sqlx::PgPool }

// Compiles, but serializes every store call behind one connection
struct MySerialStore { conn: tokio::sync::Mutex<sqlx::PgConnection> }
fn assert_send_sync<T: Send + Sync>() {}
assert_send_sync::<MyStore>();
assert_send_sync::<MySerialStore>();
}

Terminal states are final

Every store shipped here refuses a write — save or any of the delta methods — that would move a stored Completed, Failed, Canceled or Rejected task to a different state, and applies the check inside the write itself: a WHERE condition on the SQL UPDATE or upsert, the write lock for the in-memory stores. Re-writing the same terminal state is allowed. The refusal is an UnsupportedOperation error carrying a TerminalStateConflict, and the server reacts to it: CancelTask answers TaskNotCancelable, and the event processors adopt the stored state and cancel the local executor.

This is what makes a CancelTask on one replica stick against an executor still running on another (see Running More Than One Replica). A custom store does not get it for free. To give the same guarantee, apply store::refuses_write(stored, written) atomically with each write, and report a refusal with TerminalStateConflict::into_error. A read followed by a write is not enough — that is the race the rule exists to close.

Next Steps

• Production Hardening — Deployment checklist
• Configuration Reference — Every builder and store default