Vapora/crates/vapora-a2a-client

vapora-a2a-client

A2A Protocol Client - Resilient HTTP client for calling Agent-to-Agent (A2A) protocol servers.

Features

• ✅ Full A2A Protocol Support - Discovery, dispatch, status query
• ✅ Exponential Backoff Retry - Configurable retry policy with jitter
• ✅ Smart Error Handling - Retries 5xx/network, skips 4xx
• ✅ Type-Safe - Rust compile-time guarantees
• ✅ Async/Await - Built on Tokio and Reqwest
• ✅ Comprehensive Tests - 5 unit tests, all passing

Quick Start

Basic Usage

use vapora_a2a_client::A2aClient;

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    // Create client
    let client = A2aClient::new("http://localhost:8003");

    // Discover agent capabilities
    let agent_card = client.discover_agent().await?;
    println!("Connected to: {} v{}", agent_card.name, agent_card.version);

    // Dispatch task
    let task_id = client.dispatch_task(
        uuid::Uuid::new_v4().to_string(),
        "Write hello world function".to_string(),
        Some("In Rust with tests".to_string()),
        Some("developer".to_string()),
    ).await?;

    println!("Task dispatched: {}", task_id);

    // Query status
    let status = client.get_task_status(&task_id).await?;
    println!("Status: {:?}", status);

    Ok(())
}

With Custom Timeout

use std::time::Duration;
use vapora_a2a_client::A2aClient;

let client = A2aClient::with_timeout(
    "http://localhost:8003",
    Duration::from_secs(60),
);

With Custom Retry Policy

use vapora_a2a_client::{A2aClient, RetryPolicy};
use std::time::Duration;

let retry_policy = RetryPolicy {
    max_retries: 5,
    initial_delay_ms: 200,
    max_delay_ms: 10000,
    backoff_multiplier: 2.0,
    jitter: true,
};

let client = A2aClient::with_retry_policy(
    "http://localhost:8003",
    Duration::from_secs(30),
    retry_policy,
);

Retry Policy

How It Works

The client automatically retries transient failures using exponential backoff:

Attempt 1: Fail (timeout)
Wait: 100ms (± 20% jitter)

Attempt 2: Fail (5xx error)
Wait: 200ms (± 20% jitter)

Attempt 3: Success

Retryable Errors

Retries (up to max_retries):

• Network timeouts
• Connection refused
• 5xx server errors (500-599)
• Connection reset

No Retry (fails immediately):

• 4xx client errors (400-499)
• Task not found (404)
• Deserialization errors
• Invalid response format

Configuration

pub struct RetryPolicy {
    pub max_retries: u32,        // Default: 3
    pub initial_delay_ms: u64,   // Default: 100ms
    pub max_delay_ms: u64,       // Default: 5000ms
    pub backoff_multiplier: f64, // Default: 2.0
    pub jitter: bool,            // Default: true (±20%)
}

Formula:

delay = min(initial_delay * (multiplier ^ attempt), max_delay)
if jitter: delay *= random(0.8..1.2)

API Reference

Client Creation

// Default timeout (30s), default retry policy
let client = A2aClient::new("http://localhost:8003");

// Custom timeout
let client = A2aClient::with_timeout(
    "http://localhost:8003",
    Duration::from_secs(60),
);

// Custom retry policy
let client = A2aClient::with_retry_policy(
    "http://localhost:8003",
    Duration::from_secs(30),
    RetryPolicy::default(),
);

Methods

discover_agent() -> Result<AgentCard>

Fetches agent capabilities from /.well-known/agent.json:

let agent_card = client.discover_agent().await?;
println!("Name: {}", agent_card.name);
println!("Version: {}", agent_card.version);
println!("Skills: {:?}", agent_card.skills);

dispatch_task(...) -> Result<String>

Dispatches a task to the A2A server:

let task_id = client.dispatch_task(
    "task-123".to_string(),          // task_id (UUID recommended)
    "Task title".to_string(),        // title
    Some("Description".to_string()), // description (optional)
    Some("developer".to_string()),   // skill (optional)
).await?;

