azure-servicebus-py
от microsoft
Корпоративный обмен сообщениями для надежной облачной коммуникации с очередями и темами pub/sub.
npx skills add https://github.com/microsoft/skills --skill azure-servicebus-pyAzure Service Bus SDK for Python
Enterprise messaging for reliable cloud communication with queues and pub/sub topics.
Installation
pip install azure-servicebus azure-identity
Environment Variables
SERVICEBUS_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net # Required for all auth methods
SERVICEBUS_QUEUE_NAME=myqueue # Required for queue operations
SERVICEBUS_TOPIC_NAME=mytopic # Required for topic operations
SERVICEBUS_SUBSCRIPTION_NAME=mysubscription # Required for subscription operations
AZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production
Authentication & Lifecycle
🔑 Two rules apply to every code sample below:
- Prefer
DefaultAzureCredential. It works locally (Azure CLI / VS Code / Developer CLI) and in Azure (managed identity, workload identity) with no code change. Avoid connection strings, account/API keys — they bypass Entra audit and rotation.
- Local dev:
DefaultAzureCredentialworks as-is.- Production: set
AZURE_TOKEN_CREDENTIALS=prod(orAZURE_TOKEN_CREDENTIALS=<specific_credential>) to constrain the credential chain to production-safe credentials.- Wrap every client in a context manager so HTTP transports, sockets, and token caches are released deterministically:
- Sync:
with <Client>(...) as client:- Async:
async with <Client>(...) as client:andasync with DefaultAzureCredential() as credential:(fromazure.identity.aio)Snippets may abbreviate this setup, but production code should always follow both rules.
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.servicebus import ServiceBusClient
# Local dev: DefaultAzureCredential. Production: set AZURE_TOKEN_CREDENTIALS=prod or AZURE_TOKEN_CREDENTIALS=<specific_credential>
credential = DefaultAzureCredential(require_envvar=True)
# Or use a specific credential directly in production:
# See https://learn.microsoft.com/python/api/overview/azure/identity-readme?view=azure-python#credential-classes
# credential = ManagedIdentityCredential()
namespace = "<namespace>.servicebus.windows.net"
with ServiceBusClient(
fully_qualified_namespace=namespace,
credential=credential
) as client:
# Use client here (see following sections for operations)
...
Client Types
| Client | Purpose | Get From |
|---|---|---|
ServiceBusClient | Connection management | Direct instantiation |
ServiceBusSender | Send messages | client.get_queue_sender() / get_topic_sender() |
ServiceBusReceiver | Receive messages | client.get_queue_receiver() / get_subscription_receiver() |
Send Messages (Async)
import asyncio
from azure.servicebus.aio import ServiceBusClient
from azure.servicebus import ServiceBusMessage
from azure.identity.aio import DefaultAzureCredential
async def send_messages():
credential = DefaultAzureCredential()
async with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=credential
) as client:
sender = client.get_queue_sender(queue_name="myqueue")
async with sender:
# Single message
message = ServiceBusMessage("Hello, Service Bus!")
await sender.send_messages(message)
# Batch of messages
messages = [ServiceBusMessage(f"Message {i}") for i in range(10)]
await sender.send_messages(messages)
# Message batch (for size control)
batch = await sender.create_message_batch()
for i in range(100):
try:
batch.add_message(ServiceBusMessage(f"Batch message {i}"))
except ValueError: # Batch full
await sender.send_messages(batch)
batch = await sender.create_message_batch()
batch.add_message(ServiceBusMessage(f"Batch message {i}"))
await sender.send_messages(batch)
asyncio.run(send_messages())
Receive Messages (Async)
async def receive_messages():
credential = DefaultAzureCredential()
async with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=credential
) as client:
receiver = client.get_queue_receiver(queue_name="myqueue")
async with receiver:
# Receive batch
messages = await receiver.receive_messages(
max_message_count=10,
max_wait_time=5 # seconds
)
for msg in messages:
print(f"Received: {str(msg)}")
await receiver.complete_message(msg) # Remove from queue
asyncio.run(receive_messages())
Receive Modes
| Mode | Behavior | Use Case |
|---|---|---|
PEEK_LOCK (default) | Message locked, must complete/abandon | Reliable processing |
RECEIVE_AND_DELETE | Removed immediately on receive | At-most-once delivery |
from azure.servicebus import ServiceBusReceiveMode
receiver = client.get_queue_receiver(
queue_name="myqueue",
receive_mode=ServiceBusReceiveMode.RECEIVE_AND_DELETE
)
Message Settlement
async with receiver:
messages = await receiver.receive_messages(max_message_count=1)
for msg in messages:
try:
# Process message...
await receiver.complete_message(msg) # Success - remove from queue
except ProcessingError:
await receiver.abandon_message(msg) # Retry later
except PermanentError:
await receiver.dead_letter_message(
msg,
reason="ProcessingFailed",
error_description="Could not process"
)
| Action | Effect |
|---|---|
complete_message() | Remove from queue (success) |
abandon_message() | Release lock, retry immediately |
dead_letter_message() | Move to dead-letter queue |
defer_message() | Set aside, receive by sequence number |
Topics and Subscriptions
# Send to topic
sender = client.get_topic_sender(topic_name="mytopic")
async with sender:
await sender.send_messages(ServiceBusMessage("Topic message"))
# Receive from subscription
receiver = client.get_subscription_receiver(
topic_name="mytopic",
subscription_name="mysubscription"
)
async with receiver:
messages = await receiver.receive_messages(max_message_count=10)
Sessions (FIFO)
# Send with session
message = ServiceBusMessage("Session message")
message.session_id = "order-123"
await sender.send_messages(message)
# Receive from specific session
receiver = client.get_queue_receiver(
queue_name="session-queue",
session_id="order-123"
)
# Receive from next available session
from azure.servicebus import NEXT_AVAILABLE_SESSION
receiver = client.get_queue_receiver(
queue_name="session-queue",
session_id=NEXT_AVAILABLE_SESSION
)
Scheduled Messages
from datetime import datetime, timedelta, timezone
message = ServiceBusMessage("Scheduled message")
scheduled_time = datetime.now(timezone.utc) + timedelta(minutes=10)
# Schedule message
sequence_number = await sender.schedule_messages(message, scheduled_time)
# Cancel scheduled message
await sender.cancel_scheduled_messages(sequence_number)
Dead-Letter Queue
from azure.servicebus import ServiceBusSubQueue
# Receive from dead-letter queue
dlq_receiver = client.get_queue_receiver(
queue_name="myqueue",
sub_queue=ServiceBusSubQueue.DEAD_LETTER
)
async with dlq_receiver:
messages = await dlq_receiver.receive_messages(max_message_count=10)
for msg in messages:
print(f"Dead-lettered: {msg.dead_letter_reason}")
await dlq_receiver.complete_message(msg)
Sync Client (for simple scripts)
from azure.servicebus import ServiceBusClient, ServiceBusMessage
from azure.identity import DefaultAzureCredential
with ServiceBusClient(
fully_qualified_namespace="<namespace>.servicebus.windows.net",
credential=DefaultAzureCredential()
) as client:
with client.get_queue_sender("myqueue") as sender:
sender.send_messages(ServiceBusMessage("Sync message"))
with client.get_queue_receiver("myqueue") as receiver:
for msg in receiver:
print(str(msg))
receiver.complete_message(msg)
Best Practices
- Pick sync OR async and stay consistent. Do not mix
azure.xxxsync clients withazure.xxx.aioasync clients in the same call path. Choose one mode per module. - Always use context managers for clients and async credentials. Wrap every client in
with Client(...) as client:(sync) orasync with Client(...) as client:(async) for proper cleanup. For asyncDefaultAzureCredentialfromazure.identity.aio, also useasync with credential:so tokens and transports are cleaned up. - Use
DefaultAzureCredentialfor portable auth across local dev and Azure (avoid connection strings / API keys when possible). - Use async client for production workloads
- Complete messages after successful processing
- Use dead-letter queue for poison messages
- Use sessions for ordered, FIFO processing
- Use message batches for high-throughput scenarios
- Set
max_wait_timeto avoid infinite blocking
Reference Files
| File | Contents |
|---|---|
| references/patterns.md | Competing consumers, sessions, retry patterns, request-response, transactions |
| references/dead-letter.md | DLQ handling, poison messages, reprocessing strategies |
| scripts/setup_servicebus.py | CLI for queue/topic/subscription management and DLQ monitoring |
Больше skills от microsoft
oss-growth
microsoft
Персона OSS-хакера роста
official
microsoft-foundry
microsoft
Развёртывание, оценка и управление агентами Foundry «под ключ»: сборка Docker, отправка в ACR, создание хостируемых и промпт-агентов, запуск контейнера, пакетная оценка, непрерывная оценка, оптимизатор промптов, agent.yaml, курирование датасетов из трейсов. ИСПОЛЬЗУЙТЕ ДЛЯ: развёртывания агента в Foundry, хостируемого агента, создания агента, вызова агента, оценки агента, запуска пакетной оценки, непрерывной оценки, непрерывного мониторинга, статуса непрерывной оценки, оптимизации промпта, улучшения промпта, оптимизатора промптов, оптимизации инструкций агента, улучшения агента...
officialdevelopmentdevops
azure-ai
microsoft
Используется для Azure AI: поиск, речь, OpenAI, анализ документов. Помогает с поиском, векторным/гибридным поиском, преобразованием речи в текст, синтезом речи, транскрипцией, OCR. КОГДА: AI Search, поиск по запросу, векторный поиск, гибридный поиск, семантический поиск, преобразование речи в текст, синтез речи, транскрибирование, OCR, преобразование текста в речь.
officialdevelopmentapi
azure-deploy
microsoft
Выполнение развертываний Azure для УЖЕ ПОДГОТОВЛЕННЫХ приложений, имеющих существующие файлы .azure/deployment-plan.md и инфраструктуры. НЕ используйте этот навык, когда пользователь просит СОЗДАТЬ новое приложение — используйте azure-prepare. Этот навык выполняет команды azd up, azd deploy, terraform apply и az deployment со встроенным восстановлением после ошибок. Требует .azure/deployment-plan.md от azure-prepare и подтвержденный статус от azure-validate. КОГДА: "запустить azd up", "запустить azd deploy", "выполнить развертывание",...
officialdevopsaws
azure-storage
microsoft
Сервисы Azure Storage, включая Blob Storage, File Shares, Queue Storage, Table Storage и Data Lake. Отвечает на вопросы об уровнях доступа к хранилищу (горячий, холодный, холодный, архивный), когда использовать каждый уровень и сравнение уровней. Предоставляет объектное хранилище, SMB-файловые ресурсы, асинхронный обмен сообщениями, NoSQL-ключ-значение и аналитику больших данных. Включает управление жизненным циклом. ИСПОЛЬЗОВАТЬ ДЛЯ: хранилища BLOB-объектов, файловых ресурсов, хранилища очередей, табличного хранилища, Data Lake, загрузки файлов, скачивания BLOB-объектов, учетных записей хранения, уровней доступа,...
officialdevelopmentdatabase
azure-diagnostics
microsoft
Отладка проблем Azure в рабочей среде с помощью AppLens, Azure Monitor, работоспособности ресурсов и безопасной триаж. КОГДА: отладка проблем в рабочей среде, устранение неполадок службы приложений, высокая загрузка ЦП службы приложений, сбой развертывания службы приложений, устранение неполадок контейнерных приложений, устранение неполадок функций, устранение неполадок AKS, kubectl не может подключиться, сбои kube-system/CoreDNS, pod в состоянии ожидания, crashloop, узел не готов, сбои обновления, анализ журналов, KQL, аналитика, сбои извлечения образов, проблемы холодного запуска, сбои проверки работоспособности,...
officialdevopsdevelopment
azure-prepare
microsoft
Подготовка приложений Azure к развертыванию (инфра Bicep/Terraform, azure.yaml, Dockerfiles). Используйте для создания/модернизации или создания+развертывания; не для межоблачной миграции (используйте azure-cloud-migrate). НЕ ИСПОЛЬЗУЙТЕ ДЛЯ: приложений copilot-sdk (используйте azure-hosted-copilot-sdk). КОГДА: "создать приложение", "создать веб-приложение", "создать API", "создать бессерверный HTTP API", "создать фронтенд", "создать бэкенд", "собрать сервис", "модернизировать приложение", "обновить приложение", "добавить аутентификацию", "добавить кэширование", "разместить в Azure", "создать и...
officialdevelopmentdevops
azure-validate
microsoft
Предварительная проверка развертывания на готовность Azure. Выполняет глубокие проверки конфигурации, инфраструктуры (Bicep или Terraform), назначений ролей RBAC, разрешений управляемых удостоверений и предварительных требований перед развертыванием. КОГДА: проверить мое приложение, проверить готовность к развертыванию, выполнить предварительные проверки, проверить конфигурацию, проверить готовность к развертыванию, проверить azure.yaml, проверить Bicep, протестировать перед развертыванием, устранить ошибки развертывания, проверить Azure Functions, проверить приложение-функцию, проверить бессерверное...
officialdevopstesting