I Built an AI Agent That Watches Your Database and Acts on Its Own

Most AI demos follow the same script. User types a prompt. Agent responds. Everyone claps.

That is autocomplete with extra steps.

The more interesting question is: what happens when no one is typing? What if the agent watches your data continuously and acts the moment something crosses a threshold, without being asked, without a human in the loop, without a scheduled cron job?

That is the ambient agent pattern. And I built a working implementation of it on Kubernetes.

Here is everything I learned.

The Problem with How Agents Are Usually Built

The standard agent architecture looks like this:

User types something
      |
      v
LLM reasons about it
      |
      v
Agent calls a tool
      |
      v
Response comes back

This works for assistants. It does not work for operational systems.

Imagine an inventory management system. You do not want a human typing “check stock levels” every few minutes. You want the system to notice when stock drops below a threshold and immediately place a reorder and alert the operations team. On its own. Every time. At 3am if needed.

The reactive, poll-based alternative most teams reach for looks like this:

# The naive approach most teams ship
while True:
    rows = db.query("SELECT * FROM inventory WHERE stock_level < reorder_threshold")
    for row in rows:
        trigger_agent(row)
    time.sleep(60)

This works until it does not. You are polling on a fixed interval, so you are always trading off latency against database load. You need to track which rows you already processed or you will trigger the agent on the same low-stock item every 60 seconds. You need to handle missed runs, restarts, and crashes. The logic that seems simple on day one becomes a maintenance burden by month three.

There is a better way.

The Ambient Agent Pattern

An ambient agent does not wait to be asked. It watches the data continuously and wakes up exactly when something crosses a condition.

Always running, doing nothing
                               |
                    PostgreSQL row changes
                               |
                    Drasi detects it immediately
                               |
                    Smart Router routes the event
                               |
                    Agent wakes up, reasons, acts
                               |
                    Agent goes back to waiting

No polling. No interval. No “did I already process this?” bookkeeping. The agent is event-driven at the data layer.

I built this for an inventory management system where a PostgreSQL table tracks stock levels. The moment any product’s stock drops below its reorder threshold, the agent places a reorder and alerts the operations team. No human in the loop, no manual trigger.

Here is what the stack looks like.

The Stack

Drasi (Microsoft, Open Source)

Drasi is a reactive continuous query engine. You write a Cypher query. Drasi evaluates it continuously against your database. The moment a row crosses the condition, Drasi fires a ChangeEvent downstream.

MATCH (i:inventory)
WHERE i.stock_level < i.reorder_threshold
RETURN i.product_name AS productName,
       i.stock_level  AS stockLevel

This is not a cron job. Drasi uses change data capture (CDC) on PostgreSQL to detect row-level changes in real time. The latency is measured in milliseconds, not minutes.

Dapr (CNCF)

Dapr handles the distributed plumbing: pub/sub messaging, state store, durable workflow runtime. Without Dapr, I would be hand-writing Redis clients, retry logic, actor registration, and workflow state management. Dapr wraps all of that behind a clean HTTP/gRPC API.

dapr-agents

An AI agent framework built on top of Dapr Workflow. The key piece is DurableAgent, which gives you a fully durable LLM reasoning loop. If the pod crashes mid-tool-call, the workflow resumes where it left off when the pod restarts. That durability matters in production.

Smart Router (Custom Drasi Reaction)

The bridge between Drasi and Dapr. Drasi knows nothing about Dapr. Dapr knows nothing about Drasi. The Smart Router sits between them: it receives Drasi ChangeEvents, looks up routing rules from Redis, and publishes spec-compliant CloudEvents to the correct Dapr Pub/Sub topic.

The Smart Router also hosts an embedded MCP (Model Context Protocol) server on port 3001 so agents can discover and manage routing rules dynamically at runtime without redeployment.

Ollama + llama3.2:3b

A local LLM running inside the Kubernetes cluster. No API keys. No cloud calls. No per-token billing. The model runs in a pod alongside the agent.

Architecture

┌─────────────────────────────────────────────────────────────────┐
│                        Data Layer                               │
│   PostgreSQL ──► Drasi Source ──► Continuous Query              │
│                        │                                        │
│                        ▼  ChangeEvent                           │
│              ┌─────────────────────┐                            │
│              │   Smart Router      │  (Drasi Reaction)          │
│              │   (Python)          │◄── MCP Server :3001        │
│              └─────────┬───────────┘    subscribe/list queries  │
│                        │  CloudEvent (Dapr Pub/Sub)             │
│                        ▼                                        │
│              ┌─────────────────────┐                            │
│              │   Dapr Pub/Sub      │  (Redis)                   │
│              │   inventory-alerts  │                            │
│              └─────────┬───────────┘                            │
│                        │  POST /inventory-alerts                │
│                        ▼                                        │
│              ┌─────────────────────┐                            │
│              │   InventoryAgent    │  (DurableAgent)            │
│              │   llama3.2:3b       │──► place_reorder()         │
│              │   (Ollama)          │──► send_alert()            │
│              └─────────────────────┘                            │
└─────────────────────────────────────────────────────────────────┘

