# a2a-amqp

AMQP-backed EventBus and WorkQueue for scaling A2A agents with long-running tasks.

## Why?

A2A agents often need to handle long-running tasks (LLM calls, complex processing, etc.). Running these tasks inline in HTTP handlers causes:

- **Timeout issues**: HTTP connections timeout on long tasks
- **Scaling problems**: Single server bottlenecks
- **Resource waste**: Servers blocked waiting for tasks to complete

This library solves these problems by:

1. **Queuing tasks** via AMQP instead of processing inline
2. **Distributing work** across multiple worker processes
3. **Event sourcing** all task events for replay and recovery
4. **Streaming results** back via SSE while workers process in the background

## Architecture

```
HTTP Request → Server (enqueues task) → Returns immediately
                    ↓
               AMQP Queue
                    ↓
          Worker Pool (scales horizontally)
                    ↓
          Process task & publish events
                    ↓
          AMQP Stream (event sourcing)
                    ↓
          Client streams results via SSE
```

## Installation

```bash
# Installing using bun
bun add @cloudamqp/a2a-amqp @a2a-js/sdk @cloudamqp/amqp-client

# Or with npm
npm install @cloudamqp/a2a-amqp @a2a-js/sdk @cloudamqp/amqp-client
```

**Requires**: LavinMQ or RabbitMQ with stream support

```bash
docker run -d -p 5672:5672 -p 15672:15672 cloudamqp/lavinmq:latest
```

## Quick Start

### 1. HTTP Server (enqueues tasks)

```typescript
import { AMQPAgentBackend, QueuingRequestHandler } from "@84codes/a2a-amqp";
import { A2AExpressApp } from "@a2a-js/sdk/server/express";
import express from "express";

// Create AMQP backend
const backend = await AMQPAgentBackend.create({
  url: "amqp://localhost:5672",
  agentName: "my-agent",
});

// Create request handler (handles task queuing + event projection)
const requestHandler = new QueuingRequestHandler(agentCard, backend);
await requestHandler.initialize();

// Setup Express with A2A routes
const app = express();
new A2AExpressApp(requestHandler).setupRoutes(app, "/");
app.listen(3000);
```

### 2. Worker Process (processes tasks)

```typescript
import { AMQPAgentBackend, WorkerEventBus } from "@84codes/a2a-amqp";
import { AgentExecutor, RequestContext } from "@a2a-js/sdk/server";

// Create backend with same agent name as server
const backend = await AMQPAgentBackend.create({
  url: "amqp://localhost:5672",
  agentName: "my-agent",
});

// Initialize work queue
await backend.workQueue.initialize();

class MyExecutor implements AgentExecutor {
  async execute(context: RequestContext, eventBus: ExecutionEventBus) {
    // Your long-running task logic here
    eventBus.publish({
      kind: "status-update",
      taskId: context.taskId,
      contextId: context.contextId,
      status: { state: "working", timestamp: new Date().toISOString() },
      final: false,
    });

    // ... do work ...

    eventBus.publish({
      kind: "status-update",
      taskId: context.taskId,
      contextId: context.contextId,
      status: { state: "completed", timestamp: new Date().toISOString() },
      final: true,
    });
    eventBus.finished();
  }
}

const executor = new MyExecutor();

// Start consuming with async generator pattern
const messages = backend.workQueue.start();

for await (const taskMessage of messages) {
  const { taskId, contextId, requestContext } = taskMessage;

  // Create request context
  const context = new RequestContext(
    requestContext.userMessage,
    taskId,
    contextId,
    requestContext.task,
    requestContext.referenceTasks
  );

  // Create event bus for publishing task events
  const eventBus = new WorkerEventBus(backend.amqpConnection, taskId, contextId);

  // Execute task
  await executor.execute(context, eventBus);
}
```

### 3. Scale horizontally

Run multiple workers to process tasks in parallel:

```bash
# Terminal 1: HTTP Server
bun run server

# Terminal 2-N: Workers (scale as needed)
bun run worker
bun run worker  # Add more workers for higher throughput
```

## Features

- **Work Queue**: Distribute tasks across multiple worker processes
- **Event Sourcing**: All task events stored in AMQP streams for replay
- **In-Memory Projection**: Fast task lookups with automatic recovery from streams
- **SSE Streaming**: Automatic streaming of task events back to clients
- **Horizontal Scaling**: Add more workers to increase throughput
- **Graceful Shutdown**: Clean consumer and connection handling
- **Type-Safe**: Full TypeScript support with Zod validation

## Configuration

```typescript
interface AMQPAgentBackendConfig {
  url: string;                    // AMQP broker URL
  agentName: string;              // Agent identifier
  streamRetention?: string;       // Event retention (default: "7d")
  streamMaxBytes?: number;        // Max stream size (default: 1GB)
  workQueueName?: string;         // Custom work queue name
  exchangeName?: string;          // Custom exchange name
  logger?: Logger;                // Custom logger
  connection?: {
    heartbeat?: number;           // Heartbeat interval in seconds
    reconnectDelay?: number;      // Reconnection delay in ms
    maxReconnectAttempts?: number;// Max reconnection attempts
  };
  publishing?: {
    persistent?: boolean;         // Persistent messages (default: true)
    confirmMode?: boolean;        // Publisher confirms (default: true)
    messageTtl?: number;          // Message TTL in ms (0 = no expiration)
  };
}
```

## Examples

See complete working examples:

- `src/examples/http-server.ts` - HTTP server with queuing
- `src/examples/worker.ts` - Worker process

```bash
# Run the example
bun run server  # Terminal 1
bun run worker  # Terminal 2

# Send a request
curl -X POST http://localhost:3000/ \
  -H "Content-Type: application/json" \
  -d '{"jsonrpc":"2.0","id":1,"method":"messages/send","params":{"message":{"kind":"message","role":"user","messageId":"1","contextId":"ctx-1","parts":[{"kind":"text","text":"Hello"}]}}}'
```

## Testing

```bash
bun run test             # Run all tests (unit + integration)
bun run test:unit        # Unit tests only
bun run test:integration # Integration tests only
bun run test:watch       # Watch mode
bun run test:coverage    # With coverage
```

## License

MIT