get_task_status(task_id: &str) -> Result<A2aTaskStatus>

Queries task status:

let status = client.get_task_status("task-123").await?;

match status.state.as_str() {
    "waiting" => println!("Task queued"),
    "working" => println!("Task in progress"),
    "completed" => println!("Result: {:?}", status.result),
    "failed" => println!("Error: {:?}", status.error),
    _ => {}
}

health_check() -> Result<bool>

Checks server health:

if client.health_check().await? {
    println!("Server healthy");
}

Error Handling

use vapora_a2a_client::{A2aClient, A2aClientError};

match client.dispatch_task(...).await {
    Ok(task_id) => println!("Success: {}", task_id),
    Err(A2aClientError::Timeout(url)) => {
        eprintln!("Timeout connecting to: {}", url);
    }
    Err(A2aClientError::ConnectionRefused(url)) => {
        eprintln!("Connection refused: {}", url);
    }
    Err(A2aClientError::ServerError { code, message }) => {
        eprintln!("Server error {}: {}", code, message);
    }
    Err(A2aClientError::TaskNotFound(id)) => {
        eprintln!("Task not found: {}", id);
    }
    Err(e) => eprintln!("Other error: {}", e),
}

Testing

# Run all tests
cargo test -p vapora-a2a-client

# Output:
# test retry::tests::test_retry_succeeds_eventually ... ok
# test retry::tests::test_retry_exhausted ... ok
# test retry::tests::test_non_retryable_error ... ok
# test client::tests::test_client_creation ... ok
# test client::tests::test_client_with_custom_timeout ... ok
#
# test result: ok. 5 passed; 0 failed; 0 ignored

Examples

Polling for Completion

use tokio::time::{sleep, Duration};

let task_id = client.dispatch_task(...).await?;

loop {
    let status = client.get_task_status(&task_id).await?;

    match status.state.as_str() {
        "completed" => {
            println!("Success: {:?}", status.result);
            break;
        }
        "failed" => {
            eprintln!("Failed: {:?}", status.error);
            break;
        }
        _ => {
            println!("Status: {}", status.state);
            sleep(Duration::from_millis(500)).await;
        }
    }
}

Batch Task Dispatch

use futures::future::join_all;

let tasks = vec!["Task 1", "Task 2", "Task 3"];

let futures = tasks.iter().map(|title| {
    client.dispatch_task(
        uuid::Uuid::new_v4().to_string(),
        title.to_string(),
        None,
        Some("developer".to_string()),
    )
});

let task_ids = join_all(futures).await;

for result in task_ids {
    match result {
        Ok(id) => println!("Dispatched: {}", id),
        Err(e) => eprintln!("Failed: {}", e),
    }
}

Custom Retry Logic

use vapora_a2a_client::{RetryPolicy, A2aClient};
use std::time::Duration;

// Conservative retry: fewer attempts, longer delays
let conservative = RetryPolicy {
    max_retries: 2,
    initial_delay_ms: 500,
    max_delay_ms: 10000,
    backoff_multiplier: 3.0,
    jitter: true,
};

// Aggressive retry: more attempts, shorter delays
let aggressive = RetryPolicy {
    max_retries: 10,
    initial_delay_ms: 50,
    max_delay_ms: 2000,
    backoff_multiplier: 1.5,
    jitter: true,
};

let client = A2aClient::with_retry_policy(
    "http://localhost:8003",
    Duration::from_secs(30),
    conservative,
);

Dependencies

[dependencies]
vapora-a2a = { workspace = true }
reqwest = { workspace = true, features = ["json"] }
tokio = { workspace = true, features = ["full"] }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
thiserror = { workspace = true }
tracing = { workspace = true }
uuid = { workspace = true, features = ["v4", "serde"] }
rand = { workspace = true }

Related Crates

• vapora-a2a - Server implementation
• vapora-agents - Agent coordinator

License

MIT OR Apache-2.0