ReddRadar
Agent skill
azure-eventgrid-py
Azure Event Grid SDK for Python. Use for publishing events, handling CloudEvents, and event-driven architectures. Triggers: "event grid", "EventGridPublisherClient", "CloudEvent", "EventGridEvent", "publish events".
Install this agent skill to your Project
npx add-skill https://github.com/aiskillstore/marketplace/tree/main/skills/sickn33/azure-eventgrid-py
SKILL.md
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
Recommended Agent Skills
Expand your agent's capabilities with these related and highly-rated skills.
aiskillstore/marketplace
perigon-backend
Perigon ASP.NET Core + EF Core + Aspire conventions
aiskillstore/marketplace
perigon-agent
Pointers for Copilot/agents to apply Perigon conventions
aiskillstore/marketplace
perigon-angular
Angular 21+ standalone/Material/signal conventions for Perigon WebApp
aiskillstore/marketplace
fastapi-mastery
Comprehensive FastAPI development skill covering REST API creation, routing, request/response handling, validation, authentication, database integration, middleware, and deployment. Use when working with FastAPI projects, building APIs, implementing CRUD operations, setting up authentication/authorization, integrating databases (SQL/NoSQL), adding middleware, handling WebSockets, or deploying FastAPI applications. Triggered by requests involving .py files with FastAPI code, API endpoint creation, Pydantic models, or FastAPI-specific features.
aiskillstore/marketplace
context7-efficient
Token-efficient library documentation fetcher using Context7 MCP with 86.8% token savings through intelligent shell pipeline filtering. Fetches code examples, API references, and best practices for JavaScript, Python, Go, Rust, and other libraries. Use when users ask about library documentation, need code examples, want API usage patterns, are learning a new framework, need syntax reference, or troubleshooting with library-specific information. Triggers include questions like "Show me React hooks", "How do I use Prisma", "What's the Next.js routing syntax", or any request for library/framework documentation.
aiskillstore/marketplace
browser-use
Browser automation using Playwright MCP. Navigate websites, fill forms, click elements, take screenshots, and extract data. Use when tasks require web browsing, form submission, web scraping, UI testing, or any browser interaction.
Didn't find tool you were looking for?