Azure Eventgrid Py

作者 microsoft354361d83247MIT收录于 2026年10月8日更新于 2026年10月8日

Azure Event Grid SDK for Python. Use for publishing events, handling CloudEvents, and event-driven architectures. Triggers: "event grid", "EventGridPublisherClient", "CloudEvent", "EventGridEvent", "publish events".

精选仅含说明Software Development

AI 生成的概览

指导 Python 开发者使用 Azure Event Grid SDK 发布 CloudEvents 和 EventGridEvents。

功能: 该技能为 Azure Event Grid Python SDK 提供参考说明和代码示例。内容涵盖安装、使用 DefaultAzureCredential 进行身份验证、发布 CloudEvents 和 EventGridEvents、使用同步与异步客户端以及命名空间主题。还包含事件驱动架构的最佳实践。
适用场景: 适用于在 Python 中构建事件驱动应用程序并向 Azure Event Grid 发布事件的场景。适合需要处理 CloudEvents 或 EventGridEvents、配置身份验证或使用 Event Grid 命名空间的开发者。
运行要求: 需要安装 Python 以及 azure-eventgrid 和 azure-identity 包。需要 Azure Event Grid 主题或命名空间端点以及适当的 Azure 凭据（如 DefaultAzureCredential）。需要访问 Azure 的网络。该技能仅包含说明和参考文件，没有脚本。

Azure Event Grid SDK for Python

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

Installation

Environment Variables

bash

EVENTGRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events  # Required for Event Grid topic publishingEVENTGRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net  # Required for namespace operationsAZURE_TOKEN_CREDENTIALS=prod # Required only if DefaultAzureCredential is used in production

EVENTGRID_TOPIC_ENDPOINT=https://<topic-name>.<region>.eventgrid.azure.net/api/events  # Required for Event Grid topic publishingEVENTGRID_NAMESPACE_ENDPOINT=https://<namespace>.<region>.eventgrid.azure.net  # Required for namespace operationsAZURE_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.

python

import osfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.eventgrid import EventGridPublisherClient
# 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()
endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"
with EventGridPublisherClient(endpoint, credential) as client:    # Use client here (see following sections for operations)    ...

import osfrom azure.identity import DefaultAzureCredential, ManagedIdentityCredentialfrom azure.eventgrid import EventGridPublisherClient
# 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()
endpoint = "https://<topic-name>.<region>.eventgrid.azure.net/api/events"
with EventGridPublisherClient(endpoint, credential) as client:    # Use client here (see following sections for operations)    ...

Event Types

Publish CloudEvents

python

from azure.eventgrid import EventGridPublisherClient, CloudEventfrom azure.identity import DefaultAzureCredential
with EventGridPublisherClient(endpoint, DefaultAzureCredential()) as client:    # 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)

from azure.eventgrid import EventGridPublisherClient, CloudEventfrom azure.identity import DefaultAzureCredential
with EventGridPublisherClient(endpoint, DefaultAzureCredential()) as client:    # 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

python

from azure.eventgrid import EventGridEventfrom 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)

from azure.eventgrid import EventGridEventfrom 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

python

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)

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

python

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)

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

python

from azure.eventgrid.aio import EventGridPublisherClientfrom 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 asyncioasyncio.run(publish_events())

from azure.eventgrid.aio import EventGridPublisherClientfrom 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 asyncioasyncio.run(publish_events())

Namespace Topics (Event Grid Namespaces)

For Event Grid Namespaces (pull delivery):

python

from azure.eventgrid import EventGridPublisherClientfrom azure.identity import DefaultAzureCredential
# Namespace endpoint (different from custom topic)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    )

from azure.eventgrid import EventGridPublisherClientfrom azure.identity import DefaultAzureCredential
# Namespace endpoint (different from custom topic)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

Reference Files

File	Contents
references/capabilities.md [blocked]	Additional non-hero capabilities, operation-group coverage, and production checklists.
references/non-hero-scenarios.md [blocked]	Dedicated non-hero examples for secondary/advanced scenarios.

来源与署名

来源：microsoft/skills位于.github/plugins/azure-sdk-python/skills/azure-eventgrid-py提交354361d

许可证： MIT

内容归原作者所有。SourceWeft 从公开仓库中收录这些内容。

举报或申请下架

Format	Class	Use Case
Cloud Events 1.0	`CloudEvent`	Standard, interoperable (recommended)
Event Grid Schema	`EventGridEvent`	Azure-native format

Azure Eventgrid Py

Azure Event Grid SDK for Python

Installation

Environment Variables

Authentication & Lifecycle

Event Types

Publish CloudEvents

Publish EventGridEvents

Event Properties

CloudEvent Properties

EventGridEvent Properties

Async Client

Namespace Topics (Event Grid Namespaces)

Best Practices

Reference Files

来源与署名

更多来自 microsoft/skills 的技能

Customize

Discover Azure Skills

Azure Resource Visualizer

Azure Resource Lookup

Azure Messaging

Azure Kusto

更多Software Development技能

Playground

M5 Onboard

Cardputer Buddy

Testing Strategy

Tech Debt

System Design