Anthropic queue

A queue pre-configured for Anthropic API workloads, plus a reference worker. The template encodes everything universally true about a Claude workload — ack mode, long ackTimeout, retry budget that doesn't get burned by rate limits — so you don't have to choose queue config. The worker translates Anthropic's error signals (Retry-After, 529 overload) into SimpleQ's ack / nack / defer callbacks. Together: rate-aware Claude jobs in three POST requests.

Prerequisite

This example uses an Anthropic API key and a SimpleQ API key (create one on the dashboard's API Keys page).

Create the queue

cURL

curl -X POST "$SQ_API_URL/v1/queues" \
  -H "Authorization: Bearer $SIMPLEQ_API_KEY" \
  -H "content-type: application/json" \
  -d '{
    "template": "anthropic",
    "name": "claude",
    "webhookUrl": "https://your-worker.example.com/webhook"
  }'

Node

const res = await fetch(`${process.env.SQ_API_URL}/v1/queues`, {
  method: 'POST',
  headers: { 'content-type': 'application/json', authorization: `Bearer ${process.env.SIMPLEQ_API_KEY}` },
  body: JSON.stringify({
    template: 'anthropic',
    name: 'claude',
    webhookUrl: 'https://your-worker.example.com/webhook',
  }),
});
const queue = await res.json();

Python

import os, httpx

res = httpx.post(
    f"{os.environ['SQ_API_URL']}/v1/queues",
    headers={"authorization": f"Bearer {os.environ['SIMPLEQ_API_KEY']}"},
    json={
        "template": "anthropic",
        "name": "claude",
        "webhookUrl": "https://your-worker.example.com/webhook",
    },
)
queue = res.json()

Three fields. template: "anthropic" applies the Anthropic defaults; name and webhookUrl are yours. Override any default by passing it alongside template (e.g. "concurrency": 50 to raise from the default of 20).

Developing locally? webhookUrl is your tunnel's https://…/webhook — see Local development.

What this template expands to

Field	Value	Why
mode	"ack"	Claude calls routinely exceed the 15s webhook ceiling. Ack mode lets the worker return 200 immediately and report the real outcome via /ack, /nack, or /defer.
maxAttempts	4	Budgets only genuine failures. 429/503/529 are deferred and don't count against this.
concurrency	20	How many in-flight Claude calls at once. Throughput ≈ concurrency ÷ average call duration in seconds. Raise to saturate a high tier.
ackTimeout	600 (10 min)	Sized above a long Claude generation (extended thinking + large output).
ackTimeoutAction	"retry"	If the worker dies silently, redeliver.
backoffType / backoffDelay	"exponential" / 2s	Spacing for genuine transient failures (nack with retryable: true).
dlqEnabled	true	Exhausted retries or hard nacks land in the DLQ for inspection.

Rate limiting is deliberately not set — that's account-tier-specific. defer on 429/503/529 handles real backpressure. If you want a proactive guard, see Add a proactive RPM guard below.

Publish a job

The payload is passed straight to anthropic.messages.create() by the worker, so it's just an Anthropic Messages body:

cURL

curl -X POST "$SQ_API_URL/v1/queues/claude/jobs" \
  -H "Authorization: Bearer $SIMPLEQ_API_KEY" \
  -H "content-type: application/json" \
  -d '{
    "idempotencyKey": "summarize-doc-4815",
    "payload": {
      "model": "claude-sonnet-4-6",
      "max_tokens": 1024,
      "messages": [{ "role": "user", "content": "Summarize: ..." }]
    }
  }'

Node

import { SimpleQ } from '@simpleq/sdk';

const simpleq = new SimpleQ({ apiKey: process.env.SIMPLEQ_API_KEY });

const job = await simpleq.publish('claude', {
  idempotencyKey: 'summarize-doc-4815',
  payload: {
    model: 'claude-sonnet-4-6',
    max_tokens: 1024,
    messages: [{ role: 'user', content: 'Summarize: ...' }],
  },
});

Python

import os, httpx

res = httpx.post(
    f"{os.environ['SQ_API_URL']}/v1/queues/claude/jobs",
    headers={"authorization": f"Bearer {os.environ['SIMPLEQ_API_KEY']}"},
    json={
        "idempotencyKey": "summarize-doc-4815",
        "payload": {
            "model": "claude-sonnet-4-6",
            "max_tokens": 1024,
            "messages": [{"role": "user", "content": "Summarize: ..."}],
        },
    },
)
job = res.json()

idempotencyKey dedupes the publish call — a double-publish returns the existing job. See Idempotency for the broader pattern.

The worker

Same shape as the generic ack worker, specialized to Anthropic's error semantics. Pick the tab for your framework.

Express

// worker.mjs
import express from 'express';
import { SimpleQ, retryAfterSeconds } from '@simpleq/sdk';
import { simpleqWebhookHandler } from '@simpleq/sdk/express';
import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic({ maxRetries: 0 });
const simpleq = new SimpleQ({ apiKey: process.env.SIMPLEQ_API_KEY });
const app = express();

