Pluggable web frameworks and ASGI¶

Basics ¶

Faust and an ASGI server can share a single process, so an HTTP endpoint can produce to a Kafka topic directly:

import faust
from fastapi import FastAPI
from faust.contrib.asgi import faust_lifespan

faust_app = faust.App("hello", broker="kafka://localhost:9092")
greetings = faust_app.topic("greetings", value_type=str)

api = FastAPI(lifespan=faust_lifespan(faust_app))

@faust_app.agent(greetings)
async def print_greetings(stream):
    async for greeting in stream:
        print(greeting)

@api.post("/greet")
async def greet(text: str):
    await greetings.send(value=text)
    return {"ok": True}

Run it with:

$ uvicorn myapp:api

Install the optional dependencies with:

$ pip install "faust-streaming[fastapi]"

Nothing in faust.contrib.asgi imports https://pypi.org/project/fastapi/ or https://pypi.org/project/starlette/, so it works equally well with FastAPI, Starlette, Quart, Litestar, Django ASGI, or any other ASGI callable. Use faust-streaming[asgi] when your selected framework is already installed and you only need the ASGI server integration.

Django and applications without a lifespan hook ¶

Django exposes a standard ASGI callable through django.core.asgi.get_asgi_application(), but it does not accept the constructor-level lifespan handler used in the FastAPI example. Wrap that callable in FaustLifespanMiddleware in the project’s asgi.py:

import os

from django.core.asgi import get_asgi_application
from faust.contrib.asgi import FaustLifespanMiddleware

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "myproject.settings")
django_application = get_asgi_application()

from myproject.faust import app as faust_app

application = FaustLifespanMiddleware(django_application, faust_app)

Run the combined application with any ASGI server:

$ uvicorn myproject.asgi:application

The middleware implements the standard ASGI lifespan protocol. It starts Faust before replying with lifespan.startup.complete and stops Faust before replying with lifespan.shutdown.complete. HTTP and WebSocket scopes pass to Django unchanged, and Faust never imports Django.

The wrapper owns the lifespan scope instead of forwarding it to the inner application. Use faust_lifespan() or faust_app_running() when the selected framework already has startup and shutdown work that must be composed with Faust.

One process, one event loop ¶

This is the whole reason the module exists.

A Faust app resolves its event loop the first time something asks for it, and everything it builds afterwards – producers, consumers, timers, table managers – belongs to that loop. An ASGI server such as https://pypi.org/project/uvicorn/ creates its own loop with asyncio.run(). If the app has already resolved a different loop by then, every one of those objects is attached to a loop that will never run, and you get:

AssertionError: Please create objects with the same loop as running with
RuntimeError: Task ... got Future ... attached to a different loop

faust_lifespan() and faust_app_running() bind the app to the loop that is actually running before starting it, so this cannot happen.

Do not touch app.loop at import time

Declare the app, its topics, agents and tables at module scope as usual – that is supported and is what the examples do. What you must not do is ask the app for its event loop before the server starts. In practice that means avoiding, at module scope:

app.loop
app.web, app.transport, app.producer
passing your own loop to faust.App(loop=...)

If the app is already bound to another loop, bind_to_running_loop() raises LoopMismatch with a message naming the likely cause, rather than letting it fail later and less legibly.

Composing with your own lifespan ¶

When you have setup of your own, use faust_app_running() – it is the context manager faust_lifespan() is built from:

from contextlib import asynccontextmanager
from faust.contrib.asgi import faust_app_running

@asynccontextmanager
async def lifespan(api: FastAPI):
    async with faust_app_running(faust_app):
        ml_models["answer"] = load_model()
        yield
        ml_models.clear()

api = FastAPI(lifespan=lifespan)

The app is started with maybe_start(), so this composes safely with an app that is already running, and will not stop one it did not start.

Running under `faust worker`¶

The other direction: keep faust worker as your entry point and let it serve your ASGI application on its own loop.

from faust.contrib.asgi import serve_asgi

api = FastAPI()
serve_asgi(faust_app, api)

$ faust -A myapp worker -l info

serve_asgi() makes the ASGI application the worker’s web application. It replaces the legacy aiohttp server, starts only after table recovery has finished, and obeys web_enabled and --without-web. By default it binds to web_bind and web_port; extra keyword arguments are passed through to uvicorn.Config.

faust -A looks for app

The -A option imports the module and looks for an attribute named app. Bind that name to the Faust app, not to the FastAPI object:

