Azure Eventhub Py

MicrosoftOfficialCreated by Microsoft

Skill
Tool integration
Cloud and DevOps
MIT
Low risk
Has scripts

From the author

Azure Event Hubs SDK for Python streaming. Use for high-throughput event ingestion, producers, consumers, and checkpointing. Triggers: "event hubs", "EventHubProducerClient", "EventHubConsumerClient", "streaming", "partitions".

— the description by Microsoft

AutoGineer summary

Written by AutoGineer from the skill’s own files, not by the author

The azure-eventhub-py skill provides the Python SDK for Azure Event Hubs. It enables developers to build producers and consumers for event streaming, manage partitions, and implement checkpointing using Azure Blob Storage. The skill includes guidance on authentication using DefaultAzureCredential, managing client lifecycles with context managers, and strategies for handling event batches and partition distribution.

What it does

Provides clients for producing and consuming events from Azure Event Hubs
Supports both synchronous and asynchronous programming models
Enables checkpointing of consumer progress using Azure Blob Storage
Facilitates partition management including direct, key-based, and round-robin routing
Offers methods to retrieve Event Hub and partition metadata
Integrates with Azure Identity for secure authentication

When to use it

When integrating Python applications with Azure Event Hubs
When building event-driven architectures requiring high-throughput ingestion
When implementing reliable consumer patterns with stateful checkpointing
When needing to distribute event processing across multiple consumers

Advantages

Supports consistent authentication patterns across local and cloud environments
Provides flexible partition selection strategies for ordered or distributed delivery
Includes built-in support for checkpointing to track processing progress
Offers consistent API design for both sync and async operations

Usage examples

“How do I send events to an Azure Event Hub using Python?”: The agent provides a code example using EventHubProducerClient, demonstrating how to create a batch, add events, and send them to the hub.
“How can I implement a consumer that tracks progress with a checkpoint store?”: The agent provides a code snippet using EventHubConsumerClient configured with a BlobCheckpointStore to update checkpoints after processing events.
“How do I route events to a specific partition in Event Hubs?”: The agent explains how to use partition_id or partition_key within the create_batch method to control event distribution.
“What is the recommended way to authenticate with Azure Event Hubs?”: The agent explains the use of DefaultAzureCredential to handle authentication across local development and production environments.

Install

Download the skill as a ZIP:
https://autogineer.com/agent-skills/microsoft/azure-eventhub-py/download

Unzip it into the cross-client skills folder of your project:

mkdir -p .agents/skills && unzip azure-eventhub-py.zip -d .agents/skills

Documentation by Microsoft

Azure Event Hubs SDK for Python

Big data streaming platform for high-throughput event ingestion.

Installation

pip install azure-eventhub azure-identity
# For checkpointing with blob storage
pip install azure-eventhub-checkpointstoreblob-aio

Environment Variables

EVENT_HUB_FULLY_QUALIFIED_NAMESPACE=<namespace>.servicebus.windows.net  # Required for all auth methods
EVENT_HUB_NAME=my-eventhub  # Required for all auth methods
STORAGE_ACCOUNT_URL=https://<account>.blob.core.windows.net  # Required for checkpoint storage
CHECKPOINT_CONTAINER=checkpoints  # Required for checkpoint storage
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: DefaultAzureCredential works as-is.
Production: set AZURE_TOKEN_CREDENTIALS=prod (or AZURE_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: and async with DefaultAzureCredential() as credential: (from azure.identity.aio)
Snippets may abbreviate this setup, but production code should always follow both rules.

from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.eventhub import EventHubProducerClient, EventHubConsumerClient

# 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"
eventhub_name = "my-eventhub"

# Producer
with EventHubProducerClient(
    fully_qualified_namespace=namespace,
    eventhub_name=eventhub_name,
    credential=credential
) as producer:
    # Use producer here (see following sections for operations)
    ...

# Consumer
with EventHubConsumerClient(
    fully_qualified_namespace=namespace,
    eventhub_name=eventhub_name,
    consumer_group="$Default",
    credential=credential
) as consumer:
    # Use consumer here (see following sections for operations)
    ...

Client Types

Client	Purpose
`EventHubProducerClient`	Send events to Event Hub
`EventHubConsumerClient`	Receive events from Event Hub
`BlobCheckpointStore`	Track consumer progress

Send Events

from azure.eventhub import EventHubProducerClient, EventData
from azure.identity import DefaultAzureCredential

with EventHubProducerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    credential=DefaultAzureCredential()
) as producer:
    # Create batch (handles size limits)
    event_data_batch = producer.create_batch()
    
    for i in range(10):
        try:
            event_data_batch.add(EventData(f"Event {i}"))
        except ValueError:
            # Batch is full, send and create new one
            producer.send_batch(event_data_batch)
            event_data_batch = producer.create_batch()
            event_data_batch.add(EventData(f"Event {i}"))
    
    # Send remaining
    producer.send_batch(event_data_batch)