app.post(
  '/webhook',
  // Verifies x-simpleq-signature against the raw body (401 on mismatch). Ack mode:
  // send the 200 yourself, then run the work out of band.
  simpleqWebhookHandler(process.env.SQ_SIGNING_SECRET, async (job, { res }) => {
    res.status(200).end(); // ack mode: respond now, do the work after

    try {
      const message = await anthropic.messages.create(job.payload);
      console.log(`[${job.id}] completed: ${message.usage?.output_tokens} output tokens`);
      await simpleq.ack(job.id);
    } catch (err) {
      const status = err?.status;
      if (status === 429 || status === 503 || status === 529) {
        // 10s fallback covers responses without a Retry-After (e.g. 529 overloaded).
        await simpleq.defer(job.id, { retryAfter: retryAfterSeconds(err) ?? 10, reason: `anthropic ${status}` });
      } else if (status >= 400 && status < 500) {
        await simpleq.nack(job.id, { retryable: false, reason: `anthropic ${status}` });
      } else {
        await simpleq.nack(job.id, { retryable: true, reason: `anthropic ${status ?? 'network'}` });
      }
    }
  }),
);

app.listen(9000, () => console.log('anthropic ack worker on :9000'));

Nest.js

// main.ts — a self-contained Nest worker
import { Controller, HttpCode, Module, Post, UseGuards } from '@nestjs/common';
import { NestFactory } from '@nestjs/core';
import { SimpleQ, type WebhookPayload, retryAfterSeconds } from '@simpleq/sdk';
import { SimpleQModule, SimpleQSignatureGuard, SimpleQJob } from '@simpleq/sdk/nest';
import Anthropic from '@anthropic-ai/sdk';

const anthropic = new Anthropic({ maxRetries: 0 });
const simpleq = new SimpleQ({ apiKey: process.env.SIMPLEQ_API_KEY, baseUrl: process.env.SQ_API_URL });

@Controller('webhook')
class WebhookController {
  // Ack mode: the guard verifies the signature, we return 200 immediately, then
  // report the real outcome out of band. ackTimeout redelivers if we crash mid-job.
  @Post()
  @HttpCode(200)
  @UseGuards(SimpleQSignatureGuard)
  handle(@SimpleQJob() job: WebhookPayload) {
    void process(job);
  }
}

async function process(job: WebhookPayload) {
  try {
    const message = await anthropic.messages.create(job.payload as Anthropic.MessageCreateParamsNonStreaming);
    console.log(`[${job.id}] completed: ${message.usage?.output_tokens} output tokens`);
    await simpleq.ack(job.id);
  } catch (err: any) {
    const status = err?.status;
    if (status === 429 || status === 503 || status === 529) {
      // 10s fallback covers responses without a Retry-After (e.g. 529 overloaded).
      await simpleq.defer(job.id, { retryAfter: retryAfterSeconds(err) ?? 10, reason: `anthropic ${status}` });
    } else if (status >= 400 && status < 500) {
      await simpleq.nack(job.id, { retryable: false, reason: `anthropic ${status}` });
    } else {
      await simpleq.nack(job.id, { retryable: true, reason: `anthropic ${status ?? 'network'}` });
    }
  }
}

@Module({
  imports: [SimpleQModule.forRoot({ signingSecret: process.env.SQ_SIGNING_SECRET! })],
  controllers: [WebhookController],
})
class AppModule {}

// rawBody: true lets the guard verify the signature against the exact bytes received.
const app = await NestFactory.create(AppModule, { rawBody: true });
await app.listen(9000);

Python (FastAPI)

# worker.py
import hashlib
import hmac
import json
import os

import httpx
from anthropic import APIStatusError, AsyncAnthropic
from fastapi import BackgroundTasks, FastAPI, Header, Request, Response

anthropic = AsyncAnthropic(max_retries=0)
SQ_API_URL = os.environ["SQ_API_URL"]
SIMPLEQ_API_KEY = os.environ["SIMPLEQ_API_KEY"]
SQ_SIGNING_SECRET = os.environ["SQ_SIGNING_SECRET"]
app = FastAPI()


def verify_signature(raw_body: bytes, header: str | None) -> bool:
    if not header:
        return False
    expected = "sha256=" + hmac.new(SQ_SIGNING_SECRET.encode(), raw_body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header)


def parse_retry_after(value: str | None, fallback: int = 10) -> int:
    try:
        return int(value) or fallback
    except (TypeError, ValueError):
        return fallback


# Reuse one client across callbacks for connection pooling.
_client = httpx.AsyncClient(timeout=10.0)


async def callback(job_id: str, kind: str, body: dict | None = None) -> None:
    await _client.post(
        f"{SQ_API_URL}/v1/jobs/{job_id}/{kind}",
        headers={"content-type": "application/json", "authorization": f"Bearer {SIMPLEQ_API_KEY}"},
        json=body or {},
    )


