klieo-a2a

Durable A2A v1.0 protocol layer atop klieo-bus traits.

Part of the klieo Rust agent framework.

Status

0.9.x — public API may evolve before 1.0. See docs/SEMVER.md.

HTTP/SSE transport

klieo-a2a ships an HTTP/SSE transport behind the http cargo feature. Single POST /a2a endpoint per A2A v1.0 §9.4.2 with JSON-or-SSE response negotiation based on JSON-RPC method.

[dependencies]
klieo-a2a = { version = "0.9", features = ["http"] }

use klieo_a2a::auth::AllowAnonymous;
use klieo_a2a::handler::EchoHandler;
use klieo_a2a::http::A2aHttpServer;
use klieo_a2a::task_store::A2aTaskStore;
use klieo_a2a::A2aDispatcher;
use std::sync::Arc;
use tokio_util::sync::CancellationToken;

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let dispatcher = Arc::new(A2aDispatcher::with_in_process_pubsub(
        Arc::new(my_handler()),
        Arc::new(my_authenticator()),
    ));
    let kv = my_kv_store();
    let task_store = Arc::new(
        A2aTaskStore::new(kv, klieo_a2a::task_store::DEFAULT_BUCKET.into())
            .with_event_sink(dispatcher.event_sink()),
    );
    let cancel = CancellationToken::new();

    let server = Arc::new(A2aHttpServer::new(dispatcher, task_store, cancel));
    server.serve_http("127.0.0.1:8080".parse()?).await?;
    Ok(())
}

Streaming methods. SendStreamingMessage and SubscribeToTask return text/event-stream; all other methods return application/json.

Security boundary. The HTTP layer adds no authentication beyond what the configured Authenticator enforces on the JSON-RPC payload. Bind 127.0.0.1 and front with an auth-enforcing reverse proxy for any non-localhost deployment. Plain HTTP only — terminate TLS at the proxy.

See docs/adr/adr-013-a2a-http-sse-transport.md for the design rationale and out-of-scope items.

BREAKING CHANGES (0.10 → 0.11)

The A2aHandler trait's streaming methods changed return type to support the new SSE transport. Custom A2aHandler impls that override either of these methods must update their return type.

Affected methods

• send_streaming_message: was Result<SendMessageResult, A2aError> → now Result<TaskEventStream, A2aError>.
• subscribe_to_task: was Result<Task, A2aError> → now Result<TaskEventStream, A2aError>.

The trait defaults still return Err(A2aError::MethodNotFound(...)), so impls that do NOT override these methods are unaffected.

Migration

For an impl that previously returned a single Task or SendMessageResult, wrap the value in a one-shot stream:

use futures::stream;
use klieo_a2a::{TaskEvent, TaskEventStream};

async fn subscribe_to_task(
    &self,
    _ctx: &RequestContext,
    params: SubscribeToTaskParams,
) -> Result<TaskEventStream, A2aError> {
    let task = self.store.get(&params.id).await?;
    // Replay current state as a single SSE event; the
    // dispatcher's broadcast-derived fallback handles the
    // tail. Or emit your own multi-event stream if you have
    // a custom event source.
    let event = TaskEvent {
        task_id: task.id.clone(),
        status: task.status,
        message: task.history.last().cloned(),
        final_event: true,
    };
    Ok(Box::pin(stream::once(async move { event })))
}

See docs/adr/adr-013-a2a-http-sse-transport.md for the full design rationale.

License

MIT — see LICENSE.