Quick start

Some useful examples.

Simple consumer

Connect to RabbitMQ, declare a queue, and consume messages using an async iterator. Each message is automatically acknowledged when the process() context manager exits successfully.

import asyncio
import logging

import aio_pika


async def main() -> None:
    logging.basicConfig(level=logging.DEBUG)
    connection = await aio_pika.connect_robust(
        "amqp://guest:[email protected]/",
    )

    queue_name = "test_queue"

    async with connection:
        # Creating channel
        channel = await connection.channel()

        # Will take no more than 10 messages in advance
        await channel.set_qos(prefetch_count=10)

        # Declaring queue
        queue = await channel.declare_queue(queue_name, auto_delete=True)

        async with queue.iterator() as queue_iter:
            async for message in queue_iter:
                async with message.process():
                    print(message.body)

                    if queue.name in message.body.decode():
                        break


if __name__ == "__main__":
    asyncio.run(main())

Messages interrupted by a disconnection

An IncomingMessage belongs to the channel on which it was delivered. After that channel closes, its ack(), reject(), and nack() methods raise ChannelInvalidStateError. Reconnecting with connect_robust() does not make an old message valid on the replacement channel.

If the body of async with message.process(): raises an exception and the delivery channel is closed, the context manager logs that rejection could not be sent and preserves the original exception. If processing succeeds but the channel has closed, automatic acknowledgement still raises ChannelInvalidStateError; the message is not marked processed. ignore_processed=True only skips automatic handling of a message that has already been acknowledged or rejected. It does not revive its channel.

With manual acknowledgements, RabbitMQ requeues unacknowledged deliveries when their channel closes. A robust queue iterator can continue consuming after restoration and receive these messages again. Messages already held by application code or buffered before the disconnection still belong to the old channel. Handle failed acknowledgements and make processing idempotent so that a redelivery does not repeat completed side effects.

Simple publisher

Connect to RabbitMQ and publish a single message to a queue through the default exchange. The default exchange is available as channel.default_exchange; the routing key is the queue name.

import asyncio

import aio_pika


async def main() -> None:
    connection = await aio_pika.connect_robust(
        "amqp://guest:[email protected]/",
    )

    async with connection:
        routing_key = "test_queue"

        channel = await connection.channel()

        await channel.default_exchange.publish(
            aio_pika.Message(body=f"Hello {routing_key}".encode()),
            routing_key=routing_key,
        )


if __name__ == "__main__":
    asyncio.run(main())

Asynchronous message processing

Consume messages using a callback function instead of an async iterator. This allows multiple messages to be processed concurrently, controlled by the prefetch_count setting.

import asyncio

import aio_pika


async def process_message(
    message: aio_pika.abc.AbstractIncomingMessage,
) -> None:
    async with message.process():
        print(message.body)
        await asyncio.sleep(1)


async def main() -> None:
    connection = await aio_pika.connect_robust(
        "amqp://guest:[email protected]/",
    )

    queue_name = "test_queue"

    # Creating channel
    channel = await connection.channel()

    # Maximum message count which will be processing at the same time.
    await channel.set_qos(prefetch_count=100)

    # Declaring queue
    queue = await channel.declare_queue(queue_name, auto_delete=True)

    await queue.consume(process_message)

    try:
        # Wait until terminate
        await asyncio.Future()
    finally:
        await connection.close()


if __name__ == "__main__":
    asyncio.run(main())

Publishing from a consumer callback

With publisher confirms enabled (the default), await exchange.publish() inside the callback to handle broker errors such as ChannelAccessRefused. An exception in a consumer callback does not propagate to the earlier queue.consume() call: that call registers the consumer. Handle the error inside the callback or report it to your application’s supervisor.

Use async with message.process(): instead of unconditionally calling message.ack() in a finally block. If publishing closes the delivery channel, that acknowledgement raises ChannelInvalidStateError and can hide the original publishing error. The processing context preserves it.

