Multi-Agent Event Bus: Chain Agents With Events

monitoring ──alert.raised──▶ [event bus] ──▶ triage agent (event trigger, source=monitoring)
                                                 │ publish_incident tool
                                                 ▼
                              [broker] incidents.<service>.<severity>
                                                 │ topic pattern incidents.*.high
                                                 ▼
                                            on-call agent ──▶ page_engineer tool

any agent's process ──process.failed──▶ EventBusSink ──▶ [event bus] ──▶ doctor agent

"""An in-process event bus and message broker with the methods the runtime's triggers call.

EventTrigger calls   await bus.subscribe(event_type, handler) / await bus.unsubscribe(event_type, handler)
MessageTrigger calls await broker.subscribe(topic, handler) -> id / await broker.unsubscribe(id)
Escalations and EventBusSink call await bus.emit(event_type, data)

Swap these for wrappers around Redis, NATS or Kafka when agents run in several programs.
"""
from __future__ import annotations

import fnmatch
from dataclasses import dataclass, field
from typing import Any
from uuid import uuid4


@dataclass
class BusEvent:
    """What EventTrigger reads: event_id, source and data."""

    data: dict[str, Any]
    source: str = ""
    event_id: str = field(default_factory=lambda: uuid4().hex)


@dataclass
class Message:
    """What MessageTrigger reads: message_id, sender and content."""

    content: Any
    sender: str = ""
    message_id: str = field(default_factory=lambda: uuid4().hex)


class EventBus:
    """Event types may use shell-style wildcards when subscribing: "alert.*"."""

    def __init__(self) -> None:
        self._handlers: dict[str, list] = {}
        self.log: list[str] = []

    async def subscribe(self, event_type: str, handler) -> None:
        self._handlers.setdefault(event_type, []).append(handler)

    async def unsubscribe(self, event_type: str, handler) -> None:
        self._handlers.get(event_type, []).remove(handler)

    async def emit(self, event_type: str, data: dict[str, Any], source: str | None = None) -> None:
        # Runtime events (EventBusSink) carry the process name as agent_id: use it as the source.
        source = source if source is not None else str(data.get("agent_id") or "")
        self.log.append(f"event   {event_type} from {source or '?'}")
        for pattern, handlers in list(self._handlers.items()):
            if fnmatch.fnmatchcase(event_type, pattern):
                for handler in list(handlers):
                    await handler(BusEvent(data=data, source=source))


class Broker:
    """Topics are dotted names; subscriptions may use wildcards: "incidents.*.high"."""

    def __init__(self) -> None:
        self._subs: dict[str, tuple[str, Any]] = {}
        self.log: list[str] = []

    async def subscribe(self, topic: str, handler) -> str:
        sub_id = uuid4().hex
        self._subs[sub_id] = (topic, handler)
        return sub_id

    async def unsubscribe(self, sub_id: str) -> None:
        self._subs.pop(sub_id, None)

    async def publish(self, topic: str, content: Any, sender: str = "") -> int:
        # The trigger's payload names the *pattern* it subscribed to, so put the real topic
        # into dict messages: the receiving agent can see where the message came from.
        if isinstance(content, dict):
            content = {"topic": topic, **content}
        self.log.append(f"message {topic} from {sender or '?'}")
        delivered = 0
        for pattern, handler in list(self._subs.values()):
            if fnmatch.fnmatchcase(topic, pattern):
                await handler(Message(content=content, sender=sender))
                delivered += 1
        return delivered

"""Step 1: plug the bus and broker into the two triggers and look at what they produce."""
import asyncio

from bus import Broker, EventBus
from promptise.runtime import EventTrigger, MessageTrigger


async def main() -> None:
    bus, broker = EventBus(), Broker()

    alerts = EventTrigger(bus, "alert.raised", source_filter="monitoring")
    incidents = MessageTrigger(broker, "incidents.*.high")
    for trigger in (alerts, incidents):
        await trigger.start()

    await bus.emit("alert.raised", {"service": "checkout", "error_rate": 0.31}, source="load-test")
    await bus.emit("alert.raised", {"service": "checkout", "error_rate": 0.31}, source="monitoring")
    await broker.publish("incidents.checkout.low", {"summary": "slow page"}, sender="triage")
    await broker.publish("incidents.checkout.high", {"summary": "31% errors"}, sender="triage")

    print("EventTrigger:  ", (await alerts.wait_for_next()).payload)
    print("MessageTrigger:", (await incidents.wait_for_next()).payload)
    for trigger in (alerts, incidents):
        await trigger.stop()
        print("left over:     ", trigger.trigger_id, trigger.drain_pending())


asyncio.run(main())