Send to Specific Partition

# By partition ID
event_data_batch = producer.create_batch(partition_id="0")

# By partition key (consistent hashing)
event_data_batch = producer.create_batch(partition_key="user-123")

Receive Events

Simple Receive

from azure.eventhub import EventHubConsumerClient

def on_event(partition_context, event):
    print(f"Partition: {partition_context.partition_id}")
    print(f"Data: {event.body_as_str()}")
    partition_context.update_checkpoint(event)

with EventHubConsumerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    consumer_group="$Default",
    credential=DefaultAzureCredential()
) as consumer:
    consumer.receive(
        on_event=on_event,
        starting_position="-1",  # Beginning of stream
    )

With Blob Checkpoint Store (Production)

from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
from azure.identity import DefaultAzureCredential

checkpoint_store = BlobCheckpointStore(
    blob_account_url="https://<account>.blob.core.windows.net",
    container_name="checkpoints",
    credential=DefaultAzureCredential()
)

with EventHubConsumerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    consumer_group="$Default",
    credential=DefaultAzureCredential(),
    checkpoint_store=checkpoint_store
) as consumer:
    def on_event(partition_context, event):
        print(f"Received: {event.body_as_str()}")
        # Checkpoint after processing
        partition_context.update_checkpoint(event)

    consumer.receive(on_event=on_event)

Async Client

from azure.eventhub.aio import EventHubProducerClient, EventHubConsumerClient
from azure.identity.aio import DefaultAzureCredential
import asyncio

async def send_events():
    credential = DefaultAzureCredential()
    
    async with EventHubProducerClient(
        fully_qualified_namespace="<namespace>.servicebus.windows.net",
        eventhub_name="my-eventhub",
        credential=credential
    ) as producer:
        batch = await producer.create_batch()
        batch.add(EventData("Async event"))
        await producer.send_batch(batch)

async def receive_events():
    async def on_event(partition_context, event):
        print(event.body_as_str())
        await partition_context.update_checkpoint(event)
    
    async with EventHubConsumerClient(
        fully_qualified_namespace="<namespace>.servicebus.windows.net",
        eventhub_name="my-eventhub",
        consumer_group="$Default",
        credential=DefaultAzureCredential()
    ) as consumer:
        await consumer.receive(on_event=on_event)

asyncio.run(send_events())

Event Properties

event = EventData("My event body")

# Set properties
event.properties = {"custom_property": "value"}
event.content_type = "application/json"

# Read properties (on receive)
print(event.body_as_str())
print(event.sequence_number)
print(event.offset)
print(event.enqueued_time)
print(event.partition_key)

Get Event Hub Info

with producer:
    info = producer.get_eventhub_properties()
    print(f"Name: {info['name']}")
    print(f"Partitions: {info['partition_ids']}")
    
    for partition_id in info['partition_ids']:
        partition_info = producer.get_partition_properties(partition_id)
        print(f"Partition {partition_id}: {partition_info['last_enqueued_sequence_number']}")

Best Practices