A dedicated publishing channel isolates broker channel errors from the consumer’s delivery channel:

import logging

import aio_pika
from aiormq.exceptions import ChannelAccessRefused

logger = logging.getLogger(__name__)
publisher = await connection.channel()

async def consume(message):
    try:
        async with message.process():
            await publisher.default_exchange.publish(
                aio_pika.Message(b"reply"), routing_key="replies",
            )
    except ChannelAccessRefused:
        logger.exception("Publishing denied; check exchange permissions")

await queue.consume(consume)

Here queue uses a different channel from publisher. If the broker closes only the publishing channel, process() can still reject the request on its delivery channel. Its default requeue=False discards the request or routes it to a configured dead-letter exchange; choose this policy deliberately. A connection failure can still close both channels.

When publishing on the delivery channel itself, a broker channel error also prevents rejecting or acknowledging the request. RabbitMQ requeues it, and a robust consumer may receive it again after restoration. requeue=False and reject_on_redelivered=True cannot send a rejection through a closed channel. Repeating the same forbidden publication on every delivery can therefore loop. Fix the permissions, stop the consumer, or apply a retry/dead-letter policy before repeating the publication.

Working with RabbitMQ transactions

Publish messages atomically using AMQP transactions. Messages are only delivered to the queue after the transaction is committed. Shows both the context manager approach and manual select/commit/rollback usage.

import asyncio

import aio_pika


async def main() -> None:
    connection = await aio_pika.connect_robust(
        "amqp://guest:[email protected]/",
    )

    async with connection:
        routing_key = "test_queue"

        # Transactions conflicts with `publisher_confirms`
        channel = await connection.channel(publisher_confirms=False)

        # Use transactions with async context manager
        async with channel.transaction():
            # Publishing messages but delivery will not be done
            # before committing this transaction
            for i in range(10):
                message = aio_pika.Message(body="Hello #{}".format(i).encode())

                await channel.default_exchange.publish(
                    message,
                    routing_key=routing_key,
                )

        # Using transactions manually
        tx = channel.transaction()

        # start transaction manually
        await tx.select()

        await channel.default_exchange.publish(
            aio_pika.Message(body="Hello {}".format(routing_key).encode()),
            routing_key=routing_key,
        )

        await tx.commit()

        # Using transactions manually
        tx = channel.transaction()

        # start transaction manually
        await tx.select()

        await channel.default_exchange.publish(
            aio_pika.Message(body="Should be rejected".encode()),
            routing_key=routing_key,
        )

        await tx.rollback()


if __name__ == "__main__":
    asyncio.run(main())

Get single message example

Fetch a single message from a queue using queue.get() instead of continuous consumption. This is useful for polling or one-off retrieval, with manual acknowledgement of the received message.

import asyncio
from typing import Optional

from aio_pika import Message, connect_robust
from aio_pika.abc import AbstractIncomingMessage


async def main() -> None:
    connection = await connect_robust(
        "amqp://guest:[email protected]/?name=aio-pika%20example",
    )

    queue_name = "test_queue"
    routing_key = "test_queue"

    # Creating channel
    channel = await connection.channel()

    # Declaring exchange
    exchange = await channel.declare_exchange("direct", auto_delete=True)

    # Declaring queue
    queue = await channel.declare_queue(queue_name, auto_delete=True)

    # Binding queue
    await queue.bind(exchange, routing_key)

    await exchange.publish(
        Message(
            bytes("Hello", "utf-8"),
            content_type="text/plain",
            headers={"foo": "bar"},
        ),
        routing_key,
    )

    # Receiving one message
    incoming_message: Optional[AbstractIncomingMessage] = await queue.get(
        timeout=5, fail=False
    )
    if incoming_message:
        # Confirm message
        await incoming_message.ack()
    else:
        print("Queue empty")

    await queue.unbind(exchange, routing_key)
    await queue.delete()
    await connection.close()


if __name__ == "__main__":
    asyncio.run(main())

Set logging level