The Full Data Flow

Step by step, when a row changes in PostgreSQL:

1. Drasi evaluates the low-stock-query Cypher expression continuously against the source. No polling. CDC handles change detection.
2. When stock_level < reorder_threshold becomes true for a row, Drasi fires a ChangeEvent to the registered Reaction.
3. Smart Router receives the event, matches it against routing rules stored in Redis, builds a spec-compliant CloudEvent, and publishes it to the inventory-alerts Dapr Pub/Sub topic.
4. Dapr delivers the event to the agent’s /inventory-alerts HTTP endpoint via push subscription.
5. The endpoint formats the change as a task string and schedules a DurableAgent workflow.
6. llama3.2:3b via Ollama reads the task, reasons about what to do, and calls PlaceReorder() and SendAlert() tools.
7. Agent goes back to waiting.

Nobody typed anything. A database row changed and within seconds the agent had reasoned and acted.

Code Deep Dive

The Smart Router: Bridging Drasi and Dapr

The Smart Router is the most novel piece. It is a Python Drasi Reaction that does four things: receives ChangeEvents, applies routing rules, builds CloudEvents, and publishes to Dapr.

Here is the core routing logic:

# smart_router/router.py

import structlog
from smart_router.publisher import EventPublisher
from smart_router.rules import RoutingRule
from smart_router.state import RuleStore

log = structlog.get_logger()

class SmartRouter:
    def __init__(self, publisher: EventPublisher, rule_store: RuleStore):
        self.publisher = publisher
        self.rule_store = rule_store

    async def handle_change_event(self, query_id: str, event: dict) -> None:
        rules = await self.rule_store.get_rules(query_id)

        if not rules:
            log.info("smart_router.no_rules", query_id=query_id)
            return

        matched = [r for r in rules if self._matches(event, r)]
        log.info("smart_router.event_received",
                 query_id=query_id,
                 matched_rules=len(matched))

        for rule in matched:
            await self.publisher.publish(
                event=event,
                topic=rule.target_topic,
                pubsub_name=rule.pubsub_name,
            )

    def _matches(self, event: dict, rule: RoutingRule) -> bool:
        if not rule.filter:
            return True
        added = event.get("addedResults", [])
        return any(
            self._check_filter(row, rule.filter)
            for row in added
        )

    def _check_filter(self, row: dict, f) -> bool:
        value = row.get(f.field)
        if f.operator == "lt":
            return float(value) < float(f.value)
        if f.operator == "gt":
            return float(value) > float(f.value)
        if f.operator == "eq":
            return str(value) == str(f.value)
        return False

The publisher side builds a proper CloudEvent before publishing to Dapr:

# smart_router/publisher.py

import json
from datetime import datetime, timezone
from cloudevents.http import CloudEvent
from cloudevents.conversion import to_structured
from dapr.clients import DaprClient
import structlog

log = structlog.get_logger()

def _to_serializable(obj):
    """Drasi SDK returns RecordUnknown objects that are not JSON serializable."""
    if isinstance(obj, dict):
        return {k: _to_serializable(v) for k, v in obj.items()}
    if isinstance(obj, list):
        return [_to_serializable(i) for i in obj]
    try:
        json.dumps(obj)
        return obj
    except TypeError:
        return str(obj)

class EventPublisher:
    def __init__(self, dapr_client: DaprClient):
        self.client = dapr_client

    async def publish(self, event: dict, topic: str, pubsub_name: str) -> None:
        safe_data = _to_serializable(event)

        cloud_event = CloudEvent(
            attributes={
                "type": "com.drasi.changeevent",
                "source": "drasi/smart-router",
                "time": datetime.now(timezone.utc).isoformat(),
            },
            data=safe_data,
        )

        headers, body = to_structured(cloud_event)

        self.client.publish_event(
            pubsub_name=pubsub_name,
            topic_name=topic,
            data=body,
            data_content_type="application/cloudevents+json",
        )

        log.info("publisher.published",
                 event_id=cloud_event["id"],
                 topic=topic)

The Agent: Durable LLM Reasoning on Real Events

The agent receives the CloudEvent from Dapr and runs an LLM reasoning loop to decide which tools to call.

# demo/agent/agent.py

import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI, Request
from dapr_agents import DurableAgent, tool
from dapr_agents.extensions.drasi import ChangeEvent

logging.basicConfig(level=logging.INFO)
log = logging.getLogger("inventory-agent")

