aio-pika

aio-pika is a Python library for talking to RabbitMQ over AMQP 0.9.1, built specifically for asyncio rather than adapted from a synchronous client. It sits on top of aiormq (a lower-level AMQP protocol implementation from the same author) and adds an object-oriented, fully type-hinted API: async context managers for connections and channels, async iterators for consuming queues, and explicit Message/IncomingMessage wrappers instead of raw protocol frames.

Its standout feature is the “robust” connection layer (connect_robust), which transparently reconnects after a dropped connection and replays the broker-side state — declared queues and exchanges, bindings, and active consumers — so applications don’t have to hand-roll reconnection bookkeeping. On top of the core connection/channel/queue/exchange primitives, aio-pika ships two ready-made messaging patterns: a Master/Worker helper for fan-out task distribution and an RPC helper for request/response calls over AMQP, plus a generic connection/channel pool for concurrent workloads.

The project has been maintained since 2016, has complete type-hint coverage (ships a py.typed marker), and is used as the underlying broker layer by higher-level frameworks like FastStream, making it a common foundation piece in async Python messaging stacks.

What You Get

• A fully asyncio-native AMQP client with async context managers and async-iterator-based queue consumption
• connect_robust() — auto-reconnecting connections that replay declared queues, exchanges, bindings, and consumers after a dropped connection
• Publisher confirms and AMQP transactions supported out of the box
• Complete type-hint coverage (py.typed) for editor autocomplete and static analysis with mypy
• Built-in patterns.Master/patterns.Worker helpers for fan-out task distribution and patterns.RPC for request/response calls over AMQP
• A generic async connection/channel pool (aio_pika.pool.Pool) for concurrent consumer/producer workloads

Common Use Cases

• Building asyncio microservices that publish and consume messages from RabbitMQ queues
• Implementing reliable background task workers that survive broker restarts and network blips via connect_robust
• Building request/response RPC services over AMQP instead of HTTP
• Fan-out task distribution across a pool of workers using the patterns.Master helper
• Acting as the transport layer underneath higher-level frameworks such as FastStream’s RabbitBroker

Under The Hood

Architecture aio-pika is layered around abstract base classes (abc.py, over 1,000 lines defining the library’s protocols) implemented concretely by Connection, Channel, Queue, and Exchange (connection.py, channel.py, queue.py, exchange.py), with a parallel “robust” layer (RobustConnection, RobustChannel, RobustQueue, RobustExchange) that subclasses the core implementations to add auto-reconnection and full state recovery, tracking live channels via a WeakSet and coordinating reconnect attempts with an asyncio.Lock. The library wraps aiormq rather than reimplementing AMQP wire-protocol handling itself, so aio-pika’s own responsibility is the ergonomic, asyncio-idiomatic layer — async context managers, async iterators for consumption, and Message/IncomingMessage wrappers — on top of aiormq’s connection/channel primitives. A patterns/ subpackage (master.py, rpc.py) builds Master/Worker and RPC messaging patterns directly on the core Connection/Channel/Queue/Exchange API, and a generic pool (pool.py) is provided for concurrent workloads. Because the robust layer and the patterns subpackage both inherit from and compose the core classes directly, changes to the core abstractions ripple through both.

Tech Stack Built for Python 3.11+ using the modern uv_build backend with uv.lock for locked dependency resolution. Runtime dependencies are minimal: aiormq (the underlying AMQP 0.9.1 protocol implementation, same author) and yarl for URL parsing, with an optional uvloop extra for a faster event loop. Dev tooling includes mypy with a fairly strict configuration (disallow_untyped_defs, disallow_incomplete_defs, disallow_untyped_calls) and ruff for linting; testing uses pytest, pytest-cov, pytest-rst, and coverage/coveralls; documentation is built with sphinx, furo, and myst-parser. GitHub Actions workflows (tests.yml, docs.yml, publish.yml) run the test suite, build docs, and publish releases.

Code Quality The tests/ directory holds 16 test files covering connections, robust reconnection, RPC, pooling, message handling, and memory leaks (test_amqp.py, test_amqp_robust.py, test_rpc.py, test_pool.py, test_memory_leak.py), using pytest with async fixtures and a docker_client.py helper — these are integration tests that exercise a live RabbitMQ broker rather than pure mocks, which is thorough but infrastructure-dependent. mypy’s strict-ish settings for the main package and the shipped py.typed marker mean the public API is comprehensively type-hinted; the test suite itself uses a more relaxed mypy override. Error handling favors explicit, named exception classes (DeliveryError, MessageProcessError, ProbableAuthenticationError in exceptions.py) over bare exceptions or silent failures.

API Design aio-pika’s central ergonomic win over using aiormq (or most other AMQP clients) directly is transparent state recovery: connect_robust() reconnects after a dropped connection and automatically re-declares queues/exchanges and re-establishes bindings and consumers, work that application code would otherwise have to redo by hand. Queue consumption maps onto Python’s native async-iterator protocol (async for message in queue_iterator) and message acknowledgment uses async context managers (async with message.process()), reading as idiomatic asyncio rather than the callback-based APIs common in older AMQP clients. The bundled patterns.Master/patterns.RPC helpers give ready-made worker-queue and RPC-over-AMQP implementations so common messaging topologies don’t need to be hand-rolled from raw queues and exchanges.