Sometimes you want to see only your debug logs, but when you just call logging.basicConfig(logging.DEBUG) you set the debug log level for all loggers, includes all aio_pika’s modules. If you want to set logging level independently see following example:

import logging

from aio_pika import logger


logger.setLevel(logging.ERROR)

Connection and channel callbacks

Connections, channels and queues expose callback collections. A callback is a plain function, a coroutine function or a bound method. The first argument is always the object that owns the collection (the sender), the other arguments depend on the collection:

Collection	Sender	Extra arguments
connection.close_callbacks	connection	exc: BaseException | None
connection.reconnect_callbacks (robust)	connection	none
channel.close_callbacks	channel	exc: BaseException | None
channel.return_callbacks	channel	message: AbstractIncomingMessage
channel.reopen_callbacks (robust)	channel	none
queue.close_callbacks	queue	exc: BaseException | None

The sender can be None when the owner was garbage collected before the callback ran. The exc argument is the exception that closed the object. It is set for a close requested by your code too: ChannelClosed for a channel and ConnectionClosed for a connection. Use connection.close_called to tell an intentional close from a failure.

A callback with a wrong number of arguments is not called; the error is logged as Callback ... error. The collections are typed, so mypy reports such a mismatch before you run the code.

import asyncio
import logging

import aio_pika
from aio_pika.abc import (
    AbstractChannel,
    AbstractConnection,
    AbstractIncomingMessage,
    AbstractRobustConnection,
)


# Every callback receives the object that owns the collection as the
# first argument. The other arguments depend on the collection.
def on_close(
    connection: AbstractConnection | None,
    exc: BaseException | None,
) -> None:
    # exc is set even for a close requested by your code.
    logging.info("Connection %r closed: %r", connection, exc)


async def on_reconnect(connection: AbstractRobustConnection | None) -> None:
    # A coroutine function is also accepted. It is scheduled as a task.
    logging.info("Connection %r reconnected", connection)


def on_return(
    channel: AbstractChannel | None,
    message: AbstractIncomingMessage,
) -> None:
    logging.warning("Message %r was returned by the broker", message)


class Service:
    def on_channel_close(
        self,
        channel: AbstractChannel | None,
        exc: BaseException | None,
    ) -> None:
        # A bound method works too: self is bound, channel is the sender.
        logging.info("Channel %r closed: %r", channel, exc)


async def main() -> None:
    logging.basicConfig(level=logging.INFO)
    connection = await aio_pika.connect_robust(
        "amqp://guest:[email protected]/",
    )
    connection.close_callbacks.add(on_close)
    connection.reconnect_callbacks.add(on_reconnect)

    service = Service()

    async with connection:
        channel = await connection.channel()
        channel.close_callbacks.add(service.on_channel_close)
        channel.return_callbacks.add(on_return)

        await channel.default_exchange.publish(
            aio_pika.Message(b"hello"),
            routing_key="no-such-queue",
        )


if __name__ == "__main__":
    asyncio.run(main())

External credentials example

Connect to RabbitMQ using TLS client certificates (x509) for authentication instead of username/password. Requires CA certificate, client certificate, and private key files.

import asyncio
import ssl

import aio_pika
from aio_pika.abc import SSLOptions


async def main() -> None:
    connection = await aio_pika.connect_robust(
        host="127.0.0.1",
        login="",
        ssl=True,
        ssl_options=SSLOptions(
            cafile="cacert.pem",
            certfile="cert.pem",
            keyfile="key.pem",
            no_verify_ssl=ssl.CERT_REQUIRED,
        ),
        client_properties={"connection_name": "aio-pika external credentials"},
    )

    async with connection:
        routing_key = "test_queue"

        channel = await connection.channel()

        await channel.default_exchange.publish(
            aio_pika.Message(body="Hello {}".format(routing_key).encode()),
            routing_key=routing_key,
        )


if __name__ == "__main__":
    asyncio.run(main())

Connection pooling