"""Chain two agents with events: monitoring -> triage agent -> broker topic -> on-call agent.

Needs OPENAI_API_KEY. The two tool servers run in this program, so their tools can
publish to the in-process broker; in production they would publish to Redis, NATS or Kafka.
"""
import asyncio
import logging
import time

from bus import Broker, EventBus
from promptise.mcp.server import MCPServer
from promptise.runtime import (
    AgentRuntime, HookEvent, HookManager, ProcessConfig, TriggerConfig,
)

T0 = time.perf_counter()
bus, broker = EventBus(), Broker()

# -- Tools: the triage agent can publish, the on-call agent can page -------------------------
triage_tools = MCPServer(name="triage-tools")


@triage_tools.tool()
async def publish_incident(service: str, severity: str, summary: str) -> str:
    """Publish an incident for the other agents. severity is low, medium or high."""
    topic = f"incidents.{service}.{severity}"
    receivers = await broker.publish(
        topic, {"service": service, "severity": severity, "summary": summary}, sender="triage")
    return f"Published to {topic} ({receivers} subscriber(s))."


oncall_tools = MCPServer(name="oncall-tools")


@oncall_tools.tool()
async def page_engineer(service: str, message: str) -> str:
    """Page the on-call engineer of a service with a short message."""
    print(f"{time.perf_counter() - T0:5.1f}s  PAGE [{service}] {message}")
    return "Paged."


# -- Agents ----------------------------------------------------------------------------------
TRIAGE = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions=(
        "You triage monitoring alerts. For each alert, call publish_incident once. "
        "Severity: high if error_rate >= 0.10 or the service is down, medium if "
        "error_rate >= 0.02, otherwise low. The summary is one sentence with the numbers."
    ),
    servers={"tools": {"url": "http://127.0.0.1:8801/mcp"}},
    triggers=[TriggerConfig(type="event", event_type="alert.raised", event_source="monitoring")],
)
ONCALL = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions=(
        "You are the on-call assistant. For each high-severity incident, call page_engineer "
        "once with a message under 25 words: what is broken and the first thing to check."
    ),
    servers={"tools": {"url": "http://127.0.0.1:8802/mcp"}},
    triggers=[TriggerConfig(type="message", topic="incidents.*.high")],
)


async def main() -> None:
    servers = [asyncio.create_task(triage_tools.run_async("http", port=8801)),
               asyncio.create_task(oncall_tools.run_async("http", port=8802))]
    await asyncio.sleep(1.0)  # let both tool servers bind
    logging.getLogger("uvicorn.access").disabled = True  # keep the output readable

    hooks = HookManager()

    @hooks.on(HookEvent.TASK_COMPLETED)
    async def done(ctx) -> None:
        print(f"{time.perf_counter() - T0:5.1f}s  {ctx.process_id:<7} run {ctx.data['outcome']}"
              f" ({ctx.data['trigger_type']} trigger)")

    runtime = AgentRuntime(event_bus=bus, broker=broker, hooks=hooks)
    triage = await runtime.add_process("triage", TRIAGE)
    oncall = await runtime.add_process("oncall", ONCALL)
    await runtime.start_all()

    await bus.emit("alert.raised", {"service": "checkout", "error_rate": 0.31,
                                    "window": "5m"}, source="monitoring")
    await bus.emit("alert.raised", {"service": "search", "p95_latency_ms": 870,
                                    "error_rate": 0.004}, source="monitoring")

    deadline = time.perf_counter() + 120
    while time.perf_counter() < deadline:
        high = sum(".high " in line for line in broker.log)
        if triage.status()["invocation_count"] == 2 and oncall.status()["invocation_count"] == high:
            break
        await asyncio.sleep(0.2)

    print("bus traffic:", *bus.log, *broker.log, sep="\n  ")
    await runtime.stop_all()
    for task in servers:
        task.cancel()
    await asyncio.gather(*servers, return_exceptions=True)


asyncio.run(main())

TriggerConfig(type="event", event_type="alert.raised", event_source="monitoring",
              filter_expression="payload.data.error_rate >= 0.02")
TriggerConfig(type="message", topic="incidents.*.high",
              filter_expression="payload.content.service != 'sandbox'")

"""Step 4: wake a "doctor" agent whenever another agent's process fails."""
import asyncio
import json
import logging
import os
import re

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8794/v1"  # the scripted model server
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.getLogger("promptise").setLevel(logging.CRITICAL)

from bus import EventBus
from fake_llm import FakeLLM
from promptise.events import EventBusSink, EventNotifier
from promptise.runtime import AgentRuntime, ProcessConfig, TriggerConfig, TriggerEvent

