Configure the Python SDK

Last updated: October 13, 2026

Create a BabelQueue runtime and point it at your broker with a URL. The transport is chosen from the URL scheme — no separate “broker” flag.

Broker URL

from babelqueue import BabelQueue

# Redis (extra: babelqueue[redis])
app = BabelQueue("redis://localhost:6379/0", queue="orders")

# RabbitMQ (extra: babelqueue[amqp])
# app = BabelQueue("amqp://guest:guest@localhost:5672/", queue="orders")

# Amazon SQS (extra: babelqueue[sqs])
# app = BabelQueue("sqs://eu-central-1", queue="orders")

# Azure Service Bus (extra: babelqueue[azureservicebus])
# app = BabelQueue("sb://<namespace>.servicebus.windows.net", queue="orders")

# Apache Pulsar (extra: babelqueue[pulsar])
# app = BabelQueue("pulsar://localhost:6650", queue="orders")

# Apache Kafka (extra: babelqueue[kafka])
# app = BabelQueue("kafka://localhost:9092", queue="orders")

# Apache ActiveMQ Artemis (extra: babelqueue[artemis])
# app = BabelQueue("artemis://localhost:5672", queue="orders")

# In-process, great for tests/local — no broker required
# app = BabelQueue("memory://", queue="orders")

The sqs:// URL is sqs://<region> with optional query params: endpoint (LocalStack/ElasticMQ), prefix (a queue-URL prefix that skips GetQueueUrl), fifo=1, group_id, wait_time. The transport sends the canonical envelope with the §3 MessageAttributes and reconciles attempts to max(body.attempts, ApproximateReceiveCount − 1).

The sb:// URL is sb://<namespace>.servicebus.windows.net and authenticates with DefaultAzureCredential (Azure AD); pass connection_string=... to AsbTransport for SAS auth, or BabelQueue(transport=...) with a built client. It maps the envelope onto native Service Bus fields (§4) and reconciles attempts to max(body, DeliveryCount − 1).

The pulsar:// URL is pulsar://<host>:<port> (or pulsar+ssl://); pass a built client via BabelQueue(transport=...) for TLS/token auth. It projects the envelope onto native Pulsar message properties (bq-job = URN, plus the bq- fields, §5), defaults to a Shared subscription named babelqueue, and reconciles attempts to max(body, redelivery_count) — the redelivery count is 0-based, so no −1.

The kafka:// URL is kafka://<host>:<port>[,<host>:<port>]; pass an injected producer + consumer_factory for custom auth. The record value is the envelope, projected onto bq- headers (§6); consume is process-then-commit (enable.auto.commit=false) and attempts reconciles to the authoritative bq-attempts header (else the body). Kafka has no native delay/DLQ, so the runtime owns retry by republishing with attempts+1 and dead-letters to <queue>.dlq. The transport keeps the record it last handed out until it is acknowledged or rewound.

The runtime’s own settle writes (the retry republish, the release, and the envelope, poison and schema-version-rejection dead-letter writes) go through publish_settle(), which waits for the broker to confirm that write: it uses the write’s own delivery report and waits up to delivery_timeout seconds (the optional KafkaTransport(delivery_timeout=30.0) argument). On a delivery error or a timeout it raises KafkaDeliveryError, nothing is committed, and the next poll seeks the partition back to the record, so it is redelivered instead of being skipped by the next record’s commit (the handler may run again; keep handlers idempotent). A failed seek is logged and does not block a rebalance. A missing <queue>.dlq topic or an unavailable partition leader therefore no longer loses the record. Your own app.publish() calls, also from inside a handler, are unchanged: fire-and-forget, with no wait for the broker.

A poison record (a value that is not valid UTF-8, not valid JSON, not a JSON object, or that has a member of the wrong type) never raises in pop() and is never retried. It is dead-lettered wrapped, like the Go reference: a body that is exactly one JSON object keeps its keys, anything else becomes {"raw": "<body>"} (bytes that are not UTF-8 are stored with U+FFFD substitutions), and a dead_letter block with reason: "poison", original_queue, attempts (from the bq-attempts header) and lang is attached. The record key is preserved, and the offset is committed only once the DLQ write is confirmed (a failed write leaves it uncommitted and rewinds). A body whose numbers cannot be written back as valid JSON (for example 1e400, or an integer beyond 2^53) is carried verbatim under raw. Poison follows the rules the Go reference applies, with two intentional differences: a value that is not valid UTF-8 is poison here (Go substitutes U+FFFD and accepts it), and JSON nested deeper than the interpreter’s recursion limit (about 1000 levels, where Go allows 10000) is poison. Member names are matched as spelled (a JOB or Meta member is an unknown member, not a typed one), and duplicate keys are each checked. Whatever a decode raises after the shape scan is poison too, so no record can wedge a partition.

A poison record goes to the DLQ whatever on_unknown_urn is: UnknownUrnStrategy.DELETE does not drop it. It is dropped (and committed) only when no DLQ is configured. The bq-schema-version gate rejection still dead-letters its record raw. A bq-attempts header that is not valid UTF-8 does not raise either. Consumers of the DLQ that expected the raw bytes of a malformed record must read the wrapped body.

The retry republish, the envelope dead-letter write, the raw dead-letter write, the redrive write back to the work topic and a redrive restore all keep the consumed record’s key (ReceivedMessage.key, None for a null key): a null key stays null, an empty key stays empty, and a key is never invented. Partition equality across SDKs is not guaranteed.

The artemis:// URL is artemis://<host>:<port> (or artemis+ssl://) — Apache ActiveMQ Artemis over AMQP 1.0 (python-qpid-proton, not RabbitMQ’s 0-9-1 pika); pass a built connection via BabelQueue(transport=...) for auth/TLS. It projects the envelope onto the AMQP a JMS peer reads — x-opt-jms-type = URN, correlation-id = trace_id, plus the string bq_ application properties (underscored — JMS property names must be valid Java identifiers; §7) — and reconciles attempts to max(body, delivery_count); the AMQP delivery-count is 0-based, so no −1.

Schema-version gate

The sqs://, sb://, kafka:// and artemis:// transports check the bq-schema-version message attribute / property / header (Artemis: bq_schema_version) before the body is decoded. A missing or blank value (the empty string, or only ASCII whitespace: space, tab, LF, VT, FF, CR) changes nothing: the body is decoded as usual. A header or property present with a null value counts as missing too. Any other value is compared with "1" exactly as sent, with no trimming, so "2", "x", "01" and " 1" are all rejected without decoding the body. On Azure Service Bus an integer 1 is accepted as well as the string "1"; a float (including 1.0) or a boolean is rejected. A "1" value does not skip the body check: the decoded envelope still goes through the schema_version validation.

A rejected delivery reaches the app on a reserved route, babelqueue:unsupported-schema-version:<version>. The transport gates without decoding the body, and the app does not decode it either. No handler can match the route and a retry cannot change the outcome, so it is never retried. on_unknown_urn="delete" wins over dead-lettering: the delivery is acknowledged and dropped. Otherwise, with dead_letter=True (or on_unknown_urn="dead_letter") the incoming body goes straight to the dead-letter queue byte-identical, with no dead_letter block and no unknown_urn reason; with dead-lettering off it is acknowledged and dropped.

Options

app = BabelQueue(
    "redis://localhost:6379/0",
    queue="orders",
    max_attempts=3,
    dead_letter=True,
    on_unknown_urn="dead_letter",
)

Need just the wire codec, with no broker wiring? Import EnvelopeCodec directly — see Producing messages.

Next: Producing messages.