Building Event-Driven Agents

Polling is acceptable for simple loops, but event-driven coordination is better when planners and workers need to react to new memory, handoffs, feedback, checkpoints, or outcome updates. The event stream is still a low-level control-plane surface. Keep the public integration model helper-first for memory writes and reads, and use the event stream to wake orchestration logic.

Decision model

Event lane	Surface	Use when
Run-scoped control events	`GET /v2/control/events/subscribe`	One workflow should react to ingest, reflection, handoff, feedback, checkpoint, or outcome events
Memory read/write loop	`remember`, `recall`, `getContext`, `checkpoint`	Most application behavior

Minimal implementation example

Python
Node / TS

event_driven_agents.py

from mubit import Client
import os

run_id = "claims:planner:claim-42"
client = Client(
    endpoint=os.getenv("MUBIT_ENDPOINT", "https://api.mubit.ai"),
    api_key=os.environ["MUBIT_API_KEY"],
    run_id=run_id,
    transport="http",
)

stream = client.control.subscribe({"run_id": run_id})

client.remember(
    session_id=run_id,
    agent_id="planner",
    content="A new adjuster note arrived for claim-42.",
    intent="trace",
)
client.checkpoint(
    session_id=run_id,
    agent_id="planner",
    label="before-adjuster-review",
    context_snapshot="Planner stored the adjuster note and is waiting for review.",
)

for event in stream:
    if event:
        print(event)
        break

event_driven_agents.ts

import { Client } from "@mubit-ai/sdk";

const runId = "claims:planner:claim-42";
const client = new Client({
  endpoint: process.env.MUBIT_ENDPOINT ?? "https://api.mubit.ai",
  api_key: process.env.MUBIT_API_KEY,
  run_id: runId,
  transport: "http",
});

const stream = await client.control.subscribe({ run_id: runId });

await client.remember({
  session_id: runId,
  agent_id: "planner",
  content: "A new adjuster note arrived for claim-42.",
  intent: "trace",
});
await client.checkpoint({
  session_id: runId,
  agent_id: "planner",
  label: "before-adjuster-review",
  context_snapshot: "Planner stored the adjuster note and is waiting for review.",
});

const reader = stream.body?.getReader();
const first = reader ? await reader.read() : null;
if (first?.value) {
  console.log(new TextDecoder().decode(first.value));
}

Failure modes and troubleshooting

Symptom	Root cause	Fix
Workers still poll constantly	Event stream is not wired to the orchestration loop	Move wakeup logic to the control stream
Event stream is noisy	Too many memory writes without filtering in the worker	Filter by event type in the consumer
Recovery is hard after compaction	Events fire but context is not rebuilt	Re-read with `getContext()` after the wakeup

Next steps

Add planner state at Stateful task trees.
Add branch-safe experimentation at Branching memory.

Recipes

​Decision model

​Minimal implementation example

​Failure modes and troubleshooting

​Next steps

Decision model

Minimal implementation example

Failure modes and troubleshooting

Next steps