Drasi
Note
This integration is currently in Alpha; breaking changes may occur at any time.The Drasi extension in Dapr Agents enables DurableAgent runs to be triggered by Drasi events.
Why Use Drasi?
Many systems need to react to changes produced by other systems in near real-time. Traditional polling cannot detect the absence of change, or changes that occur at an extremely high frequency without unnecessary load on source systems (even without considering network delay). Raw changes are usually not actionable, requiring custom change data capture (CDC) pipelines to process changes at scale and convert them into meaningful domain events. However, these pipelines can be expensive (if managed) or difficult to set up and maintain (if self-hosted).
For many use cases, Drasi is a viable alternative. Drasi is a CNCF Sandbox project that addresses the issues mentioned above with a simple architecture centered around detecting and reacting to changes:
- Sources to ingest changes from existing systems
- Queries allowing high-level “business conditions” to be defined across a variety of data sources, which emit events when those conditions are satisfied
- Reactions to push events to downstream consumers
Installation
pip install "dapr-agents[drasi]"
uv add dapr-agents[drasi]
Usage
from dapr_agents import AgentRunner, DurableAgent
from dapr_agents.agents.configs import AgentPubSubConfig
from dapr_agents.ext.drasi import drasi_trigger
agent = DurableAgent(
name="InventoryAgent",
pubsub=AgentPubSubConfig(
pubsub_name="pubsub", # Replace with your pub/sub component
agent_topic="inventory-agent", # Replace with your agent pub/sub topic
),
)
drasi_trigger(
agent,
query_id="low-stock-products", # Replace with your Drasi query ID
)
AgentRunner().serve(agent)
API
drasi_trigger
drasi_trigger creates a static subscription to a Drasi query and allows agents to be triggered by Drasi events via Dapr pub/sub.
Parameters
| Parameter | Type | Required | Details | Example |
|---|---|---|---|---|
agent | DurableAgent | Y | The target agent. | N/A |
query_id | str | Y | The Drasi query ID to subscribe to. | "low-stock-products" |
pubsub | str | N | The name of the Dapr pub/sub component to use. Defaults to the agent’s pub/sub component. | "pubsub" |
topic | str | N | The topic to subscribe to. Defaults to "drasi-events-" + query_id. | N/A |
dead_letter_topic | str | N | Dead-letter topic to publish failed messages to. | "low-stock-events-dlq" |
task_mapper | Callable[[DrasiChangeEvent, MessageContext], TriggerAction] | N | Callable to map Drasi change events to agent task messages. Defaults to instructing the agent to return the serialized Drasi event as-is. | N/A |
operations | DrasiOperation | str | list[DrasiOperation | str] | N | Drasi operation(s) to filter change events by. Accepts DrasiOperation or equivalent string literals. | N/A |
change_model | type[Any] | N | Model to use to validate the change data in Drasi events. | N/A |
DrasiOperation
DrasiOperation is an enum representing the supported Drasi change operations.
Operations
| Operation | Value | Details |
|---|---|---|
DrasiOperation.i | "i" | A record was added to the result set tracked by the Drasi query. |
DrasiOperation.u | "u" | A record was updated in the result set tracked by the Drasi query. |
DrasiOperation.d | "d" | A record was deleted from the result set tracked by the Drasi query. |
DrasiChangeEvent
DrasiChangeEvent is a Pydantic model representing a change event emitted by a query.
Attributes
| Attribute | Type | Required | Details | Example |
|---|---|---|---|---|
op | DrasiOperation | Y | The change event operation (insert, update, delete). | DrasiOperation.u |
ts_ms | int | Y | The timestamp of the change event in milliseconds. | 42 |
seq | int | Y | The sequence number of the change event. | 1 |
payload | dict[str, Any] | Y | The change data for the change event. | {"source": {"queryId": "low-stock-products", "ts_ms": 42}, "before": {"a": 1}, "after": {"a": 2}} |
Example Structure
{
"op": "u",
"ts_ms": 42,
"seq": 1,
"payload": {
"source": {
"queryId": "low-stock-products",
"ts_ms": 42
},
"before": {"a": 1},
"after": {"a": 2}
}
}
Examples
See the Extension Examples for a list of working examples for the Drasi extension.