Async Workflows and Background Jobs
Long-running agents need queues (Celery, RQ, Temporal) — return a job id, poll for status.
Async Workflows and Background Jobs is a free AI Agents lesson on CoddyKit — lesson 2 of 4. You can read the complete lesson below for free — then practise it hands-on in the browser with a built-in code editor and a 24/7 AI tutor. It is part of the AI Agents learning path, one of 4 lessons in the course, and your progress syncs across the web and the CoddyKit app.
When the Request Cycle Is Too Slow
Some agent tasks take 30s, 5min, or hours. You cannot keep the HTTP connection open. Move them to async background jobs.
The Pattern
- Client POSTs the task -> server creates job, returns job_id
- Worker picks up the job from a queue
- Client polls GET /jobs/{id} for status
- When done, server returns the result
Submit Endpoint
from uuid import uuid4
@app.post('/jobs')
def submit(req: AgentRequest):
job_id = str(uuid4())
redis.hset(f'job:{job_id}', mapping={'status': 'queued', 'user_id': req.user_id})
queue.enqueue('run_agent_job', job_id, req.query)
return {'job_id': job_id, 'status': 'queued'}Status Endpoint
@app.get('/jobs/{job_id}')
def status(job_id: str):
data = redis.hgetall(f'job:{job_id}')
if not data:
raise HTTPException(404)
return {
'status': data['status'],
'result': data.get('result'),
'error': data.get('error')
}Worker
A separate process consumes the queue:
def run_agent_job(job_id, query):
redis.hset(f'job:{job_id}', 'status', 'running')
try:
result = run_agent(query)
redis.hset(f'job:{job_id}', mapping={'status': 'done', 'result': result})
except Exception as e:
redis.hset(f'job:{job_id}', mapping={'status': 'failed', 'error': str(e)})Queue Choices
- RQ (Redis Queue) — minimalist Python
- Celery — Python classic, lots of features
- Temporal — durable workflows, retries, observability
- Dramatiq — Celery alternative
- Cloud-native — Cloud Tasks, SQS, Pub/Sub
Temporal for Durable Workflows
Agent workflows are stateful. Temporal's durable execution model is a great fit:
import temporalio
@temporalio.workflow.defn
class AgentWorkflow:
@temporalio.workflow.run
async def run(self, query: str) -> str:
plan = await workflow.execute_activity(plan_step, query, schedule_to_close_timeout=timedelta(minutes=2))
results = await workflow.execute_activity(execute_step, plan, schedule_to_close_timeout=timedelta(minutes=10))
return await workflow.execute_activity(synthesise_step, results)Streaming Partial Updates
Long-running jobs benefit from progress updates. Use Server-Sent Events or WebSockets:
@app.get('/jobs/{job_id}/stream')
def stream_progress(job_id):
def gen():
while True:
update = redis.brpop(f'updates:{job_id}', timeout=30)
if not update:
yield 'data: {"status": "timeout"}\n\n'
break
yield f'data: {update[1].decode()}\n\n'
if 'done' in update[1].decode():
break
return StreamingResponse(gen(), media_type='text/event-stream')Job TTLs
Don't keep job records forever:
redis.expire(f'job:{job_id}', 86400) # 1 dayRetries
Transient failures retry automatically; permanent failures go to a DLQ for human review:
import time
def retry(retries=3, retry_backoff=True):
def decorator(func):
def wrapper(*args, **kwargs):
for attempt in range(1, retries + 1):
try:
return func(*args, **kwargs)
except Exception as e:
print(f'attempt {attempt} failed: {e}')
if attempt == retries:
raise
return None
return wrapper
return decorator
attempts = {'n': 0}
@retry(retries=3, retry_backoff=True)
def run_agent_job(job_id):
attempts['n'] += 1
if attempts['n'] < 3:
raise RuntimeError('transient error')
return f'job {job_id} done'
print(run_agent_job('job-1'))
Concurrency Limits
To control GPU/API costs, limit concurrent workers:
rq worker --burst --max-jobs 1000 --queue agent
# Or per-queue concurrency in Temporal worker config.Per-User Quotas
Track in-flight jobs per user:
key = f'inflight:{user_id}'
if redis.scard(key) >= 5:
raise HTTPException(429, 'Too many in-flight jobs')
redis.sadd(key, job_id)Observability
Trace each job end-to-end. Tag spans with job_id and user_id. Failed jobs should automatically open a ticket or alert.
Status Polling Pattern
Why use the submit/status-polling pattern for long-running agents?
Recap
POST job, return id, worker processes from queue, client polls. RQ/Celery for simple; Temporal for stateful workflows. TTLs, retries, quotas, traces.
Frequently asked questions
Is the “Async Workflows and Background Jobs” lesson free?
Yes — the full text of “Async Workflows and Background Jobs” is free to read here on the web, and the AI Agents course includes 4 lessons in total. To practise it interactively (a built-in code editor and a 24/7 AI tutor) and unlock the rest of the AI Agents course, upgrade to CoddyKit PRO.
What will I learn in “Async Workflows and Background Jobs”?
Long-running agents need queues (Celery, RQ, Temporal) — return a job id, poll for status. You practise AI Agents with hands-on code you run directly in the browser, and a 24/7 AI tutor answers your questions as you work through the lesson.
Do I need any experience to start AI Agents?
No prior experience is required. AI Agents on CoddyKit is structured for beginners through advanced learners; this is — lesson 2 of 4, so you can start here or from the beginning and move at your own pace.
How long does the “Async Workflows and Background Jobs” lesson take?
Most CoddyKit lessons take about 5–10 minutes. Each one is bite-sized and interactive, so you make steady progress and pick up exactly where you left off across the web and the app.
Can I write and run code in this AI Agents lesson?
Yes. Every AI Agents lesson includes a built-in code editor, so you write and run real code right in your browser and get instant AI feedback — no local setup required.
All lessons in this course
- Serving Agents Behind an API
- Async Workflows and Background Jobs
- Rate Limiting and Quota Management
- Blue-Green and Canary Deploys for Agents