Skip to content
Pydantic Docs

FastStream

See every message your FastStream app publishes and consumes: which channel it went to, how long handling took, and whether it failed, as a span (one unit of work with a name, a start, and a duration) in Logfire. Spans link together into a trace (the full journey of one message), so you can follow a message from where it was published to where it was handled.

FastStream ships its own OpenTelemetry middleware: a small piece that wraps each broker to record this. So instead of a logfire.instrument_* call, you add FastStream’s middleware for your broker; Logfire receives what it emits.

What you’ll capture

  • Each published and consumed message as a span, with its duration and status
  • The channel or subject the message went to
  • Failed message handling, with the error

Before you start

You’ll need a Logfire project. Open Add data in your project (top navigation) and follow the setup for your language: it signs your machine in with logfire auth (a browser sign-in, no token to copy) and, for production or other languages, creates a write token (the credential your app uses to send data). New to Logfire? Start with Getting Started.

Installation

FastStream has no separate Logfire extra, but its OpenTelemetry middleware needs FastStream’s own otel extra (which pulls in the OpenTelemetry SDK). Install logfire alongside it:

Terminal
pip install logfire
Terminal
pip install 'faststream[otel]'

Usage

Two steps:

  1. Call logfire.configure() to connect to your project.
  2. Add FastStream’s OpenTelemetry middleware for your broker.

The example below uses Redis, so it adds the RedisTelemetryMiddleware. If you use a different broker, add that broker’s matching middleware instead.

main.py
from faststream import FastStream
from faststream.redis import RedisBroker
from faststream.redis.opentelemetry import RedisTelemetryMiddleware

import logfire

logfire.configure()

broker = RedisBroker(middlewares=(RedisTelemetryMiddleware(),))

app = FastStream(broker)


@broker.subscriber('test-channel')
@broker.publisher('another-channel')
async def handle():
    return 'Hi!'


@broker.subscriber('another-channel')
async def handle_next(msg: str):
    assert msg == 'Hi!'


@app.after_startup
async def test():
    await broker.publish('', channel='test-channel')

Verify it worked

Run your app so it publishes a message, then open the Live view. Within a few seconds you’ll see spans for the published and consumed messages: click one to see the channel and how long handling took.

Troubleshooting

Not seeing your messages? Check that logfire.configure() ran before your app started, that your write token is set (run logfire projects use <your-project> locally, or set the LOGFIRE_TOKEN environment variable in production; see Getting Started), and that you added the telemetry middleware for the broker you’re actually using.

Reference