bus = EventBus()
notifier = EventNotifier(sinks=[EventBusSink(bus, events=["process.failed"])])

WORKER = ProcessConfig(model="openai:gpt-5-mini", instructions="You reconcile invoices.",
                       max_consecutive_failures=1)
DOCTOR = ProcessConfig(
    model="openai:gpt-5-mini",
    instructions="A process failed. Read the error and suggest the most likely cause.",
    triggers=[TriggerConfig(
        type="event",
        event_type="process.failed",
        filter_expression="source != 'doctor'",  # never react to its own failures
    )],
)


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    llm.poison = {"INV-9"}  # this invoice makes the worker's model call fail
    llm.reply = "Likely cause: the model provider rejected the input (HTTP 400)."

    runtime = AgentRuntime(event_bus=bus, event_notifier=notifier)
    worker = await runtime.add_process("invoice-worker", WORKER)
    doctor = await runtime.add_process("doctor", DOCTOR)
    await runtime.start_all()

    await worker.inject(TriggerEvent(trigger_id="manual", trigger_type="queue",
                                     payload={"invoice": "INV-9"}))
    await asyncio.sleep(1.5)

    print("worker state:", worker.state.value, "| doctor runs:",
          doctor.status()["invocation_count"])
    print("bus:", bus.log)
    block = re.search(r"<(untrusted-[\w-]+)>\n(.*)\n</\1>", llm.prompts[-1], re.S).group(2)
    payload = json.loads(block)  # what the doctor agent read
    event = payload["data"]      # the AgentEvent that EventBusSink put on the bus
    print("doctor payload: source =", payload["source"], "| event_type =", event["event_type"])
    print("  data =", {k: str(v)[:60] for k, v in event["data"].items()})
    await runtime.stop_all()
    await llm.stop()


asyncio.run(main())

"""Step 5: what happens when events arrive faster than an agent can take them."""
import asyncio
import logging
import os

os.environ["OPENAI_BASE_URL"] = "http://127.0.0.1:8794/v1"
os.environ["OPENAI_API_KEY"] = "not-a-real-key"
logging.getLogger("promptise").setLevel(logging.CRITICAL)

from bus import EventBus
from fake_llm import FakeLLM
from promptise.runtime import AgentRuntime, ProcessConfig, TriggerConfig


class CountWarnings(logging.Handler):
    count = 0

    def emit(self, record: logging.LogRecord) -> None:
        if "queue full" in record.getMessage():
            CountWarnings.count += 1


trigger_log = logging.getLogger("promptise.runtime.triggers.event")
trigger_log.setLevel(logging.WARNING)
trigger_log.addHandler(CountWarnings())
trigger_log.propagate = False


async def main() -> None:
    llm = FakeLLM()
    await llm.start()
    llm.delay = 0.05  # each run takes a moment
    bus = EventBus()
    runtime = AgentRuntime(event_bus=bus)
    process = await runtime.add_process("order-agent", ProcessConfig(
        model="openai:gpt-5-mini", instructions="Confirm orders.",
        triggers=[TriggerConfig(type="event", event_type="order.created")]))
    await runtime.start_all()

    async def burst(label: str, yield_between: bool) -> None:
        before, dropped_before = process.status()["invocation_count"], CountWarnings.count
        for n in range(150):
            await bus.emit("order.created", {"order": n}, source="shop")
            if yield_between:
                await asyncio.sleep(0)  # give the trigger's listener a turn
        await asyncio.sleep(0.2)
        s = process.status()
        accepted = s["invocation_count"] - before + s["queue_size"] + 1  # +1 running
        print(f"{label:<28} sent=150  accepted={accepted}  queued={s['queue_size']}"
              f"  dropped={CountWarnings.count - dropped_before}")

    await burst("150 events, no yielding", yield_between=False)
    await asyncio.sleep(10)  # let the agent catch up
    await burst("150 events, yield each time", yield_between=True)
    await runtime.stop_all()
    print("dead letters after stop:", len(process.dead_letters),
          {d["reason"] for d in process.dead_letters})
    await llm.stop()


asyncio.run(main())

event-bus/
├── bus.py               # EventBus and Broker with wildcards (step 1)
├── bus_basics.py        # what the triggers receive (step 1)
├── chain.py             # monitoring -> triage -> broker -> on-call (step 2, needs OPENAI_API_KEY)
├── lifecycle_chain.py   # process.failed -> doctor agent (step 4)
├── burst.py             # full trigger buffers (step 5)
└── fake_llm.py          # scripted model server for steps 4 and 5

pip install "promptise>=1.5.0"
export OPENAI_API_KEY=sk-...
python chain.py