async-api-design

name	async-api-design
description	Design asynchronous API patterns for long-running operations, webhooks, and event-driven integrations. Outputs polling endpoints, webhook delivery, callback patterns, and async job management.
argument-hint	["operation duration","client types","reliability requirements","event volume"]
allowed-tools	Read, Write

Async API Design

Synchronous APIs work when operations complete in under a few seconds. For longer operations — AI inference, video processing, bulk imports — async APIs decouple request acceptance from result delivery. The three patterns are polling, webhooks, and server-sent events.

Pattern 1: Job-Based Polling (Most Reliable)

from fastapi import FastAPI, BackgroundTasks
from pydantic import BaseModel
from enum import Enum
import uuid
from datetime import datetime

app = FastAPI()

class JobStatus(str, Enum):
    PENDING    = "pending"
    PROCESSING = "processing"
    COMPLETED  = 
    FAILED     = 



  ():
    job_id = (uuid.uuid4())

     jobs_store.create({
        : job_id,
        : JobStatus.PENDING,
        : datetime.utcnow().isoformat(),
        : request.(),
    })

    background_tasks.add_task(run_export_job, job_id, request)

     {
        : job_id,
        : ,
        : ,
        : ,
    }



  ():
    job =  jobs_store.get(job_id)
      job:
         HTTPException(, )

    response = {
        : job_id,
        : job[],
        : job[],
        : job.get(),
    }

     job[] == JobStatus.COMPLETED:
        response[] = {: job[], : }

     job[] == JobStatus.FAILED:
        response[] = job.get()

    
     fastapi.responses  JSONResponse
    headers = {}
     job[]  [JobStatus.PENDING, JobStatus.PROCESSING]:
        headers[] =   

     JSONResponse(content=response, headers=headers)



  ():
    job =  jobs_store.get(job_id)
      job:
         HTTPException()
     job[]   [JobStatus.PENDING, JobStatus.PROCESSING]:
         HTTPException(, )
     jobs_store.update(job_id, {: })