app = faust_app = faust.App("myapp", ...)
api = FastAPI()

Naming the FastAPI object app – the FastAPI convention – is what makes faust -A fail to find your app.

Replacing `faust.web`¶

Faust keeps its aiohttp-based faust.web stack as the compatibility default, but applications are no longer required to use it. Registering a web server service replaces that stack completely:

from mode import Service

@faust_app.web_server
class MyFrameworkServer(Service):
    async def on_start(self):
        # Start Sanic, Tornado, or another framework here.
        ...

    async def on_stop(self):
        ...

Faust only owns the service lifecycle. The service owns its framework, application object, routes, and HTTP server. It starts after table recovery, uses the same event loop as the worker, and is disabled by the same web_enabled switch as the compatibility server.

serve_asgi() is the ready-made implementation of this hook for ASGI. Because the framework now owns routing, @app.page, @app.table_route, and Faust’s built-in aiohttp endpoints are not mounted when a custom server is registered. Define their replacements in your chosen framework. The new performance payload is framework-neutral, so it can be exposed directly:

from faust.sensors.metrics import performance_metrics

@api.get("/performance")
async def performance():
    return performance_metrics(faust_app)

OpenTelemetry ¶

$ pip install "faust-streaming[opentelemetry]"

With the instrumentation packages installed, most of a distributed trace works already: https://pypi.org/project/opentelemetry-instrumentation-aiokafka/ wraps the same AIOKafkaProducer.send and AIOKafkaConsumer.getmany calls that Faust’s driver uses, so an HTTP request that produces to Kafka carries its trace context onto the wire in a traceparent header.

Two things Faust adds:

Your FastAPI application is instrumented automatically. faust_lifespan() and serve_asgi() attach FastAPIInstrumentor when OpenTelemetry is installed and the application has configured a real TracerProvider. Until an SDK is configured the OpenTelemetry API is a no-op, so nothing is enabled behind your back. Pass opentelemetry=False to opt out, or True to force it. Applications already instrumented (for example under opentelemetry-instrument) are left alone rather than double-wrapped.

The consumer-to-agent hop is bridged. Faust’s consumer runs in its own thread, and contextvars do not cross threads – so the receive span created inside getmany is closed before your agent ever runs, leaving the agent’s work unparented. Register the sensor to close that gap:

from faust.contrib.opentelemetry import setup_opentelemetry

setup_opentelemetry(app)

It extracts the trace context from each message’s headers and opens a {topic} process span that stays current for as long as the stream is processing the event, so everything the agent does nests underneath it. The resulting trace reads:

FastAPI server span
  └─ {topic} send        (PRODUCER, from opentelemetry-instrumentation-aiokafka)
      └─ {topic} process (CONSUMER, from faust.contrib.opentelemetry)
          └─ ...whatever your agent does

Do not run this alongside the opentracing sensor

faust.sensors.distributed_tracing.TracingSensor also injects trace headers into Kafka messages. Running both puts two traceparent headers on the wire. setup_opentelemetry() warns if it sees one registered. Note also that the OpenTracing bridge was deprecated upstream in March 2026; new work should target OpenTelemetry directly.

Examples ¶

Two complete examples ship with Faust:

examples/fastapi_example.py: A single file – hello_world.py plus a FastAPI application.
examples/fastapi_project/: The same thing as a package, with agents, tables, timers and routers in their own modules. main.py is served by uvicorn; worker_main.py shows the same API served from inside faust worker.

Caveats ¶

A topic must exist when the worker starts for its agent to consume from it. This bites co-hosted apps particularly often, because they typically produce to and consume from the same topic: on a first run against a fresh cluster the topic does not exist yet, the agent subscribes to nothing, and messages produced by your endpoints are written but never processed until the process is restarted. Nothing is lost – the backlog is picked up on the next start – but the first run looks like the agent is broken.

This is not specific to co-hosting; faust worker behaves the same way. Create your topics ahead of time (or restart once) when bootstrapping a new environment.
producer_threaded=True spawns a producer with its own thread and loop. It is untested in a co-hosted process and is not supported here yet.
uvicorn --reload and --workers fork or re-exec the process. Faust is not designed to be forked; use a single worker.

Pluggable web frameworks and ASGI¶

Basics ¶

Django and applications without a lifespan hook ¶

One process, one event loop ¶

Composing with your own lifespan ¶

Running under `faust worker`¶

Replacing `faust.web`¶

OpenTelemetry ¶

Examples ¶

Caveats ¶

Faust

Navigation

Related Topics