Pick sync OR async and stay consistent. Do not mix azure.xxx sync clients with azure.xxx.aio async 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) or async with Client(...) as client: (async) for proper cleanup. For async DefaultAzureCredential from azure.identity.aio, also use async with credential: so tokens and transports are cleaned up.
Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).
Use batches for sending multiple events
Use checkpoint store in production for reliable processing
Use async client for high-throughput scenarios
Use partition keys for ordered delivery within a partition
Handle batch size limits — catch ValueError when batch is full
Set appropriate consumer groups for different applications

Reference Files

File	Contents
references/checkpointing.md	Checkpoint store patterns, blob checkpointing, checkpoint strategies
references/partitions.md	Partition management, load balancing, starting positions
scripts/setup_consumer.py	CLI for Event Hub info, consumer setup, and event sending/receiving

Files

ATTRIBUTION.mdAdded by AutoGineer962 B
LICENSELicence file added by AutoGineer1.1 KiB
SKILL.md9.1 KiB
references/checkpointing.md9.8 KiB
references/partitions.md10.4 KiB
scripts/setup_consumer.pyScript13.3 KiB

references/checkpointing.md

Checkpointing with Azure Event Hubs

Patterns for reliable event processing with checkpoint stores.

Why Checkpointing?

Checkpointing tracks which events have been processed, enabling:

Resume after failure — Pick up where you left off
Scalable consumers — Multiple consumers share work without duplication
At-least-once delivery — Ensure no events are lost

Blob Checkpoint Store (Recommended)

from azure.eventhub import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore
from azure.identity import DefaultAzureCredential

credential = DefaultAzureCredential()

# Create checkpoint store
checkpoint_store = BlobCheckpointStore(
    blob_account_url="https://<account>.blob.core.windows.net",
    container_name="checkpoints",
    credential=credential
)

# Consumer with checkpoint store
consumer = EventHubConsumerClient(
    fully_qualified_namespace="<namespace>.servicebus.windows.net",
    eventhub_name="my-eventhub",
    consumer_group="$Default",
    credential=credential,
    checkpoint_store=checkpoint_store
)

Async Blob Checkpoint Store

from azure.eventhub.aio import EventHubConsumerClient
from azure.eventhub.extensions.checkpointstoreblob.aio import BlobCheckpointStore
from azure.identity.aio import DefaultAzureCredential

async def create_consumer():
    credential = DefaultAzureCredential()
    
    checkpoint_store = BlobCheckpointStore(
        blob_account_url="https://<account>.blob.core.windows.net",
        container_name="checkpoints",
        credential=credential
    )
    
    consumer = EventHubConsumerClient(
        fully_qualified_namespace="<namespace>.servicebus.windows.net",
        eventhub_name="my-eventhub",
        consumer_group="$Default",
        credential=credential,
        checkpoint_store=checkpoint_store
    )
    
    return consumer

Checkpoint Strategies

After Every Event (Most Reliable)

async def on_event(partition_context, event):
    """Checkpoint after every event - highest reliability, highest overhead."""
    try:
        await process_event(event)
        await partition_context.update_checkpoint(event)
    except Exception as e:
        # Don't checkpoint on failure - event will be reprocessed
        print(f"Processing failed: {e}")

Batch Checkpointing (Balanced)

class BatchCheckpointer:
    def __init__(self, batch_size: int = 100):
        self.batch_size = batch_size
        self.counts: dict[str, int] = {}
    
    async def on_event(self, partition_context, event):
        partition_id = partition_context.partition_id
        
        await process_event(event)
        
        self.counts[partition_id] = self.counts.get(partition_id, 0) + 1
        
        if self.counts[partition_id] >= self.batch_size:
            await partition_context.update_checkpoint(event)
            self.counts[partition_id] = 0
            print(f"Checkpointed partition {partition_id}")

Time-Based Checkpointing

import asyncio
from datetime import datetime, timedelta

class TimedCheckpointer:
    def __init__(self, interval_seconds: float = 30.0):
        self.interval = interval_seconds
        self.last_checkpoint: dict[str, datetime] = {}
        self.last_event: dict[str, any] = {}
    
    async def on_event(self, partition_context, event):
        partition_id = partition_context.partition_id
        now = datetime.utcnow()
        
        await process_event(event)
        self.last_event[partition_id] = event
        
        last = self.last_checkpoint.get(partition_id)
        if not last or (now - last).total_seconds() >= self.interval:
            await partition_context.update_checkpoint(event)
            self.last_checkpoint[partition_id] = now
            print(f"Timed checkpoint for partition {partition_id}")

