Hi there! Thanks for detailed write-up. This is a very common scenario when dealing with streaming responses in Python and ASGI frameworks.

The issue you're experiencing with the "generator trap" (where pre-stream exceptions aren't raised until iteration begins) is tied to how Python generators work.

Fortunately, your second approach (separating the pre-stream logic from the generator) is exactly the right mental model! The reason it raised a TypeError is due to a clash with the response_class argument when you are already returning a custom Response instance directly.

When you return a Response object (like EventSourceResponse) directly from the endpoint, you should remove response_class=EventSourceResponse from the @app.get decorator.

Here is the corrected, idiomatic way to achieve exactly what you want: clean HTTP exceptions before the stream, and standard SSE behavior after.

from fastapi import FastAPI, HTTPException
# Assuming you are using sse_starlette (FastAPI doesn't have fastapi.sse built-in)
from sse_starlette.sse import EventSourceResponse, ServerSentEvent 

app = FastAPI()

# 1. REMOVE response_class=EventSourceResponse from the decorator
@app.get("/hello")
async def hello_world():
    # 2. Pre-stream logic runs synchronously here.
    # This will correctly return a standard JSON HTTP 501 response!
    raise HTTPException(status_code=501, detail="This endpoint is not yet implemented")

    # 3. Define the async generator for the stream
    async def _hello_world_generator():
        for char in "Hello, World!":
            yield ServerSentEvent(data=char)
            
    # 4. Return the instantiated Response object directly
    return EventSourceResponse(_hello_world_generator())

Why this solves your problem:
Because hello_world no longer contains a yield in its outer scope, FastAPI evaluates it as a standard coroutine immediately upon request. Any HTTPException raised here is caught by FastAPI's standard exception handlers and returned as normal JSON. If no exception is raised, it instantiates and returns the EventSourceResponse, which FastAPI will then stream seamlessly.

Because this pattern fully handles the separation of "normal request errors" vs "active stream errors" without requiring modifications to FastAPI's core routing logic, a change to fastapi/routing.py is likely not necessary.

Hope this helps! Let me know if you run into any other issues.

I found a potential workaround by moving the pre-stream logic into FastAPI dependencies, since dependencies are resolved before the generator starts being consumed.

This allows validation and similar checks to raise HTTPException through the normal error handling flow.

That said, it still seems worth discussing whether this behavior should be handled more explicitly at the framework level.

Quick correction first: fastapi.sse is built directly into FastAPI — from fastapi.sse import EventSourceResponse, ServerSentEvent is the correct import, no extra package needed.

Why pre-stream exceptions are silently lost

When your endpoint is an async generator (async def … yield …), Python does not execute the function body when the endpoint is called. It returns a generator object immediately. The body only runs when FastAPI starts iterating the generator to stream data — at that point HTTP headers (200 OK, Content-Type: text/event-stream) have already been sent to the client, so there is no way to send an HTTP error response.

# BAD — raise before yield never fires at request time
@app.get("/hello", response_class=EventSourceResponse)
async def hello_world() -> AsyncIterable[ServerSentEvent]:
    raise HTTPException(status_code=501)  # ← runs during iteration, not on request
    for char in "Hello, World!":
        yield ServerSentEvent(data=char)

Solution 1 — FastAPI Dependency (recommended)

Dependencies are resolved synchronously before the generator is created, so HTTPException raised there works exactly like on any normal endpoint:

from collections.abc import AsyncIterable
from fastapi import Depends, FastAPI, HTTPException
from fastapi.sse import EventSourceResponse, ServerSentEvent

app = FastAPI()

async def check_access():
    """Raise HTTPException here to block the stream before it starts."""
    authorized = False  # your real check here
    if not authorized:
        raise HTTPException(status_code=403, detail="Forbidden")

@app.get("/hello", response_class=EventSourceResponse)
async def hello_world(
    _: None = Depends(check_access),
) -> AsyncIterable[ServerSentEvent]:
    for char in "Hello, World!":
        yield ServerSentEvent(data=char)

A GET /hello with failing auth → 403 JSON response, never touches the generator.

Solution 2 — Separate pre-stream logic from the generator

If your pre-stream logic needs the same scope as the generator (e.g. a DB connection), keep them in the same function but separate the non-generator part:

@app.get("/hello")
async def hello_world():
    # Everything here runs synchronously on the request
    authorized = False  # your real check
    if not authorized:
        raise HTTPException(status_code=403, detail="Forbidden")

    async def _stream() -> AsyncIterable[ServerSentEvent]:
        for char in "Hello, World!":
            yield ServerSentEvent(data=char)

    # Return the EventSourceResponse directly (no response_class on the decorator)
    return EventSourceResponse(_stream())

Note: when you return EventSourceResponse(...) directly, omit response_class=EventSourceResponse from the decorator — FastAPI only needs that decorator argument when the endpoint itself is the async generator.

TL;DR: use a Depends() for pre-stream validation — that is the idiomatic FastAPI pattern and requires zero changes to your generator logic.

I also have a setup where I directly return an EventSourceResponse and omit response_class=EventSourceResponse from the route decorator because I need to handle several HTTP response errors before I start streaming. One major drawback I have noticed is that the "keep alive" ping comments (see bullet point 1 in the SSE docs) are not sent when you have this setup. You will either have to accept this or implement your own system to send ping messages for long-running requests.