@tool
def place_reorder(product_id: str, quantity: str) -> str:
    """Place a reorder for a product that is low on stock."""
    log.info(f"PlaceReorder: {quantity} units of {product_id}")
    return f"Reorder placed: {quantity} units of {product_id}"

@tool
def send_alert(message: str, severity: str) -> str:
    """Send an alert to the operations team."""
    log.info(f"SendAlert [{severity}]: {message}")
    return f"Alert sent: {message}"

agent = DurableAgent(
    name="InventoryAgent",
    instructions=(
        "You are an inventory management agent. "
        "When you receive a low-stock alert, always place a reorder "
        "and send an alert to the operations team."
    ),
    tools=[place_reorder, send_alert],
    message_bus_name="pubsub",
    state_store_name="statestore",
    llm_base_url="http://ollama:11434/v1",
    llm_model="llama3.2:3b",
)

runner = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    global runner
    runner = await agent.start()
    yield

app = FastAPI(lifespan=lifespan)

@app.post("/inventory-alerts")
async def handle_inventory_alert(request: Request):
    """
    Explicit push subscription endpoint.
    dapr-agents v1.0.0 bypasses @drasi_trigger decorated handlers for DurableAgent,
    so we wire the endpoint manually and call runner.run() directly.
    """
    body = await request.json()
    raw_data = body.get("data", body)
    event = ChangeEvent(**raw_data)

    if not event.has_additions:
        return {"status": "no low-stock items in event"}

    items = "\n".join(
        f"  - {r.get('productName')}: {r.get('stockLevel')} units"
        for r in event.addedResults
    )

    task = (
        f"Low stock alert:\n{items}\n\n"
        f"Place reorders and notify the operations team."
    )

    await runner.run(agent, {"task": task})
    return {"status": "workflow scheduled"}

The MCP Server: Dynamic Routing Without Redeployment

One of the more interesting pieces is the embedded MCP server inside the Smart Router. Any agent can connect to it and subscribe to new Drasi queries at runtime without redeploying anything.

# smart_router/mcp_server.py

from fastmcp import FastMCP
from smart_router.rules import RoutingRule, FieldFilter
from smart_router.state import RuleStore

mcp = FastMCP("SmartRouter", host="0.0.0.0", port=3001)
_rule_store: RuleStore = None

def init_mcp(rule_store: RuleStore):
    global _rule_store
    _rule_store = rule_store

@mcp.tool()
async def list_queries() -> dict:
    """List all active routing rules."""
    rules = await _rule_store.get_all_rules()
    return {"rules": [r.model_dump() for r in rules]}

@mcp.tool()
async def subscribe(
    query_id: str,
    target_topic: str,
    pubsub_name: str = "pubsub",
    filter_field: str = None,
    filter_operator: str = None,
    filter_value: str = None,
) -> dict:
    """Add a routing rule. Optionally filter on a specific field."""
    rule = RoutingRule(
        query_id=query_id,
        target_topic=target_topic,
        pubsub_name=pubsub_name,
        filter=FieldFilter(
            field=filter_field,
            operator=filter_operator,
            value=filter_value,
        ) if filter_field else None,
    )
    await _rule_store.save_rule(rule)
    return {"status": "subscribed", "rule": rule.model_dump()}

@mcp.tool()
async def unsubscribe(query_id: str, target_topic: str) -> dict:
    """Remove a routing rule."""
    await _rule_store.delete_rule(query_id, target_topic)
    return {"status": "unsubscribed"}

An agent that wants to wire itself to a new Drasi query calls the MCP server directly:

import httpx

with httpx.Client(timeout=10) as client:
    with client.stream("GET", "http://smart-router:3001/sse") as r:
        for line in r.iter_lines():
            if line.startswith("data:"):
                endpoint = line.removeprefix("data:").strip()
                break

    resp = client.post(
        f"http://smart-router:3001{endpoint}",
        json={
            "jsonrpc": "2.0", "id": 1,
            "method": "tools/call",
            "params": {
                "name": "subscribe",
                "arguments": {
                    "query_id": "high-temperature-query",
                    "target_topic": "temperature-alerts",
                    "pubsub_name": "pubsub",
                    "filter_field": "temperature",
                    "filter_operator": "gt",
                    "filter_value": "80",
                }
            }
        }
    )
    print(resp.json())

No redeployment. No config file update. The routing rule is live immediately and persists in Redis across restarts.

Scale to Zero with KEDA

The agent does nothing when there are no events. Running it at full replicas 24/7 wastes resources. KEDA handles this with a ScaledObject that watches the Redis stream depth.

apiVersion: keda.sh/v1alpha1
kind: ScaledObject
metadata:
  name: inventory-agent-scaler
  namespace: default
spec:
  scaleTargetRef:
    name: inventory-agent
  minReplicaCount: 0
  maxReplicaCount: 5
  triggers:
    - type: redis-streams
      metadata:
        address: redis.default.svc.cluster.local:6379
        stream: pubsub-inventory-alerts
        consumerGroup: inventory-agent
        pendingEntriesCount: "1"

