Azure Event Grid SDK for Python

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

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

1. Use CloudEvents for new applications (industry standard)
2. Batch events when publishing multiple events
3. Include meaningful subjects for filtering
4. Use async client for high-throughput scenarios
5. Handle retries — Event Grid has built-in retry
6. Set appropriate event types for routing and filtering

When to Use

This skill is applicable to execute the workflow or actions described in the overview.

---

<!-- AGI-INTEGRATION-START -->

AGI Framework Integration

Adapted for @techwavedev/​agi-agent-kit Original source: antigravity-awesome-skills

Memory-First Protocol

Retrieve prior deployment configurations, rollback procedures, and incident post-mortems. Avoid re-discovering infrastructure patterns.

# Check for prior infrastructure context before starting
python3 execution/​memory_manager.py auto --query "deployment configuration and patterns for Azure Eventgrid Py"

Storing Results

After completing work, store infrastructure decisions for future sessions:

python3 execution/​memory_manager.py store \
  --content "Deployment pipeline: configured blue-green deployment with health checks on port 8080" \
  --type technical --project <project> \
  --tags azure-eventgrid-py devops

Multi-Agent Collaboration

Broadcast deployment changes so frontend and backend agents update their configurations accordingly.

python3 execution/​cross_agent_context.py store \
  --agent "<your-agent>" \
  --action "Deployed infrastructure changes — updated CI/​CD pipeline with new health check endpoints" \
  --project <project>

Playbook Integration

Use the ship-saas-mvp or full-stack-deploy playbook to sequence this skill with testing, documentation, and deployment verification.

<!-- AGI-INTEGRATION-END -->