A single AMQP connection multiplexes multiple channels over one TCP socket, which is sufficient for most applications. However, a single connection has a finite throughput limited by its TCP link and the serialization of frames. Connection pooling may help when:

• You are publishing or consuming a very high volume of messages and a single TCP connection becomes saturated.

• You want to isolate groups of channels so that a blocked or slow connection does not affect other workloads.

• Your application runs many concurrent tasks that would contend for the same connection’s write lock.

In most cases you do not need connection pooling — start with a single connect_robust() connection and only add pooling after profiling shows that the connection is a bottleneck.

Use aio_pika.pool.Pool to manage a pool of connections and channels.

import asyncio

import aio_pika
from aio_pika.abc import AbstractRobustConnection
from aio_pika.pool import Pool


async def main() -> None:
    async def get_connection() -> AbstractRobustConnection:
        return await aio_pika.connect_robust("amqp://guest:guest@localhost/")

    connection_pool: Pool = Pool(get_connection, max_size=2)

    async def get_channel() -> aio_pika.Channel:
        async with connection_pool.acquire() as connection:
            return await connection.channel()

    channel_pool: Pool = Pool(get_channel, max_size=10)
    queue_name = "pool_queue"

    async def consume() -> None:
        async with channel_pool.acquire() as channel:  # type: aio_pika.Channel
            await channel.set_qos(10)

            queue = await channel.declare_queue(
                queue_name,
                durable=False,
                auto_delete=False,
            )

            async with queue.iterator() as queue_iter:
                async for message in queue_iter:
                    print(message)
                    await message.ack()

    async def publish() -> None:
        async with channel_pool.acquire() as channel:  # type: aio_pika.Channel
            await channel.default_exchange.publish(
                aio_pika.Message(("Channel: %r" % channel).encode()),
                queue_name,
            )

    async with connection_pool, channel_pool:
        task = asyncio.create_task(consume())
        await asyncio.wait([asyncio.create_task(publish()) for _ in range(50)])
        await task


if __name__ == "__main__":
    asyncio.run(main())

FastAPI example

Integrate aio-pika with a FastAPI application. The connection is established during startup via the lifespan context manager and stored in app.state for use across request handlers.

Note

A single robust connection is sufficient for most use cases. AMQP multiplexes work over channels within one connection. Consider connection pooling only if you have measured that a single connection is a bottleneck.

from contextlib import asynccontextmanager
from typing import AsyncGenerator

import aio_pika
from fastapi import FastAPI


@asynccontextmanager
async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
    app.state.connection = await aio_pika.connect_robust(
        "amqp://guest:guest@localhost/",
    )
    app.state.channel = await app.state.connection.channel()
    try:
        yield
    finally:
        await app.state.connection.close()


app = FastAPI(lifespan=lifespan)


@app.post("/publish")
async def publish_message(message: str) -> dict:
    await app.state.channel.default_exchange.publish(
        aio_pika.Message(body=message.encode()),
        routing_key="test_queue",
    )
    return {"status": "ok"}


@app.get("/consume")
async def consume_message() -> dict:
    queue = await app.state.channel.declare_queue(
        "test_queue",
        auto_delete=True,
    )
    message = await queue.get(timeout=5, fail=False)

    if message:
        await message.ack()
        return {"body": message.body.decode()}

    return {"body": None}

Starlette example

Integrate aio-pika with a Starlette application. The connection is managed through startup/shutdown event handlers and stored in app.state for use across request handlers.

Note

A single robust connection is sufficient for most use cases. AMQP multiplexes work over channels within one connection. Consider connection pooling only if you have measured that a single connection is a bottleneck.

import aio_pika
from starlette.applications import Starlette
from starlette.requests import Request
from starlette.responses import JSONResponse
from starlette.routing import Route


async def publish_message(request: Request) -> JSONResponse:
    channel = request.app.state.channel
    body = await request.body()
    await channel.default_exchange.publish(
        aio_pika.Message(body=body),
        routing_key="test_queue",
    )
    return JSONResponse({"status": "ok"})