With minReplicaCount: 0, the agent pod does not exist when there is nothing to process. The moment a ChangeEvent arrives in the Redis stream, KEDA spins up a replica. Once the backlog clears, it scales back to zero. True scale-to-zero for an event-driven AI agent.

Real Demo Output

Three terminals. One database change.

kubectl exec -n default deployment/postgres -- psql -U postgres -d inventory_db \
  -c "UPDATE inventory SET stock_level = 2 WHERE product_id = 'GADGET-002';"

Smart Router terminal:

smart_router.event_received   matched_rules=1   query_id=low-stock-query
smart_router.routing          pubsub=pubsub     topic=inventory-alerts
publisher.published           event_id=b199a3...
POST /low-stock-query  200 OK

Agent terminal:

POST /inventory-alerts  200 OK

user:
  Low stock alert:
    - Super Gadget: 2 units
  Place reorders and notify team.

InventoryAgent(assistant):
  Running tool: PlaceReorder { product_id: "Super Gadget", quantity: "2" }
  Running tool: SendAlert { message: "Low stock: Super Gadget at 2 units", severity: "LOW" }

PlaceReorder: Reorder placed: 2 units of Super Gadget
SendAlert:    Alert sent.

No human in the loop. A row changed, Drasi saw it, the Smart Router routed it, the agent reasoned and acted. The whole pipeline ran in under five seconds.

Tech Stack

• Kubernetes (kind) — container orchestration
• Drasi — continuous reactive query engine with CDC on PostgreSQL
• Dapr — distributed runtime for pub/sub, state store, and workflow
• dapr-agents — durable AI agent framework on Dapr Workflow
• Ollama + llama3.2:3b — local LLM with tool calling support
• PostgreSQL + Debezium — source database with change data capture
• Redis — state store and pub/sub broker
• FastAPI + uvicorn — agent HTTP server
• KEDA — scale-to-zero based on Redis stream depth
• FastMCP — embedded MCP server in the Smart Router
• Python 3.11 — implementation language

Where This Pattern Actually Makes Sense

Ambient agents are not a demo trick. The pattern is useful anywhere you need automated reasoning and action in response to data changes.

Compliance monitoring — agent watches audit logs, flags anomalies, and opens a ticket automatically when a policy is violated.

Infrastructure operations — agent monitors pod crash rates, scales affected deployments, or creates a PagerDuty incident when thresholds are crossed.

Finance — agent watches transaction patterns in real time and pauses a card when a fraud signal matches defined conditions.

Healthcare — agent monitors patient vitals in a database and alerts the care team when thresholds are crossed.

Supply chain — exactly what this demo does. Stock drops, reorder fires. No human touchpoint needed for routine operations.

The common thread: continuous query on structured data, change event to an agent, agent reasons and takes action. The LLM is not answering questions. It is making operational decisions based on live data state.

Honest Assessment

What works well:

• Drasi’s continuous query approach eliminates polling entirely. The latency between a row change and an agent response is genuinely low.
• Dapr’s Workflow runtime makes the agent durable. A pod crash mid-tool-call means the tool runs again on restart, not that the operation is silently lost.
• The MCP server in the Smart Router is a clean way to manage routing rules without touching config files or redeploying.
• Scale-to-zero with KEDA is real. The agent pod does not exist when there is nothing to process.

What needs work:

• dapr-agents v1.0.0 has real compatibility bugs with DurableAgent pub/sub handlers. The workaround is not complex, but it should not be necessary.
• Drasi’s dependency on MongoDB for internal state is not obvious from the documentation. The race condition on restart caught me completely off-guard.
• Running llama3.2:3b inside Kubernetes adds significant resource requirements. Not suitable for environments where you cannot provision enough CPU/memory for the model.
• Namespace isolation requirements for Dapr components add operational complexity in multi-namespace deployments.

Conclusion

The ambient agent pattern is fundamentally different from the prompt-response loop that defines most AI applications. The agent is not answering questions. It is watching your systems, making decisions, and taking action the moment conditions are met.

Drasi handles the hard part: CDC-based reactive queries with no polling, no interval management, no duplicate detection. Dapr handles the distributed systems plumbing. dapr-agents provides the durable reasoning loop on top. The Smart Router bridges the two worlds together.

The bugs I hit were real and non-obvious. The solutions were straightforward once I understood what was actually happening under the hood. That is usually how it goes.

The code is all open source. If you are building anything that needs reactive, event-driven AI agents, this architecture is worth understanding even if you swap out individual components for something that fits your stack better.

Thanks for reading. If you have questions about the architecture, the bugs, or the ambient agent pattern in general, drop them in the comments.