Hybrid Checkpointing (Batch + Time)

class HybridCheckpointer:
    def __init__(self, batch_size: int = 100, interval_seconds: float = 30.0):
        self.batch_size = batch_size
        self.interval = interval_seconds
        self.counts: dict[str, int] = {}
        self.last_checkpoint: dict[str, datetime] = {}
    
    async def on_event(self, partition_context, event):
        partition_id = partition_context.partition_id
        now = datetime.utcnow()
        
        await process_event(event)
        
        self.counts[partition_id] = self.counts.get(partition_id, 0) + 1
        last = self.last_checkpoint.get(partition_id)
        
        should_checkpoint = (
            self.counts[partition_id] >= self.batch_size or
            (last and (now - last).total_seconds() >= self.interval)
        )
        
        if should_checkpoint:
            await partition_context.update_checkpoint(event)
            self.counts[partition_id] = 0
            self.last_checkpoint[partition_id] = now

Checkpoint on Batch Complete

async def on_event_batch(partition_context, events):
    """Process batch and checkpoint once at the end."""
    if not events:
        return
    
    for event in events:
        await process_event(event)
    
    # Checkpoint only the last event
    await partition_context.update_checkpoint(events[-1])
    print(f"Processed {len(events)} events, checkpointed partition {partition_context.partition_id}")

async with consumer:
    await consumer.receive_batch(
        on_event_batch=on_event_batch,
        max_batch_size=100,
        max_wait_time=5.0
    )

Manual Checkpoint Management

from azure.eventhub.extensions.checkpointstoreblob import BlobCheckpointStore

async def inspect_checkpoints(checkpoint_store: BlobCheckpointStore):
    """List all checkpoints for debugging."""
    checkpoints = await checkpoint_store.list_checkpoints(
        fully_qualified_namespace="<namespace>.servicebus.windows.net",
        eventhub_name="my-eventhub",
        consumer_group="$Default"
    )
    
    for cp in checkpoints:
        print(f"Partition: {cp['partition_id']}")
        print(f"  Offset: {cp['offset']}")
        print(f"  Sequence: {cp['sequence_number']}")

Checkpoint Data Structure

Blob checkpoint stores data in this format:

Container: checkpoints
└── <namespace>/<eventhub>/<consumer-group>/checkpoint/
    ├── 0  (partition 0 checkpoint)
    ├── 1  (partition 1 checkpoint)
    └── 2  (partition 2 checkpoint)

Each checkpoint blob contains:

{
    "offset": "12345",
    "sequence_number": 100
}

Graceful Shutdown with Final Checkpoint

import signal
import asyncio

class GracefulConsumer:
    def __init__(self, consumer):
        self.consumer = consumer
        self.running = True
        self.last_events: dict[str, any] = {}
        self.partition_contexts: dict[str, any] = {}
    
    async def on_event(self, partition_context, event):
        if not self.running:
            return
        
        await process_event(event)
        
        # Track for final checkpoint
        self.last_events[partition_context.partition_id] = event
        self.partition_contexts[partition_context.partition_id] = partition_context
    
    async def shutdown(self):
        """Checkpoint all partitions before stopping."""
        self.running = False
        
        for partition_id, event in self.last_events.items():
            context = self.partition_contexts[partition_id]
            await context.update_checkpoint(event)
            print(f"Final checkpoint for partition {partition_id}")
        
        await self.consumer.close()

# Usage
consumer = GracefulConsumer(client)

def signal_handler(sig, frame):
    asyncio.create_task(consumer.shutdown())

signal.signal(signal.SIGTERM, signal_handler)
signal.signal(signal.SIGINT, signal_handler)

Error Handling and Retry

