The Problem
Direct HTTP requests to an agent have a fundamental scaling problem: the HTTP connection stays open while the agent works. For tasks that take 30 seconds, 2 minutes, or longer, this breaks down completely. Load balancers time out (typically 60 seconds). Mobile connections drop. The server can only handle as many concurrent users as it has threads.
Queue-based architecture solves this. Users submit tasks and get a job_id immediately. Workers pull tasks from the queue and process them independently. Users poll for status or receive a webhook when their job completes.
Queue-based architecture
The system has four components:
Queue (Redis): Stores pending tasks as serialized payloads. Tasks wait here until a worker picks them up. Redis supports atomic operations, so two workers can never accidentally pick the same task.
Producer (API server): Accepts task submissions from users, validates them, enqueues them, and returns a job_id immediately with HTTP 202 Accepted. The producer doesn't run the agent — it just queues work.
Workers (N agent processes): Each worker process independently loops: pull task from queue → run agent → store result → pull next task. Workers are stateless — you can add or remove them without coordination.
Result Store (Redis or database): Maps job_id → {status, result, timestamps}. The status API reads from here.
A user submits a 3-minute agent task. What's the correct API design?
HTTP connections should not be held open for minutes — load balancers time out (typically 60s), mobile connections drop, and the UX is just a spinning UI. The async job pattern: accept immediately (202), process in background, notify or let the user poll. This is the standard pattern for any operation that takes more than 5 seconds.
Job lifecycle and status tracking
Every job transitions through states:
queued → running → complete
→ failed
Track timestamps at each transition:
queued_at: when the API accepted the taskstarted_at: when a worker picked it upcompleted_at: when the agent finished
This data is invaluable for monitoring: the difference between queued_at and started_at is queue latency. High queue latency means you need more workers. The difference between started_at and completed_at is processing time — useful for capacity planning.
For long-running tasks (>30s), polling is acceptable but webhook notifications are better. Instead of the user checking every few seconds, the API calls a user-provided URL when the job completes. Most enterprise use cases require webhooks.
A dead letter queue (DLQ) in a task queue system is used for:
Without a DLQ, failed tasks are silently discarded. A DLQ captures all tasks that couldn't be processed after N retries — you can inspect them to understand failure patterns, fix the underlying issue, and replay them. It's the difference between "we lost some tasks" and "we have a record of every failure and can recover them."
Horizontal scaling
Adding workers is the scaling mechanism. Each worker process independently polls the Redis queue. Redis handles the concurrency — two workers can't pick the same task because Redis LPOP is atomic.
Scaling pattern:
- 1 worker: handles 1 task at a time → throughput = (1 / avg_task_duration) tasks/minute
- 5 workers: handles 5 tasks simultaneously → throughput scales ~5x
- 20 workers: limited by Redis connection pool and your AI API rate limits
Priority queues allow different task classes to have different queues. A HIGH queue is always drained before NORMAL, which drains before LOW. Use cases: paid users get HIGH priority, free users get NORMAL, background jobs get LOW.
Build
Build it
Before wiring up Redis and RQ, this in-memory simulation shows the core dynamic — a priority queue drained by a small worker pool, with retries pushing failed tasks to a dead letter queue.
import collections
import random
random.seed(7)
HIGH, NORMAL, LOW = "high", "normal", "low"
queues = {HIGH: collections.deque(), NORMAL: collections.deque(), LOW: collections.deque()}
dead_letter = []
MAX_RETRIES = 2
# Enqueue sample tasks with priorities and a simulated failure chance
for i in range(8):
priority = random.choice([HIGH, NORMAL, NORMAL, LOW])
queues[priority].append({"id": f"task-{i}", "priority": priority, "attempts": 0, "fail_chance": random.random()})
def next_task():
for level in (HIGH, NORMAL, LOW):
if queues[level]:
return queues[level].popleft()
return None
def run_worker(worker_id):
task = next_task()
if task is None:
return
task["attempts"] += 1
if task["fail_chance"] < 0.3 and task["attempts"] <= MAX_RETRIES:
print(f"worker-{worker_id}: {task['id']} FAILED (attempt {task['attempts']}), re-queued")
queues[task["priority"]].append(task)
elif task["fail_chance"] < 0.3:
dead_letter.append(task)
print(f"worker-{worker_id}: {task['id']} FAILED permanently -> dead letter queue")
else:
print(f"worker-{worker_id}: {task['id']} ({task['priority']}) completed")
NUM_WORKERS = 2
while any(queues.values()):
for w in range(NUM_WORKERS):
run_worker(w)
print(f"\nDead letter queue: {[t['id'] for t in dead_letter]}")
The code implements a complete queue-based agent API with FastAPI, Redis, RQ, priority queues, and dead letter queue handling.
Setup:
# Install dependencies
pip install rq redis fastapi uvicorn anthropic
# Start Redis
docker run -p 6379:6379 redis:alpine
# Start workers (in a separate terminal)
rq worker high normal low
# Run the API
python lesson_08_02_queue_dispatch.py
Then test the API:
# Submit a task
curl -X POST http://localhost:8001/tasks \
-H "Content-Type: application/json" \
-d '{"query": "Explain async programming", "user_id": "user-001", "priority": "normal"}'
# → {"job_id": "abc-123", "status": "queued", ...}
# Poll status
curl http://localhost:8001/jobs/abc-123/status
# → {"status": "running", "started_at": "..."}
# Get result
curl http://localhost:8001/jobs/abc-123/result
# → {"result": {"answer": "..."}}
# Check queue stats
curl http://localhost:8001/queue/stats
The full implementation is in lesson_08_02_queue_dispatch.py (under Code artifacts at the end of this lesson).