AMQP 1.0 Client Library for Python

1.6.11 · maintenance · verified Thu Apr 09

uAMQP is an AMQP 1.0 Client Library for Python, serving as a C-extension wrapper around the Azure uAMQP C library. It provides low-level AMQP protocol implementation. The current version is 1.6.11. As of v1.6.5, the library is in maintenance mode, with major Azure SDKs (Event Hubs, Service Bus) now utilizing a pure Python AMQP library for new development.

Warnings

deprecated The `uamqp` library is no longer in active development and has moved to a maintenance-only status.
Fix: For new Azure messaging development (Event Hubs, Service Bus), consider using the higher-level `azure-eventhub` or `azure-servicebus` libraries, which now use a pure Python AMQP implementation. `uamqp` will only receive critical bug and security fixes.
gotcha As a C-extension wrapper, `uamqp` requires C/C++ build tools and Python development headers for installation from source, or if pre-built wheels are not available for your platform/Python version.
Fix: Ensure your system has the necessary build tools (e.g., `build-essential` on Linux, Visual Studio on Windows) and Python development packages installed. It also depends on OpenSSL, which might cause issues on some systems with specific versions.
gotcha Direct use of `uamqp` for Azure messaging services like Event Hubs or Service Bus is generally not recommended for most users.
Fix: The Azure SDKs for Python (`azure-eventhub`, `azure-servicebus`) provide a more idiomatic and higher-level API for interacting with these services. `uamqp` is a lower-level library primarily intended for internal SDK use or advanced AMQP scenarios.

Install

pip install uamqp Install latest version

Imports

Connection
```
from uamqp import Connection
```
Message
```
from uamqp import Message
```
SenderLink
```
from uamqp import SenderLink
```
ReceiverLink
```
from uamqp import ReceiverLink
```
authentication
```
from uamqp import authentication
```

Quickstart

This quickstart demonstrates how to establish an AMQP connection, authenticate using SASLPlain, and send a message to a target entity (e.g., an Azure Service Bus queue or Event Hub). It assumes authentication details are provided via environment variables.

import os
import time
import uamqp
from uamqp import authentication, Message, SenderLink, Connection

# Set these environment variables for a runnable example:
# AMQP_ADDRESS: e.g., "amqps://<namespace>.servicebus.windows.net/<entity_name>"
# SAS_POLICY: e.g., "RootManageSharedAccessKey"
# SAS_KEY: The shared access key for the policy
AMQP_ADDRESS = os.environ.get("AMQP_ADDRESS", "amqps://localhost:5672/my_queue")
SAS_POLICY = os.environ.get("SAS_POLICY", "your_sas_policy_name")
SAS_KEY = os.environ.get("SAS_KEY", "your_sas_key")

conn = None
sender = None

try:
    if not all([AMQP_ADDRESS, SAS_POLICY, SAS_KEY]):
        print("Please set AMQP_ADDRESS, SAS_POLICY, and SAS_KEY environment variables.")
        exit(1)

    # Extract hostname (e.g., <namespace>.servicebus.windows.net) from AMQP_ADDRESS
    hostname = AMQP_ADDRESS.split('//', 1)[-1].split('/', 1)[0].split(':', 1)[0]
    # Extract target address (e.g., <entity_name>) from AMQP_ADDRESS
    target_address = AMQP_ADDRESS.split('/', 3)[-1]

    # Create SASL authentication
    sasl_auth = authentication.SASLPlain(
        hostname=hostname,
        username=SAS_POLICY,
        password=SAS_KEY
    )

    # Create an AMQP connection
    print(f"Connecting to {hostname}...")
    conn = Connection(
        hostname,
        sasl_auth=sasl_auth,
        idle_timeout=10000,
        debug=False # Set to True for verbose AMQP tracing
    )

    # Create a SenderLink to send messages
    sender = SenderLink(conn, target_address, debug=False)
    sender.open()
    print(f"SenderLink opened to {target_address}")

    # Create and send a message
    message_body = f"Hello from uAMQP at {time.time()}!"
    message = Message(message_body.encode('utf-8'))
    sender.send_messages([message])
    print(f"Sent message: '{message_body}'")

    time.sleep(1) # Give time for message to be sent

except Exception as e:
    print(f"An error occurred: {e}")
    print("Ensure your environment variables are correctly configured and the AMQP endpoint is accessible.")
finally:
    if sender and sender.is_open:
        sender.close()
        print("SenderLink closed.")
    if conn and conn.is_open:
        conn.close()
        print("Connection closed.")

view raw JSON →