Queueing Agent Jobs with Celery or a Message Broker

Agent runs are multi-second to multi-minute workflows with variable turn counts.

Putting them on a job queue decouples HTTP intake from model and tool work so you can retry, scale workers, and survive process restarts.

This recipe uses Celery + Redis as a concrete broker pattern. The same design applies to RQ, Dramatiq, ARQ, Sidekiq, SQS + workers, or cloud task queues.

Summary

Accept agent work as durable jobs, process them in workers with bounded concurrency, persist status and results, and let clients poll or subscribe instead of holding open request connections through the full loop.

Recipe

1. Define a job payload: job_id, goal, tenant_id, priority, trace_id, optional model_hint.
2. Run a message broker (Redis, RabbitMQ, SQS, etc.) as the source of truth for pending work.
3. Enqueue from the API after auth, validation, and admission checks.
4. Implement a worker task that runs the agent loop and writes status transitions.
5. Cap worker prefetch/concurrency so one process does not open unbounded model streams.
6. Configure retries only for infrastructure failures, not for "model gave a bad answer."
7. Store results outside the broker (DB/object store); treat messages as commands, not archives.
8. Emit metrics: enqueue rate, depth, time-in-queue, run duration, failure reason, cost.

Working Example

Minimal Celery-shaped sketch. Install versions at build (celery, Redis broker). Adjust imports if your stack differs.

# tasks.py - teaching sketch; verify Celery API at build
from celery import Celery
from dataclasses import asdict, dataclass
import time
import uuid
 
app = Celery("agents", broker="redis://localhost:6379/0", backend="redis://localhost:6379/1")
app.conf.worker_prefetch_multiplier = 1
app.conf.task_acks_late = True
 
 
@dataclass
class AgentJob:
    job_id: str
    goal: str
    tenant_id: str = "default"
 
 
def run_agent_loop(goal: str) -> str:
    time.sleep(0.05)  # stand-in for model + tools
    return f"done:{goal[:40]}"
 
 
@app.task(bind=True, max_retries=3, name="agents.run")
def run_agent_job(self, payload: dict) -> dict:
    job = AgentJob(**payload)
    try:
        result = run_agent_loop(job.goal)
        return {"job_id": job.job_id, "status": "done", "result": result}
    except ConnectionError as exc:
        # infrastructure only - not quality failures
        raise self.retry(exc=exc, countdown=2 ** self.request.retries)
 
 
def enqueue_agent(goal: str, tenant_id: str = "default") -> str:
    job = AgentJob(job_id=str(uuid.uuid4()), goal=goal, tenant_id=tenant_id)
    run_agent_job.delay(asdict(job))
    return job.job_id
 
 
if __name__ == "__main__":
    print("enqueued", enqueue_agent("triage support ticket"))

Run a worker in another process (verify flags at build):

celery -A tasks worker --loglevel=INFO --concurrency=2

Deep Dive

Why queues fit agents

Agents are closer to workflows than to CRUD handlers.

Broker choices

Pick durability and ops familiarity first. Agent payloads are usually small; throughput is limited by model RPM before broker limits.

Job payload design

Include:

• Stable job_id (also use as idempotency key).
• tenant_id for quotas and noisy-neighbor isolation.
• trace_id / run_id for observability.
• Priority only if you will honor it end-to-end.
• Version of prompt/tool schema if workers can be mixed during deploys.

Exclude:

• Huge context blobs (store in object storage; pass a pointer).
• Secrets (workers load from a secret manager).
• Mutable "live conversation" state without a single writer.

Ack, visibility, and long runs

Long agent jobs interact badly with default timeouts:

• Late ack (ack after success) avoids losing work on crash, but can redeliver.
• Visibility timeout on SQS-style systems must exceed worst-case run or you need heartbeat extension.
• Idempotent tools matter because at-least-once delivery will double-run side effects.

Design tools as if every job may run twice.

Retries: split the classes

Stacking job retries on top of tool retries and model re-turns creates storms. Share budgets with Concurrency Limits and Backpressure for Agent Workers and reliability breakers.

Status API shape

Typical states: queued → running → done | failed | cancelled.

Expose:

• job_id, status, timestamps
• stop_reason when failed/cancelled
• Result pointer or inline small result
• Optional progress events (turn count) via pub/sub or SSE on a side channel

Multi-queue patterns

• Fast lane: short FAQs, cached answers, cheap models
• Default lane: normal agents
• Heavy lane: browser, code sandboxes, multi-agent graphs

Separate queues prevent one heavy class from starving interactive work. Workers can subscribe with different concurrency.

Gotchas

1. Prefetch > 1 with long jobs. Workers reserve many jobs then sit busy; set prefetch low for agents.
2. Using the broker as the result archive. Messages expire; persist results properly.
3. Retrying non-idempotent side effects. Duplicate emails and double charges.
4. Visibility timeout shorter than the agent. Another worker steals an in-flight job.
5. No admission control. Infinite queue depth becomes infinite user wait.
6. Passing full chat history in every message. Payload bloat and PII sprawl; store and reference.
7. Same concurrency for all job types. Browser jobs and cheap classifiers need different caps.

Alternatives

Many products combine a durable job system for work with websockets only for streaming tokens of an already-accepted job.

FAQs

Related

• Scaling Agent Systems Basics - in-process queue intro
• What Actually Bottlenecks an Agent System at Scale - why decoupling matters
• Concurrency Limits and Backpressure for Agent Workers - caps after enqueue
• Architecture Guide: A Horizontally Scalable Agent Worker Pool - full pool design
• Scaling Agent Systems Best Practices - checklist
• Timeout Strategies for Long-Running Agent Steps - time bounds inside jobs
• Long-Running Agents on a VPS: Process Management and Restarts - host process care

Stack versions: Pins from the category manifest (verify at build): OpenRouter (~315+ models, July 2026 pricing/fees); LangGraph 1.0+; CrewAI 1.14+; Microsoft Agent Framework 1.0; Vercel AI SDK 6; Pydantic AI (latest); LlamaIndex (latest); OpenAI Agents SDK (latest + MCP); MCP (Linux Foundation governance); A2A (HTTP+SSE+JSON-RPC 2.0); Solana @solana/web3.js + @solana/spl-token.