Week 13 of 16

Build: Async FastAPI Endpoint

Add a /generate endpoint to your FastAPI app that returns prompts for all platforms in one parallel call.

Day 65 60 minutes Build

Day 65 of 80

FastAPI is Already Async

FastAPI was built for async from day one. Unlike Flask (which needed an extension for async support), every FastAPI route can be async def and use await directly. No special setup.

This means you can fire parallel Claude API calls directly inside a route handler — and FastAPI will handle them correctly without blocking other requests.

Add to api.py

Open your api.py from Week 12 and add these imports and the new endpoint:

api.py — additions Python 
# Add to the imports at the top of api.py:
import asyncio
import anthropic
from dotenv import load_dotenv

load_dotenv()

# Add after the existing db and app setup:
# The async Anthropic client — used in the generate endpoint.
async_client = anthropic.AsyncAnthropic()


# --- New Pydantic model for generate requests ---

class GenerateRequest(BaseModel):
    """Request body for POST /generate."""
    shot: str
    platforms: list[str] = ["Kling", "Runway", "Veo"]

    @field_validator("shot")
    @classmethod
    def shot_not_empty(cls, v):
        if not v or not v.strip():
            raise ValueError("Shot description cannot be empty")
        return v.strip()


# --- The generate endpoint ---

@app.post("/generate")
async def generate_prompts(data: GenerateRequest):
    """Generate prompts for a shot across multiple platforms — in parallel.

    POST body:
    {
        "shot": "Wide aerial over foggy mountains",
        "platforms": ["Kling", "Runway", "Veo"]
    }

    Returns:
    {
        "shot": "...",
        "results": [
            {"platform": "Kling", "prompt": "..."},
            {"platform": "Runway", "prompt": "..."},
            {"platform": "Veo", "prompt": "..."}
        ]
    }
    """

    async def gen_one(platform):
        """Generate for a single platform. Called in parallel."""
        message = await async_client.messages.create(
            model="claude-sonnet-4-6",
            max_tokens=150,
            messages=[{
                "role": "user",
                "content": f"Write a {platform} AI video prompt for: {data.shot}. "
                           f"Under 75 words. Only the prompt."
            }]
        )
        return {"platform": platform, "prompt": message.content[0].text}

    # Run all platform generations in parallel.
    # [gen_one(p) for p in data.platforms] builds the list of coroutines.
    # *[...] unpacks the list into positional arguments for gather().
    results = await asyncio.gather(
        *[gen_one(p) for p in data.platforms]
    )

    logger.info(f"Generated {len(results)} prompts for: {data.shot[:50]}")

    return {
        "shot": data.shot,
        "results": list(results),
    }

The nested async def gen_one() defines a helper function inside the route handler. This is fine in Python — functions can be defined anywhere. It captures data from the outer scope (a closure), so you don't need to pass it as an argument.

FastAPI routes can be async def directly. When a request comes in, FastAPI awaits your handler. Inside the handler, you can await other things. The event loop handles multiple concurrent requests automatically — you don't need to set anything up.

The list comprehension inside gather() builds the tasks dynamically from data.platforms. If the user sends 2 platforms, you get 2 parallel calls. If they send 3, you get 3. The code doesn't change.

Test in /docs

Run your API: uvicorn api:app --reload

Open http://localhost:8000/docs and find the POST /generate endpoint. Use "Try it out" with this body:

Test request body JSON 
{
  "shot": "Slow motion golf swing at sunset, backlit",
  "platforms": ["Kling", "Runway", "Veo"]
}

You'll get back three prompts in one response. The API call took roughly as long as one individual Claude call.

Week 13 Complete

You now write async Python. Your batch generator can process dozens of API calls in the time it used to take for one. Your FastAPI app handles parallel generation natively.

Async is one of the skills that separates hobbyist Python from production Python. You have it.

End of Week Checklist