azure-servicebus-py
作者: microsoft
用于可靠云通信的企业级消息传递,支持队列和发布/订阅主题。
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 |
来自 microsoft 的更多技能
oss-growth
microsoft
OSS增长黑客角色
official
microsoft-foundry
microsoft
端到端部署、评估和管理Foundry代理:Docker构建、ACR推送、托管/提示代理创建、容器启动、批量评估、持续评估、提示优化工作流、agent.yaml、从追踪中整理数据集。用途:将代理部署到Foundry、托管代理、创建代理、调用代理、评估代理、运行批量评估、持续评估、持续监控、持续评估状态、优化提示、改进提示、提示优化器、优化代理指令、改进代理...
officialdevelopmentdevops
azure-ai
microsoft
用于Azure AI:搜索、语音、OpenAI、文档智能。支持搜索、向量/混合搜索、语音转文字、文字转语音、转录、OCR。适用场景:AI搜索、查询搜索、向量搜索、混合搜索、语义搜索、语音转文字、文字转语音、转录、OCR、文字转语音。
officialdevelopmentapi
azure-deploy
microsoft
对已准备好的应用程序执行Azure部署,这些程序需包含现有的.azure/deployment-plan.md和基础设施文件。当用户要求创建新应用程序时,请勿使用此技能——应改用azure-prepare。此技能运行azd up、azd deploy、terraform apply和az deployment命令,并内置错误恢复机制。需要来自azure-prepare的.azure/deployment-plan.md以及来自azure-validate的已验证状态。适用场景:"运行azd up"、"运行azd deploy"、"执行部署"...
officialdevopsaws
azure-storage
microsoft
Azure存储服务,包括Blob存储、文件共享、队列存储、表存储和Data Lake。解答关于存储访问层(热、冷、冷、归档)的问题,说明各层的使用场景及对比。提供对象存储、SMB文件共享、异步消息传递、NoSQL键值存储和大数据分析。包含生命周期管理。用途:Blob存储、文件共享、队列存储、表存储、Data Lake、上传文件、下载Blob、存储账户、访问层等。
officialdevelopmentdatabase
azure-diagnostics
microsoft
使用AppLens、Azure Monitor、资源健康和安全分类调试Azure生产问题。适用场景:调试生产问题、排查应用服务、应用服务CPU过高、应用服务部署失败、排查容器应用、排查函数、排查AKS、kubectl无法连接、kube-system/CoreDNS故障、Pod挂起、CrashLoop、节点未就绪、升级失败、分析日志、KQL、洞察、镜像拉取失败、冷启动问题、健康探测失败……
officialdevopsdevelopment
azure-prepare
microsoft
为Azure应用准备部署(基础设施Bicep/Terraform、azure.yaml、Dockerfile)。用于创建/现代化或创建+部署;不用于跨云迁移(使用azure-cloud-migrate)。请勿用于:copilot-sdk应用(使用azure-hosted-copilot-sdk)。适用场景:"创建应用"、"构建Web应用"、"创建API"、"创建无服务器HTTP API"、"创建前端"、"创建后端"、"构建服务"、"现代化应用"、"更新应用"、"添加身份验证"、"添加缓存"、"托管在Azure上"、"创建并...
officialdevelopmentdevops
azure-validate
microsoft
部署前对Azure就绪状态进行验证。对配置、基础设施(Bicep或Terraform)、RBAC角色分配、托管标识权限及先决条件进行深度检查,然后再部署。适用场景:验证我的应用、检查部署就绪状态、运行预检、验证配置、检查是否可部署、验证azure.yaml、验证Bicep、部署前测试、排查部署错误、验证Azure Functions、验证函数应用、验证无服务器...
officialdevopstesting