async def on_event_with_retry(partition_context, event, max_retries: int = 3):
    """Process with retries, only checkpoint on success."""
    for attempt in range(max_retries):
        try:
            await process_event(event)
            await partition_context.update_checkpoint(event)
            return
        except TransientError as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
                continue
            raise
        except PermanentError as e:
            # Log and checkpoint to skip - event cannot be processed
            print(f"Permanent failure, skipping: {e}")
            await partition_context.update_checkpoint(event)
            return

Checkpoint Store Configuration

Option	Description	Default
`blob_account_url`	Storage account URL	Required
`container_name`	Blob container for checkpoints	Required
`credential`	Auth credential	Required
`api_version`	Blob storage API version	Latest

Best Practices

Use dedicated container — Separate checkpoint data from application data
Match consumer groups — Each consumer group needs its own checkpoints
Balance frequency — Too frequent = overhead, too rare = reprocessing on failure
Handle idempotency — Design for at-least-once delivery (events may replay)
Monitor checkpoint lag — Track time between event enqueue and checkpoint
Test failure scenarios — Verify resume works correctly after crashes
Use async store — The async BlobCheckpointStore is more performant
Checkpoint on success only — Don't checkpoint failed events

Checkpoint Frequency Guidelines

Scenario	Strategy	Trade-off
Critical data, low volume	Every event	Max reliability, higher latency
Normal processing	Every 100 events	Good balance
High throughput	Every 30 seconds	Lower overhead, more reprocessing
Batch processing	End of batch	Natural fit for batch workloads

Requirements and compatibility

Works with: Claude Code, Claude apps, OpenAI Codex, GitHub Copilot, Cursor, Gemini CLI, Kiro, OpenCode, Goose, Amp, Roo Code, JetBrains Junie, Factory Droid, Devin (ex-Windsurf), Mistral Vibe, OpenHands, Letta

“Works with” means the tool reads the open Agent Skills format; a skill written for one tool may use features others lack.

Needs: Needs an API key, Needs Python

Security

What our scanner noticed· Low risk

Instructions that use the network
Scripts that your agent may run

This skill includes scripts that your agent may run. Read them before you use it.

A static scan is not a guarantee. Skills are third-party content: review a skill before you let an agent use it.

Attribution and licence

Created by

Microsoft

Publisher

Microsoft (Official)

Source

microsoft/skills · .github/plugins/azure-sdk-python/skills/azure-eventhub-py · commit 0696600

Licence

MIT License (redistribution allowed)

LICENSE (added by AutoGineer from LICENSE)
ATTRIBUTION.md (written by AutoGineer)

Synced

2026-11-09

AutoGineer is not affiliated with or endorsed by Microsoft or Microsoft.

Download ZIP

The ZIP holds the skill’s files, its licence files and an ATTRIBUTION.md. Download azure-eventhub-py.zip

From the author

AutoGineer summary

What it does

When to use it

Advantages

Usage examples

Install

Documentation by Microsoft

Azure Event Hubs SDK for Python

Installation

Environment Variables

Authentication & Lifecycle

Client Types

Send Events

Send to Specific Partition

Receive Events

Simple Receive

With Blob Checkpoint Store (Production)

Async Client

Event Properties

Get Event Hub Info

Best Practices

Reference Files

Files

references/checkpointing.md

Checkpointing with Azure Event Hubs

Why Checkpointing?

Blob Checkpoint Store (Recommended)

Async Blob Checkpoint Store

Checkpoint Strategies

After Every Event (Most Reliable)

Batch Checkpointing (Balanced)

Time-Based Checkpointing

Hybrid Checkpointing (Batch + Time)

Checkpoint on Batch Complete

Manual Checkpoint Management

Checkpoint Data Structure

Graceful Shutdown with Final Checkpoint

Error Handling and Retry

Checkpoint Store Configuration

Best Practices

Checkpoint Frequency Guidelines

Requirements and compatibility

Security

Attribution and licence

Related skills

Modal

Cloud Infrastructure Network Engineer

Deployment Validation Cloud Architect

Airflow Dag Patterns

Cost Optimization

Hybrid Cloud Networking

Download ZIP