Relay to a foreign broker

Want a worked end-to-end example? See Tutorial: Add a Kafka relay.

In the outbox pattern, domain code writes a row to the outbox in the same DB transaction as its other writes, and a separate worker relays those rows to a real bus (Kafka, RabbitMQ, NATS, Redis…). faststream-outbox supports this directly via FastStream's cross-broker chain: stack a foreign-broker publisher decorator on an outbox subscriber.

If you don't have a database write to atomically commit alongside, use the foreign broker directly. See Comparison.

Why an outbox relay

When a request must (a) update your database and (b) emit an event onto a message bus, the naive shape (DB commit, then bus publish) leaks events on crashes between the two steps. The outbox pattern fixes this by writing the event as a row in the same transaction as the domain update; a separate worker reads the row and publishes to the bus. The row is the durability boundary, and the relay carries an at-least-once guarantee end to end.

Minimal relay

The snippets below assume engine and outbox_table from the Basic usage setup (make_outbox_table(metadata) + create_async_engine(...)).

from faststream.kafka import KafkaBroker
from faststream_outbox import OutboxBroker

broker_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("127.0.0.1:9092")
publisher_kafka = broker_kafka.publisher("kafka_topic")


@publisher_kafka
@broker_outbox.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body

await broker_outbox.publish(body, queue="outbox_queue", session=session) in your domain transaction writes a row; the subscriber dispatches it; the handler returns it; the Kafka publisher decorator picks it up and publishes to kafka_topic. Failure handling, retries, and DLQ are unchanged from the rest of the outbox subscriber's behavior.

Two-broker lifecycle

Both brokers must be started for the relay to work. As a safety net, at start() the outbox broker logs a WARNING (one per unstarted foreign broker) naming the affected queue(s), and a relay to an unstarted foreign broker fails and retries until that broker is started, so the row is never lost. There are two idiomatic shapes:

FastAPI (recommended)

Mount OutboxRouter and the foreign broker's router on the same FastAPI app. Both auto-start via FastAPI's lifespan.

from fastapi import FastAPI
from faststream.kafka.fastapi import KafkaRouter
from faststream_outbox.fastapi import OutboxRouter

outbox_router = OutboxRouter(engine, outbox_table=outbox_table)
kafka_router = KafkaRouter("127.0.0.1:9092")

publisher_kafka = kafka_router.publisher("kafka_topic")


@publisher_kafka
@outbox_router.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body


app = FastAPI()
app.include_router(outbox_router)
app.include_router(kafka_router)

Standalone

A single FastStream app with the foreign broker's connect hooked into on_startup:

from faststream import FastStream
from faststream.kafka import KafkaBroker
from faststream_outbox import OutboxBroker

broker_outbox = OutboxBroker(engine, outbox_table=outbox_table)
broker_kafka = KafkaBroker("127.0.0.1:9092")
publisher_kafka = broker_kafka.publisher("kafka_topic")


@publisher_kafka
@broker_outbox.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body


app = FastStream(broker_outbox, on_startup=[broker_kafka.connect])

At-least-once contract

If the foreign publish raises (Kafka down, partition unavailable, etc.), the exception propagates through FastStream's AcknowledgementMiddleware, the outbox row is nacked, and the configured retry_strategy reschedules it. The next dispatch re-runs the handler and re-attempts the foreign publish. The net effect is at-least-once delivery to the foreign broker.

Downstream consumers should handle duplicates idempotently, the same way they would behind any at-least-once bus.

Header propagation

By default, FastStream's Response(value) ships with empty headers, so the inbound outbox row's headers (content-type, custom trace keys, etc.) are not forwarded to the foreign publish. Two ways to override:

Set them explicitly, per handler:

from faststream.response import Response
from faststream_outbox.annotations import OutboxMessage


@publisher_kafka
@broker_outbox.subscriber("outbox_queue")
async def relay(body: dict, msg: OutboxMessage) -> Response:
    return Response(body, headers=msg.headers)

Or opt in per subscriber:

@publisher_kafka
@broker_outbox.subscriber("outbox_queue", propagate_inbound_headers=True)
async def relay(body: dict) -> dict:
    return body

With propagate_inbound_headers=True, the subscriber fills Response.headers from the inbound OutboxMessage.headers unless the handler returned a Response(..., headers=...) explicitly.

Using routers

Both halves of the chain can live on routers. The FastAPI shape above already does this with KafkaRouter and OutboxRouter. The constraint is that broker.include_router(router) must happen before the brokers start. Inside FastAPI(..., lifespan=...) the include happens during app construction (before lifespan), so it's automatic.

For the standalone (non-FastAPI) lifecycle, use the plain broker routers (faststream.kafka.KafkaRouter and faststream_outbox.OutboxRouter; the FastAPI routers above only mount on a FastAPI app) and include them before starting:

from faststream.kafka import KafkaRouter
from faststream_outbox import OutboxRouter

kafka_router = KafkaRouter()
outbox_router = OutboxRouter()
publisher_kafka = kafka_router.publisher("kafka_topic")


@publisher_kafka
@outbox_router.subscriber("outbox_queue")
async def relay(body: dict) -> dict:
    return body


broker_kafka.include_router(kafka_router)
broker_outbox.include_router(outbox_router)
# then start

What not to do

Do not combine OutboxResponse(...) and a foreign-publisher decorator:

from faststream_outbox import OutboxResponse


@publisher_kafka
@broker_outbox.subscriber("outbox_queue")
async def relay(body: dict) -> OutboxResponse:
    return OutboxResponse(body=body, queue="next_queue", session=session)  # rejected at dispatch

This would both insert a row into the outbox AND publish to Kafka. The subscriber raises RuntimeError at dispatch time when it detects the combination, so pick one path. The worker catches that error and logs it at ERROR; it does not flush a nack and does not route the row through the retry_strategy. The row's lease expires and a later fetch reclaims it, so the row keeps being retried (not lost) until you fix the configuration.

Do not stack an outbox publisher on a foreign subscriber:

@broker_outbox.publisher("outbox_queue")  # NotImplementedError at decoration
@broker_kafka.subscriber("kafka_topic")
async def relay(body: dict) -> dict:
    return body

This direction would need the Kafka subscriber's dispatch loop to provide an AsyncSession for the outbox insert, and there isn't one without breaking the transactional contract. OutboxPublisher.__call__ raises NotImplementedError at decoration time. Call await broker_outbox.publish(...) inside the handler instead, on a session you opened yourself.

Other foreign brokers

The same pattern works for Confluent, RabbitMQ, NATS, and Redis; only the publisher line changes:

Foreign broker	Publisher line
Kafka	broker_kafka.publisher("topic")
Confluent	broker_confluent.publisher("topic")
RabbitMQ	broker_rabbit.publisher("queue")
NATS	broker_nats.publisher("subject")
Redis	broker_redis.publisher("channel")

Any FastStream broker whose publisher's _publish accepts a generic PublishCommand works as a relay destination. That is the FastStream cross-broker contract, not an outbox-specific feature.