AMQP Python Client

The recommended Python client is Pika. It’s pure Python with no dependencies, and its blocking API keeps simple publishers and workers short. Install it with:

pip install pika

Publish and consume

The example reads the broker URL from BROKER_URL and falls back to a local LavinMQ at amqp://guest:guest@localhost/%2f.

Save the connection helper as amqp.py:

import os

import pika


def connect():
    url = os.environ.get("BROKER_URL", "amqp://guest:guest@localhost/%2f")
    return pika.BlockingConnection(pika.URLParameters(url))

Save the publisher as publisher.py:

import pika

import amqp


def publish(channel, queue, message):
    channel.queue_declare(queue=queue, durable=True)
    channel.confirm_delivery()
    channel.basic_publish(
        exchange="",
        routing_key=queue,
        body=message,
        properties=pika.BasicProperties(delivery_mode=pika.DeliveryMode.Persistent),
    )
    print(f" [x] Sent {message!r}")


if __name__ == "__main__":
    with amqp.connect() as connection:
        publish(connection.channel(), "hello", "Hello world!")

Save the consumer as worker.py in the same directory:

import amqp


def consume(channel, queue, handler):
    channel.queue_declare(queue=queue, durable=True)
    channel.basic_qos(prefetch_count=10)

    def on_message(ch, method, properties, body):
        handler(body.decode())
        ch.basic_ack(delivery_tag=method.delivery_tag)

    channel.basic_consume(queue=queue, on_message_callback=on_message)


if __name__ == "__main__":
    with amqp.connect() as connection:
        channel = connection.channel()
        consume(channel, "hello", lambda message: print(f" [x] Received {message!r}"))
        print(" [*] Waiting for messages. To exit press CTRL+C")
        try:
            channel.start_consuming()
        except KeyboardInterrupt:
            pass

Run python worker.py in one terminal and python publisher.py in another.

The full, runnable example lives in python/pika in cloudamqp/examples, and CI tests it against LavinMQ.

Good to know

• The worker acks manually. basic_consume defaults to auto_ack=False, and the worker calls basic_ack after the handler returns. If the handler raises, the exception escapes start_consuming, the connection closes and LavinMQ requeues the unacked message.
• Set a prefetch limit. Pika sets none on its own, and basic_qos defaults to prefetch_count=0, meaning unlimited. The worker calls basic_qos(prefetch_count=10) to cap how many unacked messages it holds.
• Pika doesn’t reconnect. After a network failure, catch pika.exceptions.AMQPConnectionError, reconnect and consume again. A message can arrive twice after a reconnect, so keep handlers idempotent.
• Use amqps:// for TLS. Pika then uses Python’s default SSL context, which verifies the server certificate and hostname.

The Pika README and documentation cover connection recovery, threading and the asynchronous adapters.

Other Python clients

• aio-pika, an asyncio client with automatic reconnects
• Kombu, the messaging library behind Celery