faststream-celery

A FastStream broker that speaks Celery.

Write async FastStream handlers that consume the tasks your Celery producers already send, and publish tasks your Celery workers already run. Both sides keep working while you move services across one at a time — nothing has to be migrated in a big bang, and neither side needs to know the other changed.

from faststream import FastStream
from faststream_celery import CeleryBroker

broker = CeleryBroker("amqp://guest:guest@localhost:5672//")
app = FastStream(broker)


@broker.subscriber("celery", task="proj.tasks.send_email")
async def send_email(args: list, kwargs: dict) -> str:
    return "sent"

That handler now serves app.send_task("proj.tasks.send_email", ...) from any Celery client, result included.

Install

pip install faststream-celery

Extras, all optional:

pip install "faststream-celery[redis]"       # Redis transport and result backend
pip install "faststream-celery[fastapi]"     # FastAPI router
pip install "faststream-celery[otel]"        # OpenTelemetry tracing
pip install "faststream-celery[prometheus]"  # Prometheus metrics

Consuming tasks

task= picks the Celery task a handler serves. Several handlers share one queue, the way Celery routes:

@broker.subscriber("celery", task="proj.tasks.send_email")
async def send_email(args: list, kwargs: dict) -> None: ...


@broker.subscriber("celery", task="proj.tasks.resize")
async def resize(args: list, kwargs: dict) -> None: ...


@broker.subscriber("celery")  # anything the two above do not claim
async def fallback(args: list, kwargs: dict) -> None: ...

Leave task= off and the handler takes every task on the queue. A task nobody claims is rejected, the way a Celery worker rejects one it does not know.

Returning a value is enough to answer the caller: with an AMQP reply_to or a result backend configured, AsyncResult.get() on the client side returns it. A handler that raises comes back as FAILURE with a traceback.

Publishing tasks

from faststream_celery import CeleryTask

await broker.publish(
    CeleryTask("proj.tasks.send_email", args=[user_id], kwargs={"urgent": True}),
    queue="celery",
)

Deferred execution works as in apply_async:

await broker.publish(
    CeleryTask("proj.tasks.digest", countdown=60, expires=3600),
    queue="celery",
)

Deferred tasks are held in memory until they are due, and dropped once their expires has passed.

To wait for a result:

response = await broker.request(CeleryTask("proj.tasks.add", args=[2, 3]), queue="celery")
result = await response.decode()  # {"status": "SUCCESS", "result": 5, ...}

Without a result backend that waits on a temporary AMQP reply queue. With one it reads celery-task-meta-<id>, exactly as AsyncResult.get() does.

Transports and results

CeleryBroker(
    "redis://localhost:6379/0",
    result_backend="redis://localhost:6379/0",
    transport_options={"visibility_timeout": 3600},
)

RabbitMQ and Redis are both supported, and both are tested against a live Celery on every commit. On Redis, visibility_timeout has to match across every client and worker sharing the bus — otherwise one side redelivers what another is still working on.

Messages are JSON, as Celery does by default.

Testing

TestCeleryBroker runs handlers in memory — middlewares, parsing and dependency injection included — with no broker and no connection:

from faststream_celery import CeleryTask, TestCeleryBroker


async def test_send_email() -> None:
    async with TestCeleryBroker(broker):
        await broker.publish(CeleryTask("proj.tasks.send_email", args=[1]), queue="celery")

        send_email.mock.assert_called_once_with({"args": [1], "kwargs": {}})

Assert inside the block: the mocks are reset on the way out.

A broker configured with result_backend= works there too — the backend is swapped for an in-memory one, so results are recorded and request() reads them back with no Redis to run:

from faststream_celery.backend import InMemoryResultBackend


async def test_the_result_is_recorded() -> None:
    async with TestCeleryBroker(broker):
        response = await broker.request(
            CeleryTask("proj.tasks.add", args=[2, 3]),
            queue="celery",
            correlation_id="task-id-1",
        )

        envelope = await response.decode()
        assert envelope["status"] == "SUCCESS"

        backend = broker.config.broker_config.result_backend
        assert isinstance(backend, InMemoryResultBackend)
        assert backend.results["task-id-1"]["result"] == 5

FastAPI

from fastapi import FastAPI
from faststream_celery.fastapi import CeleryRouter

router = CeleryRouter("amqp://guest:guest@localhost:5672//")


@router.subscriber("celery", task="proj.tasks.add")
async def add(args: list[int], kwargs: dict) -> int:
    return sum(args)


app = FastAPI(lifespan=router.lifespan_context)
app.include_router(router)

Observability

from faststream_celery.opentelemetry import CeleryTelemetryMiddleware
from faststream_celery.prometheus import CeleryPrometheusMiddleware

broker = CeleryBroker(
    "amqp://guest:guest@localhost:5672//",
    middlewares=(
        CeleryTelemetryMiddleware(),
        CeleryPrometheusMiddleware(registry=registry),
    ),
)

Spans carry the Celery task name and id, so two tasks on one queue stay apart in a trace. Metrics are labelled by queue; add the task name where you want the two apart there too:

CeleryPrometheusMiddleware(
    registry=registry,
    custom_labels={"task": lambda msg: (msg.message.headers or {}).get("task", "")},
)

faststream docs gen produces an AsyncAPI schema with the queue, exchange and routing key of every subscriber and publisher.

Annotations

from faststream_celery.annotations import CeleryMessage, Logger


@broker.subscriber("celery", task="proj.tasks.add")
async def add(message: CeleryMessage, logger: Logger) -> None:
    logger.info("running %s", message.headers["task"])

Examples

examples/ has one directory per scenario, with both sides present — a FastStream app and the stock Celery process it talks to — so you can start them and watch the tasks cross:

Contributing

uv sync
uv run pytest tests/unit                        # fast, no broker needed
uv run pytest tests/integration -m connected    # live RabbitMQ, Redis and Celery (needs Docker)
uv run nox                                      # every Python, both ends of the dependency ranges
uv run nox -s lint                              # ruff and mypy

CI tests pinned minimum and latest released FastStream/Kombu versions across Python 3.10–3.14. The latest-release combination also runs daily at 04:23 UTC:

uv run nox -s tests -k 'faststream_latest and kombu_latest'

AGENTS.md has the development guidelines and the justfile the day-to-day commands.

License

Apache-2.0.