Agentuity Documentation

Use queues when a request should hand work off and return quickly. The handler accepts the request, the publish is durable, and a separate consumer or destination handles the slow part. Start with QueueClient; Hono apps can use c.var.queue after installing the Agentuity middleware.

npm install @agentuity/queue

import { QueueClient } from '@agentuity/queue';
 
interface ReportRequest {
  readonly userId: string;
  readonly report: 'daily' | 'weekly';
}
 
const queue = new QueueClient();
 
await queue.createQueue('email-reports', {
  description: 'Reports generated outside the request path',
  settings: {
    defaultTtlSeconds: 60 * 60 * 24,
    defaultMaxRetries: 3,
  },
});
 
export async function queueReport(userId: string): Promise<string> {
  const result = await queue.publish(
    'email-reports',
    { userId, report: 'daily' } satisfies ReportRequest,
    {
      partitionKey: userId,
      idempotencyKey: `daily-report:${userId}`,
    }
  );
 
  return result.id;
}

QueueClient reads AGENTUITY_SDK_KEY, then AGENTUITY_CLI_KEY, from the environment. Keep that key in .env for local development and configure the same variable for deployed apps.

When to use queues

Need	Use
background jobs or async handoff	Queues
recurring delivery on a timer	Schedules
managed ingest from external services	Webhooks
append-only ordered data with replay	Durable Streams
exact key lookup or counters	Key-Value Storage

Client Setup

Construct the client once at module scope and reuse it from handlers, routes, or scripts.

import { QueueClient } from '@agentuity/queue';
 
const queue = new QueueClient({
  orgId: process.env.AGENTUITY_CLOUD_ORG_ID,
});

Option	Description
`apiKey`	Optional API key. Defaults to `AGENTUITY_SDK_KEY`, then `AGENTUITY_CLI_KEY`.
`orgId`	Optional organization ID. Used when the API key is org-scoped or when calling from a CLI context.
`url`	Optional Queue API URL. Defaults to `AGENTUITY_QUEUE_URL`, then the regional Agentuity service URL.
`logger`	Optional logger instance.

Create a Queue

createQueue() is idempotent and caches the queue name in memory. Calling it again with the same name from the same client is a no-op.

await queue.createQueue('order-events', {
  queueType: 'worker',
  description: 'Orders waiting for fulfillment',
  settings: {
    defaultVisibilityTimeoutSeconds: 60,
    defaultMaxRetries: 5,
    maxInFlightPerClient: 10,
    retentionSeconds: 60 * 60 * 24 * 30,
  },
});

Field	Description
`queueType`	Optional. `worker` delivers each message to one consumer. `pubsub` broadcasts to all subscribers. Defaults to `worker`.
`description`	Optional human-readable purpose.
`settings.defaultTtlSeconds`	Optional message expiration in seconds. Pass `null` for no expiration.
`settings.defaultVisibilityTimeoutSeconds`	Optional invisibility window after a worker receives a message, before redelivery.
`settings.defaultMaxRetries`	Optional retry limit before a failed message moves to the dead letter queue.
`settings.maxInFlightPerClient`	Optional concurrent message limit per consumer.
`settings.retentionSeconds`	Optional retention period for acknowledged messages.

Queue names must match ^[a-z_][a-z0-9_-]*$ and be 256 characters or fewer. The client validates this before sending the request.

Publish Messages

Payloads can be strings or any JSON-serializable object. Objects are stringified before sending.

const result = await queue.publish(
  'order-events',
  {
    orderId: 'ord_123',
    event: 'paid',
  },
  {
    metadata: { source: 'checkout' },
    partitionKey: 'ord_123',
    idempotencyKey: 'order-paid:ord_123',
    ttl: 60 * 60,
  }
);

Option	Description
`metadata`	Optional JSON metadata for routing or inspection.
`partitionKey`	Optional key for ordering related messages. Max 256 characters.
`idempotencyKey`	Optional dedup key for retrying publishers. Max 256 characters.
`ttl`	Optional time-to-live in seconds.
`projectId`	Optional project ID for cross-project publishing.
`agentId`	Optional agent ID for attribution.
`sync`	Optional. When `true`, the call waits until the message is persisted before returning.

publish() returns the assigned message ID, queue offset, and publish timestamp:

interface QueuePublishResult {
  readonly id: string;
  readonly offset: number;
  readonly publishedAt: string;
}

Use sync: true when the caller must know the message is durable before continuing, for example before responding to a webhook that will be retried on non-2xx.

await queue.publish('billing-events', { invoiceId: 'inv_123' }, { sync: true });

Payloads are limited to 1 MB. Publishing an empty payload throws QueueValidationError.

