OpenAI queue

A queue pre-configured for OpenAI API workloads, plus a reference worker. The template encodes everything universally true about an OpenAI workload — ack mode, 5-minute ackTimeout, retry budget that doesn't get burned by rate limits — so you don't have to choose queue config. The worker translates OpenAI's error signals (Retry-After, 5xx) into SimpleQ's ack / nack / defer callbacks.

Prerequisite

This example uses an OpenAI 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": "openai",
    "name": "openai-jobs",
    "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: 'openai',
    name: 'openai-jobs',
    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": "openai",
        "name": "openai-jobs",
        "webhookUrl": "https://your-worker.example.com/webhook",
    },
)
queue = res.json()

Three fields. template: "openai" applies the OpenAI 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"	OpenAI calls can 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 are deferred and don't count against this.
concurrency	20	How many in-flight OpenAI calls at once. Raise to saturate a high tier.
ackTimeout	300 (5 min)	Sized above a typical OpenAI generation. Raise if using long-context models with large outputs.
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 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 openai.chat.completions.create() by the worker, so it's just an OpenAI Chat Completions body:

cURL

curl -X POST "$SQ_API_URL/v1/queues/openai-jobs/jobs" \
  -H "Authorization: Bearer $SIMPLEQ_API_KEY" \
  -H "content-type: application/json" \
  -d '{
    "idempotencyKey": "summarize-doc-123",
    "payload": {
      "model": "gpt-4o-mini",
      "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('openai-jobs', {
  idempotencyKey: 'summarize-doc-123',
  payload: {
    model: 'gpt-4o-mini',
    messages: [{ role: 'user', content: 'Summarize: ...' }],
  },
});

Python

import os, httpx

res = httpx.post(
    f"{os.environ['SQ_API_URL']}/v1/queues/openai-jobs/jobs",
    headers={"authorization": f"Bearer {os.environ['SIMPLEQ_API_KEY']}"},
    json={
        "idempotencyKey": "summarize-doc-123",
        "payload": {
            "model": "gpt-4o-mini",
            "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 OpenAI'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 OpenAI from 'openai';

const openai = new OpenAI({ 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 openai.chat.completions.create(job.payload);
      console.log(`[${job.id}] completed: ${message.usage?.total_tokens} tokens`);
      await simpleq.ack(job.id);
    } catch (err) {
      const status = err?.status;
      if (status === 429 || status === 503) {
        await simpleq.defer(job.id, { retryAfter: retryAfterSeconds(err) ?? 10, reason: `openai ${status}` });
      } else if (status >= 400 && status < 500) {
        await simpleq.nack(job.id, { retryable: false, reason: `openai ${status}` });
      } else {
        await simpleq.nack(job.id, { retryable: true, reason: `openai ${status ?? 'network'}` });
      }
    }
  }),
);

app.listen(9000, () => console.log('openai 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 OpenAI from 'openai';

const openai = new OpenAI({ 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 openai.chat.completions.create(job.payload as OpenAI.ChatCompletionCreateParamsNonStreaming);
    console.log(`[${job.id}] completed: ${message.usage?.total_tokens} tokens`);
    await simpleq.ack(job.id);
  } catch (err: any) {
    const status = err?.status;
    if (status === 429 || status === 503) {
      await simpleq.defer(job.id, { retryAfter: retryAfterSeconds(err) ?? 10, reason: `openai ${status}` });
    } else if (status >= 400 && status < 500) {
      await simpleq.nack(job.id, { retryable: false, reason: `openai ${status}` });
    } else {
      await simpleq.nack(job.id, { retryable: true, reason: `openai ${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 fastapi import BackgroundTasks, FastAPI, Header, Request, Response
from openai import APIStatusError, AsyncOpenAI, RateLimitError

openai = AsyncOpenAI(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:
        message = await openai.chat.completions.create(**job["payload"])
        total = message.usage.total_tokens if message.usage else None
        print(f"[{job['id']}] completed: {total} tokens")
        await callback(job["id"], "ack")
    except RateLimitError as err:  # 429
        retry_after = parse_retry_after(err.response.headers.get("retry-after"))
        await callback(job["id"], "defer", {"retryAfter": retry_after, "reason": "openai 429"})
    except APIStatusError as err:
        status = err.status_code
        if status == 503:
            retry_after = parse_retry_after(err.response.headers.get("retry-after"))
            await callback(job["id"], "defer", {"retryAfter": retry_after, "reason": "openai 503"})
        elif 400 <= status < 500:
            await callback(job["id"], "nack", {"retryable": False, "reason": f"openai {status}"})
        else:
            await callback(job["id"], "nack", {"retryable": True, "reason": f"openai {status}"})
    except Exception:
        await callback(job["id"], "nack", {"retryable": True, "reason": "openai 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 openai
OPENAI_API_KEY=sk-... \
SQ_API_URL=https://api.simpleq.io \
SIMPLEQ_API_KEY=sq_live_... \
SQ_SIGNING_SECRET=... \
node worker.mjs

Python

pip install fastapi uvicorn httpx openai
OPENAI_API_KEY=sk-... \
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 OpenAI SDK is the same across all tabs.

• OPENAI_API_KEY — from platform.openai.com/api-keys.
• 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

OpenAI response	Worker calls	Effect
Success	/ack	Job completed.
429 / 503 (rate-limited / unavailable)	/defer with Retry-After relayed	Held and redelivered; no attempt burned.
4xx (bad request, auth, not found)	/nack {retryable: false}	Job dead-lettered.
Non-503 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 OpenAI 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. 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. The template is workload-shaped, not account-shaped. The defaults encode facts true for any OpenAI workload (ack mode, 5-min ackTimeout, retry budget). They don't encode your account's RPM/TPM tier — that's account-shaped, not workload-shaped, and tuning is below.

3. OpenAI doesn't have a 529. Unlike Anthropic, OpenAI uses 429 for both rate limiting and overload. The Retry-After header is your signal for both. The worker handles them the same way — defer with the relayed delay.

Add a proactive RPM guard

If you know your OpenAI 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": 4000, "rateLimitWindow": 60 }'

Two caveats: the limiter is per-worker (N workers running = N × the configured rate), and OpenAI's TPM usually binds before RPM for large-context models — 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.