Azure Event Grid SDK for Python

Event routing service for building event-driven applications with pub/sub semantics.

Installation

pip install azure-eventgrid azure-identity

Environment Variables

EVENTGRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events EVENTGRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net

Authentication

from azure.identity import DefaultAzureCredential from azure.eventgrid import EventGridPublisherClient

credential = DefaultAzureCredential() endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"

client = EventGridPublisherClient(endpoint, credential)

Event Types

Format Class Use Case

Cloud Events 1.0 CloudEvent

Standard, interoperable (recommended)

Event Grid Schema EventGridEvent

Azure-native format

Publish CloudEvents

from azure.eventgrid import EventGridPublisherClient, CloudEvent from azure.identity import DefaultAzureCredential

client = EventGridPublisherClient(endpoint, DefaultAzureCredential())

Single event

event = CloudEvent( type="MyApp.Events.OrderCreated", source="/myapp/orders", data={"order_id": "12345", "amount": 99.99} ) client.send(event)

Multiple events

events = [ CloudEvent( type="MyApp.Events.OrderCreated", source="/myapp/orders", data={"order_id": f"order-{i}"} ) for i in range(10) ] client.send(events)

Publish EventGridEvents

from azure.eventgrid import EventGridEvent from datetime import datetime, timezone

event = EventGridEvent( subject="/myapp/orders/12345", event_type="MyApp.Events.OrderCreated", data={"order_id": "12345", "amount": 99.99}, data_version="1.0" )

client.send(event)

Event Properties

CloudEvent Properties

event = CloudEvent( type="MyApp.Events.ItemCreated", # Required: event type source="/myapp/items", # Required: event source data={"key": "value"}, # Event payload subject="items/123", # Optional: subject/path datacontenttype="application/json", # Optional: content type dataschema="https://schema.example", # Optional: schema URL time=datetime.now(timezone.utc), # Optional: timestamp extensions={"custom": "value"} # Optional: custom attributes )

EventGridEvent Properties

event = EventGridEvent( subject="/myapp/items/123", # Required: subject event_type="MyApp.ItemCreated", # Required: event type data={"key": "value"}, # Required: event payload data_version="1.0", # Required: schema version topic="/subscriptions/.../topics/...", # Optional: auto-set event_time=datetime.now(timezone.utc) # Optional: timestamp )

Async Client

from azure.eventgrid.aio import EventGridPublisherClient from azure.identity.aio import DefaultAzureCredential

async def publish_events(): credential = DefaultAzureCredential()

async with EventGridPublisherClient(endpoint, credential) as client:
    event = CloudEvent(
        type="MyApp.Events.Test",
        source="/myapp",
        data={"message": "hello"}
    )
    await client.send(event)

import asyncio asyncio.run(publish_events())

Namespace Topics (Event Grid Namespaces)

For Event Grid Namespaces (pull delivery):

from azure.eventgrid.aio import EventGridPublisherClient

Namespace endpoint (different from custom topic)

namespace_endpoint = "https://<namespace>.<region>.eventgrid.azure.net" topic_name = "my-topic"

async with EventGridPublisherClient( endpoint=namespace_endpoint, credential=DefaultAzureCredential() ) as client: await client.send( event, namespace_topic=topic_name )

Best Practices

• Use CloudEvents for new applications (industry standard)

• Batch events when publishing multiple events

• Include meaningful subjects for filtering

• Use async client for high-throughput scenarios

• Handle retries — Event Grid has built-in retry

• Set appropriate event types for routing and filtering