| name | azure-eventgrid-py |
| description | Azure Event Grid SDK for Python. Use for publishing events, handling CloudEvents, and event-driven architectures.
Triggers: "event grid", "EventGridPublisherClient", "CloudEvent", "EventGridEvent", "publish events".
|
| license | MIT |
| metadata | {"author":"Microsoft","version":"1.0.0","package":"azure-eventgrid"} |
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
AZURE_TOKEN_CREDENTIALS=prod
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.
import os
from azure.identity import DefaultAzureCredential, ManagedIdentityCredential
from azure.eventgrid import EventGridPublisherClient
credential = DefaultAzureCredential(require_envvar=True)
endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"
with EventGridPublisherClient(endpoint, credential) as client:
...
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
with EventGridPublisherClient(endpoint, DefaultAzureCredential()) as client:
event = CloudEvent(
type="MyApp.Events.OrderCreated",
source="/myapp/orders",
data={"order_id": "12345", "amount": 99.99}
)
client.send(event)
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",
source="/myapp/items",
data={"key": "value"},
subject="items/123",
datacontenttype="application/json",
dataschema="https://schema.example",
time=datetime.now(timezone.utc),
extensions={"custom": "value"}
)
EventGridEvent Properties
event = EventGridEvent(
subject="/myapp/items/123",
event_type="MyApp.ItemCreated",
data={"key": "value"},
data_version="1.0",
topic="/subscriptions/.../topics/...",
event_time=datetime.now(timezone.utc)
)
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 import EventGridPublisherClient
from azure.identity import DefaultAzureCredential
namespace_endpoint = "https://<namespace>.<region>.eventgrid.azure.net"
topic_name = "my-topic"
with EventGridPublisherClient(
endpoint=namespace_endpoint,
credential=DefaultAzureCredential()
) as client:
client.send(
event,
namespace_topic=topic_name
)
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 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 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