Use idempotency keys on retryable publishers

If the caller may retry the same request, set idempotencyKey from the upstream event ID or business key. Use partitionKey separately when messages for the same entity should stay ordered.

Processing Lifecycle

After a message is published, a worker or destination receives it. A received message is hidden from other workers for defaultVisibilityTimeoutSeconds. A successful worker acknowledges the message and it becomes delivered. A failed worker negatively acknowledges it, or lets the visibility timeout expire, so the queue can retry it. After defaultMaxRetries is exhausted, the message moves to the dead letter queue.

Step	State	Result
publish	pending	message is durable and available to process
receive	leased	one worker has the message until ack, nack, or visibility timeout
ack	delivered	message is complete and will not be delivered again
nack or timeout	pending or failed	message can be retried until the retry limit
retry limit reached	dead	message is held in the dead letter queue for inspection, replay, purge, or delete
replay from dead letter queue	pending	message returns to the normal queue flow

@agentuity/queue currently covers queue creation, publishing, and deletion. Consumer receive, ack, nack, destination delivery logs, and dead letter queue operations are available through the Queue API reference. Use those lower-level endpoints when you are writing a custom worker, inspecting failed deliveries, or replaying dead letter messages.

Receive Message: lock the next pending message
Ack Message: mark a message complete
Nack Message: return a message for retry
Queue Destinations: manage HTTP destinations and delivery logs
Dead Letter Queue: list, replay, purge, or delete failed messages

Receive a Webhook and Hand It Off

Most apps publish from inside a framework route. The route validates the payload, hands it to a queue, and returns 2xx so the upstream sender does not retry.

typescriptsrc/routes/stripe.ts

import { Hono } from 'hono';
import { agentuity } from '@agentuity/hono';
import type { Services } from '@agentuity/hono';
 
type Variables = Pick<Services, 'queue'>;
 
const app = new Hono<{ Variables: Variables }>();
 
app.use('*', agentuity());
 
app.post('/stripe/events', async (c) => {
  const eventId = c.req.header('stripe-event-id');
  if (!eventId) {
    return c.json({ error: 'Missing stripe-event-id header' }, 400);
  }
 
  const payload = await c.req.text();
 
  await c.var.queue.publish('stripe-events', payload, {
    idempotencyKey: eventId,
  });
 
  return c.json({ received: true });
});
 
export default app;

The same shape works in Next.js route handlers, TanStack Start server functions, or plain scripts: build the client at module scope, validate, publish, return.

Delete a Queue

await queue.deleteQueue('old-order-events');

deleteQueue() throws QueueNotFoundError if the queue is already gone, which lets a manual cleanup script tell "deleted" from "missing".

Deleting a queue removes its messages

deleteQueue() permanently removes the queue and any pending, in-flight, or dead-lettered messages. It cannot be undone.

Handle Errors

The package exports typed errors for the common failure cases.

import {
  QueueClient,
  QueueNotFoundError,
  QueueValidationError,
  QueuePublishError,
} from '@agentuity/queue';
 
const queue = new QueueClient();
 
export async function publishNotification(userId: string): Promise<boolean> {
  try {
    await queue.publish('notifications', { userId });
    return true;
  } catch (error) {
    if (error instanceof QueueNotFoundError) {
      return false;
    }
 
    if (error instanceof QueueValidationError) {
      // bad name, oversized payload, partition/idempotency key too long
      throw error;
    }
 
    if (error instanceof QueuePublishError) {
      // unexpected response shape from the queue API
      throw error;
    }
 
    throw error;
  }
}

QueueValidationError carries field and optional value so the caller can surface a useful message without parsing the error string.

Hono Middleware Variant

@agentuity/hono constructs QueueClient once and exposes it on c.var.queue. Use it when several routes share the same client and you do not want to manage instances by hand.

npm install @agentuity/hono hono

import { Hono } from 'hono';
import { agentuity } from '@agentuity/hono';
import type { Services } from '@agentuity/hono';
 
type Variables = Pick<Services, 'queue'>;
 
const app = new Hono<{ Variables: Variables }>();
 
app.use('*', agentuity());
 
app.post('/jobs', async (c) => {
  const body = await c.req.json<{ userId: string }>();
 
  const result = await c.var.queue.publish('email-reports', {
    userId: body.userId,
    report: 'daily',
  });
 
  return c.json({ messageId: result.id }, 202);
});
 
export default app;

Next Steps

Webhooks: create managed ingest URLs that forward payloads to your app
Schedules: run recurring work on a cron expression
Tasks: track work items with status, comments, and audit history
Using Standalone Packages: configure service clients outside Agentuity projects