AutoGineer
Generated banner for Azure Eventhub Py
Generated logo for Azure Eventhub Py

Azure Eventhub Py

MicrosoftOfficialCreated by Microsoft

  • Skill
  • Tool integration
  • Cloud and DevOps
  • MIT
  • Low risk
  • Has scripts
Download ZIP

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

  1. Download the skill as a ZIP:

    https://mp-staging.autogineer.com/agent-skills/microsoft/azure-eventhub-py/download

  2. 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:

  1. 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.

  2. 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

ClientPurpose
EventHubProducerClientSend events to Event Hub
EventHubConsumerClientReceive events from Event Hub
BlobCheckpointStoreTrack 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

  1. 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.

  2. 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.

  3. Use DefaultAzureCredential for portable auth across local dev and Azure (avoid connection strings / API keys when possible).

  4. Use batches for sending multiple events

  5. Use checkpoint store in production for reliable processing

  6. Use async client for high-throughput scenarios

  7. Use partition keys for ordered delivery within a partition

  8. Handle batch size limits — catch ValueError when batch is full

  9. Set appropriate consumer groups for different applications

Reference Files

FileContents
references/checkpointing.mdCheckpoint store patterns, blob checkpointing, checkpoint strategies
references/partitions.mdPartition management, load balancing, starting positions
scripts/setup_consumer.pyCLI for Event Hub info, consumer setup, and event sending/receiving

Files

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

OptionDescriptionDefault
blob_account_urlStorage account URLRequired
container_nameBlob container for checkpointsRequired
credentialAuth credentialRequired
api_versionBlob storage API versionLatest

Best Practices

  1. Use dedicated container — Separate checkpoint data from application data

  2. Match consumer groups — Each consumer group needs its own checkpoints

  3. Balance frequency — Too frequent = overhead, too rare = reprocessing on failure

  4. Handle idempotency — Design for at-least-once delivery (events may replay)

  5. Monitor checkpoint lag — Track time between event enqueue and checkpoint

  6. Test failure scenarios — Verify resume works correctly after crashes

  7. Use async store — The async BlobCheckpointStore is more performant

  8. Checkpoint on success only — Don't checkpoint failed events

Checkpoint Frequency Guidelines

ScenarioStrategyTrade-off
Critical data, low volumeEvery eventMax reliability, higher latency
Normal processingEvery 100 eventsGood balance
High throughputEvery 30 secondsLower overhead, more reprocessing
Batch processingEnd of batchNatural 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

AutoGineer is not affiliated with or endorsed by Microsoft.

Download ZIP

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