async def consume_message(request: Request) -> JSONResponse:
    channel = request.app.state.channel
    queue = await channel.declare_queue("test_queue", auto_delete=True)
    message = await queue.get(timeout=5, fail=False)

    if message:
        await message.ack()
        return JSONResponse({"body": message.body.decode()})

    return JSONResponse({"body": None})


async def on_startup() -> None:
    app.state.connection = await aio_pika.connect_robust(
        "amqp://guest:guest@localhost/",
    )
    app.state.channel = await app.state.connection.channel()


async def on_shutdown() -> None:
    await app.state.connection.close()


app = Starlette(
    routes=[
        Route("/publish", publish_message, methods=["POST"]),
        Route("/consume", consume_message),
    ],
    on_startup=[on_startup],
    on_shutdown=[on_shutdown],
)

aiohttp example

Integrate aio-pika with an aiohttp web application. The connection is created on application startup and stored in the app dict for access from request handlers.

Note

A single robust connection is sufficient for most use cases. AMQP multiplexes work over channels within one connection. Consider connection pooling only if you have measured that a single connection is a bottleneck.

import aio_pika
from aiohttp import web


async def publish_message(request: web.Request) -> web.Response:
    channel = request.app["channel"]
    body = await request.read()
    await channel.default_exchange.publish(
        aio_pika.Message(body=body),
        routing_key="test_queue",
    )
    return web.json_response({"status": "ok"})


async def consume_message(request: web.Request) -> web.Response:
    channel = request.app["channel"]
    queue = await channel.declare_queue("test_queue", auto_delete=True)
    message = await queue.get(timeout=5, fail=False)

    if message:
        await message.ack()
        return web.json_response({"body": message.body.decode()})

    return web.json_response({"body": None})


async def on_startup(app: web.Application) -> None:
    app["connection"] = await aio_pika.connect_robust(
        "amqp://guest:guest@localhost/",
    )
    app["channel"] = await app["connection"].channel()


async def on_shutdown(app: web.Application) -> None:
    await app["connection"].close()


app = web.Application()
app.router.add_post("/publish", publish_message)
app.router.add_get("/consume", consume_message)
app.on_startup.append(on_startup)
app.on_shutdown.append(on_shutdown)

if __name__ == "__main__":
    web.run_app(app)

Tornado example

Integrate aio-pika with a Tornado web application. The publisher handler sends messages on POST requests, while the subscriber handler waits for incoming messages and returns them as HTTP responses.

Note

A single robust connection is sufficient for most use cases. AMQP multiplexes work over channels within one connection. Consider connection pooling only if you have measured that a single connection is a bottleneck.

import asyncio

import tornado.ioloop
import tornado.web

from aio_pika import Message, connect_robust


class Base:
    QUEUE: asyncio.Queue


class SubscriberHandler(tornado.web.RequestHandler, Base):
    async def get(self) -> None:
        message = await self.QUEUE.get()
        await self.finish(message.body)


class PublisherHandler(tornado.web.RequestHandler):
    async def post(self) -> None:
        connection = self.application.settings["amqp_connection"]
        channel = await connection.channel()

        try:
            await channel.default_exchange.publish(
                Message(body=self.request.body),
                routing_key="test",
            )
        finally:
            await channel.close()

        await self.finish("OK")


async def make_app() -> tornado.web.Application:
    amqp_connection = await connect_robust()

    channel = await amqp_connection.channel()
    queue = await channel.declare_queue("test", auto_delete=True)
    Base.QUEUE = asyncio.Queue()

    await queue.consume(Base.QUEUE.put, no_ack=True)

    return tornado.web.Application(
        [(r"/publish", PublisherHandler), (r"/subscribe", SubscriberHandler)],
        amqp_connection=amqp_connection,
    )


async def main() -> None:
    app = await make_app()
    app.listen(8888)
    await asyncio.Future()


if __name__ == "__main__":
    asyncio.run(main())