async def process(job: dict) -> None:
    try:
        # AsyncAnthropic so a long generation never blocks the event loop.
        message = await anthropic.messages.create(**job["payload"])
        print(f"[{job['id']}] completed: {message.usage.output_tokens} output tokens")
        await callback(job["id"], "ack")
    except APIStatusError as err:
        status = err.status_code
        if status in (429, 503, 529):
            # 529 (overloaded) carries no Retry-After → parse_retry_after falls back to 10s.
            retry_after = parse_retry_after(err.response.headers.get("retry-after"))
            await callback(job["id"], "defer", {"retryAfter": retry_after, "reason": f"anthropic {status}"})
        elif 400 <= status < 500:
            await callback(job["id"], "nack", {"retryable": False, "reason": f"anthropic {status}"})
        else:
            await callback(job["id"], "nack", {"retryable": True, "reason": f"anthropic {status}"})
    except Exception:
        await callback(job["id"], "nack", {"retryable": True, "reason": "anthropic network"})


@app.post("/webhook")
async def webhook(
    request: Request,
    background_tasks: BackgroundTasks,
    x_simpleq_signature: str | None = Header(default=None),
):
    # Verify over the raw bytes BEFORE parsing.
    raw = await request.body()
    if not verify_signature(raw, x_simpleq_signature):
        return Response(status_code=401)

    job = json.loads(raw)
    # BackgroundTasks runs in-process — if this worker crashes mid-job the
    # work is lost, but ackTimeout + ackTimeoutAction: "retry" redelivers it.
    background_tasks.add_task(process, job)
    return Response(status_code=200)


# uvicorn worker:app --port 9000

Run it

Node

npm init -y
npm i express @simpleq/sdk @anthropic-ai/sdk
ANTHROPIC_API_KEY=sk-ant-... \
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
node worker.mjs

Python

pip install fastapi uvicorn httpx anthropic
ANTHROPIC_API_KEY=sk-ant-... \
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
uvicorn worker:app --port 9000

Swap express (or FastAPI) for your framework of choice. The Anthropic SDK is the same across all tabs.

• ANTHROPIC_API_KEY — from console.anthropic.com.
• SQ_API_URL — the SimpleQ API base your worker calls back to: https://api.simpleq.io.
• SIMPLEQ_API_KEY — create one on the dashboard's API Keys page.
• SQ_SIGNING_SECRET — your queue's signing secret. See Signature verification for where to find it.

Your worker listens on port 9000, but SimpleQ delivers from the cloud and can't reach localhost. Open a tunnel and point the queue at it:

cloudflared tunnel --url http://localhost:9000

Set the printed https://… URL (with /webhook appended) as your queue's webhookUrl. See Local development for the full flow and updating the URL on restart.

Error routing

Anthropic response	Worker calls	Effect
Success	/ack	Job → completed.
429 / 503 / 529 (rate-limited, unavailable, overloaded)	/defer, relaying Retry-After (10s fallback when absent, e.g. 529)	Held and redelivered; no attempt burned.
4xx (bad request, auth, not found)	/nack {retryable: false}	Job → dead, lands in DLQ.
Non-529 5xx, network, SDK timeout	/nack {retryable: true}	Retried with backoff; counts against maxAttempts.

Three gotchas worth internalizing

1. Disable the SDK's own retries (maxRetries: 0). The Anthropic SDK retries 429 and ≥500 twice by default with its own backoff. That swallows the error, you never see the Retry-After, and you can't relay it to /defer. Worse, the SDK's retries and SimpleQ's redelivery fight each other (the classic nested-retry deadlock). With retries off, the 429 surfaces in your catch, you read its Retry-After, and SimpleQ owns the backoff. The example sets this; don't undo it.

2. 529 never burns an attempt — by design, with a tail. A 529 is Anthropic being overloaded: not your fault, never burns a retry attempt, no trustworthy Retry-After. The worker defers it, so it costs nothing against maxAttempts. The tradeoff: during a sustained Anthropic outage the job waits politely rather than dying. If you want a hard ceiling, count 529s in the worker and switch to nack after N — or fail over to the same model on Bedrock / Vertex, which run on separate capacity.

3. The template is workload-shaped, not account-shaped. The defaults encode facts true for any Anthropic workload (ack mode, long ackTimeout, retry budget). They don't encode your account's RPM tier — that's account-shaped, not workload-shaped, and tuning is below.

Add a proactive RPM guard

If you know your Anthropic tier and want SimpleQ to space out deliveries below your RPM ceiling, set rateLimitMax to ~80% of your RPM and rateLimitWindow to 60:

curl -X PUT "$SQ_API_URL/v1/queues/<queue-id>" \
  -H "Authorization: Bearer $SIMPLEQ_API_KEY" \
  -H "content-type: application/json" \
  -d '{ "rateLimitMax": 800, "rateLimitWindow": 60 }'

Two caveats worth knowing: the limiter is per-worker (N workers running = N × the configured rate), and Anthropic's TPM usually binds before RPM — a request counter can't see token cost. The proactive guard shaves the count of 429s; defer is what actually saves you at a real limit.

A rate-limit change takes effect once jobs already being processed finish — see Editing a queue.