Frontend Events - mAIvn Docs

#Enrichment Events and Streaming to a Frontend

Show your users what the agent is doing in real time (thinking, planning, calling a tool, and answering) instead of staring at a spinner. This guide is the canonical home for mAIvn's live event surface: the enrichment phase indicators that fill the gaps between tool calls and the final answer, plus everything you need to forward those events to a browser, mobile app, or any other frontend.

It is written for SDK consumers. Everything here is built on the public maivn and maivn.events interfaces: no internal endpoints, no private wire details. The wire protocol your frontend ultimately reads is Server-Sent Events (SSE), and the contract that travels over it is a small, versioned, public shape.

#Why It Matters

When an agent works on a request, several seconds can pass between "user hit send" and "the answer appears." Under the hood the agent is evaluating the request, planning actions, calling tools, and synthesizing a response. If your UI shows nothing during that window, it feels broken.

Enrichment events close that gap. They are short, fixed-label progress signals the runtime emits at the meaningful transitions of a turn, for example evaluating the request, planning actions, and synthesizing the response, plus memory and redaction phases when those subsystems run. Your frontend renders them as a live status line or a row of phase chips, so the user always sees forward motion.

Two things make them easy to consume:

They arrive on the same stream as tool activity, assistant text, and the final answer. You read one stream, not several.
They share one normalized shape (AppEvent), so a single handler can dispatch every event type your UI cares about.

Enrichment vs. status messages. Enrichment phases are fixed-label lifecycle signals. Status messages are free-form dynamic text that may appear on the same stream. Keep them distinct: phases say which stage; status messages say what is happening right now.

#Reading Events From a Stream

Every agent and swarm exposes a streaming surface. agent.stream(...) yields events as the turn unfolds, and agent.astream(...) is the async equivalent. Swarms expose the same stream / astream methods.

Python

from maivn import Agent
from maivn.messages import HumanMessage

agent = Agent(name='support_agent', api_key='...')

for event in agent.stream([HumanMessage(content='Summarize this ticket')]):
    print(event.name, event.payload)

Each yielded value is a maivn.StreamEvent with canonical position, event_type, and data fields plus compatibility name and payload properties:

Python

class StreamEvent(BaseModel):
    position: int
    event_type: str
    data: dict[str, object]

Raw events are fine for quick experiments, but for product code you almost always want the normalized shape described below.

#Filtering with `events()`

BaseScope.events(...) (available on both Agent and Swarm) wraps an invocation with filtered event reporting. Use it when you want a callback fired for each event without writing your own stream loop:

Python

def on_event(event: dict) -> None:
    # Each dict has 'category', 'event' (the event name), and 'payload'.
    if event['event'] == 'enrichment':
        print('phase:', event['payload'].get('phase'))

response = agent.events(
    include=['enrichment', 'tool_event'],   # only these categories reach on_event
    on_event=on_event,
).invoke([HumanMessage(content='Look up the order status')])

Each dict passed to on_event has three keys: category (the event category used for filtering), event (the event name), and payload (the event's data). For enrichment, the payload carries phase, message, and any scope/memory/redaction/reevaluate fields.

The events() builder takes:

include / exclude — select or drop event categories (a single string or a sequence).
on_event — a callable invoked with each event dict.
auto_verbose — defaults to True, which turns on verbose event reporting for the wrapped call unless you override it.

The builder is terminal on .invoke(...) / .stream(...), so you choose blocking or streaming consumption at the call site.

#The Enrichment Contract

An enrichment event is just one event type on the stream. On the normalized contract it carries event_kind == 'enrichment' and an enrichment descriptor with two core fields:

Field	Meaning
`phase`	A fixed label for the lifecycle stage (e.g. the evaluate, planning, and synthesis phases, memory phases, and the `reevaluate_accrued` cycle marker). Render it as a chip.
`message`	A short human-readable line for the current phase. Render it as the live status text.

The phase labels are a small, fixed vocabulary the runtime owns. Your UI keys behavior off phase and shows message as text. Treat unknown phases gracefully (display the message, skip any phase-specific styling) so your frontend stays forward-compatible as the vocabulary grows.

#Reevaluate cycles

The event contract supports phase == "reevaluate_accrued" with attribution under a reevaluate bag. The current server emits this marker in a raw system_tool_chunk payload with stage == "reevaluate". The SDK normalizer drops that chunk when it has no text, so normalized enrichment listeners alone do not currently receive every reevaluate cycle. Inspect the raw stream when your UI requires this marker; legacy enrichment events use these fields:

source — "dependency" or "llm", i.e. whether a tool dependency or the model itself triggered the replan.
trigger_tool / target_tool — for dependency-driven cycles, which tool's result prompted re-planning of which target.
reevaluate_count / collected_count — cycle and collected counts.

#Scope attribution (swarm cards)

In a multi-agent run, enrichment events can carry scope attribution so each chip lands on the correct agent or swarm card in a UI like mAIvn Studio:

scope_id — stable identifier for the logical scope; key your cards off this.
scope_name — display label.
scope_type — e.g. agent vs. swarm.

For single-agent sessions these are omitted, because there is only one scope. For swarms, prefer scope_id over scope_name as the identity key: a supervised swarm can deploy the same agent name more than once, and only the id is stable across deployments.

#Normalizing Raw Events

Raw stream events are a transport detail. For anything beyond a quick script, normalize them once at your backend boundary into the public AppEvent contract using normalize_stream():

Python

from maivn import Agent
from maivn.events import RawSSEEvent, normalize_stream
from maivn.messages import HumanMessage

agent = Agent(name='support_agent', api_key='...')

raw_events = (
    RawSSEEvent(name=event.name, payload=event.payload)
    for event in agent.stream(
        [HumanMessage(content='Summarize this ticket')],
    )
)

for event in normalize_stream(raw_events, default_agent_name='support_agent'):
    send_to_frontend(event.model_dump())

normalize_stream() converts each adapted RawSSEEvent into a stable, validated AppEvent. Pass optional defaults when your application has stronger local context than the transport payload. normalize_stream_event() does the same thing one event at a time.

#The `AppEvent` shape

AppEvent gives your frontend one envelope with typed, optional descriptors for each concern:

Python

class AppEvent(BaseModel):
    contract_version: str
    event_name: str
    event_kind: str | None
    scope: ScopeDescriptor | None
    participant: ParticipantDescriptor | None
    lifecycle: LifecycleDescriptor | None
    tool: ToolDescriptor | None
    assistant: AssistantDescriptor | None
    assignment: AssignmentDescriptor | None
    enrichment: EnrichmentDescriptor | None
    interrupt: InterruptDescriptor | None
    output: OutputDescriptor | None
    error_info: ErrorInfoDescriptor | None
    session: SessionDescriptor | None
    chunk: ChunkDescriptor | None
    hook: HookDescriptor | None

For an enrichment event, event_kind == 'enrichment' and the enrichment descriptor is populated:

Python

class EnrichmentDescriptor(BaseModel):
    phase: str | None = None
    message: str | None = None
    memory: dict | None = None       # per-phase memory metrics, when present
    redaction: dict | None = None    # per-phase redaction metrics, when present
    reevaluate: dict | None = None   # attribution for reevaluate_accrued cycles

Descriptors use extra="allow", so the contract can grow without breaking consumers that validate normalized payloads.

#Versioning

The normalized contract is versioned with APP_EVENT_CONTRACT_VERSION:

Python

from maivn.events import APP_EVENT_CONTRACT_VERSION

APP_EVENT_CONTRACT_VERSION == "v1"

Preserve contract_version when forwarding events so you can support multiple frontend versions during a migration. Payloads deliberately keep legacy flat fields alongside the nested descriptors so existing consumers can migrate incrementally.

#Building enrichment payloads directly

If your backend produces or forwards events itself (rather than only relaying the SDK stream), use build_enrichment_payload() to emit a canonical enrichment payload that matches what the SDK would have produced:

Python

from maivn.events import build_enrichment_payload

payload = build_enrichment_payload(
    phase='synthesizing',
    message='Synthesizing response...',
    scope_id='agent-research',
    scope_name='research',
    scope_type='agent',
)

There is a matching build_*_payload() for each event family (tool events, assistant chunks, status messages, agent assignments, interrupts, final, error, and the system-tool lifecycle). Reach for these when you own the producer side and want contract-shaped output without hand-assembling envelopes.

#Custom Reporters

Most apps consume events through normalize_stream() or the events(on_event=...) callback above. Those are the supported public extension points, and they give you the full event stream in one place.

A custom "reporter" is just your own sink for those events. The cleanest pattern is a small dispatcher keyed off event_kind / event_name, with a dedicated branch for enrichment:

Python

from maivn import Agent
from maivn.events import normalize_stream
from maivn.messages import HumanMessage

class MyReporter:
    def report_enrichment(self, *, phase, message, scope_id=None, scope_name=None,
                          scope_type=None, memory=None, redaction=None, reevaluate=None):
        # Update your UI: chip for `phase`, status text for `message`.
        # `scope_*` route the chip to the right agent/swarm card.
        # `reevaluate` is populated when phase == "reevaluate_accrued".
        ...

    def dispatch(self, event) -> None:
        if event.event_kind == 'enrichment' and event.enrichment is not None:
            e = event.enrichment
            self.report_enrichment(
                phase=e.phase,
                message=e.message,
                scope_id=event.scope.id if event.scope else None,
                scope_name=event.scope.name if event.scope else None,
                scope_type=event.scope.type if event.scope else None,
                memory=e.memory,
                redaction=e.redaction,
                reevaluate=e.reevaluate,
            )

reporter = MyReporter()
agent = Agent(name='support_agent', api_key='...')
raw_events = (
    RawSSEEvent(name=event.name, payload=event.payload)
    for event in agent.stream([HumanMessage(content='Hi')])
)
for event in normalize_stream(raw_events):
    reporter.dispatch(event)

The shape of report_enrichment mirrors the enrichment contract: a required phase and message, plus optional scope_*, memory, redaction, and reevaluate attribution kwargs. Implement only the branches your UI needs; everything else falls through.

#Status Messages vs. Enrichment

Status messages are a separate event family. They are free-form dynamic text the runtime may emit to describe what's happening in the moment, distinct from the fixed-label enrichment phases. The current invocation API has no separate status_messages switch; consume them when they appear on stream():

Python

for event in agent.stream(messages):
    ...

When the stream carries status_message events, keep that dynamic text separate from fixed-label enrichment phases in your UI:

Enrichment phases answer which stage are we in — render as chips or a stepper.
Status messages answer what is happening right now, in words — render as a transient status line.

Swarm.stream(...) exposes the same event shape.

#Sending Events to a Browser

To get events into a frontend you put an EventBridge between your backend and the SSE connection. The bridge buffers history, validates payloads, handles reconnection replay, and applies audience-based redaction so you do not accidentally leak injected private data to an end user.

#One-line FastAPI mount

If your backend is FastAPI, the maivn.events.fastapi adapter wires up the streaming endpoint in a single call. Install the optional extra:

Shell

pip install "maivn[fastapi]"

Python

from fastapi import FastAPI
from maivn.events.fastapi import get_event_bridge, mount_events

app = FastAPI()
mount_events(app)  # mounts GET /maivn/events/{session_id}; bridges are frontend_safe by default


# Your own endpoint kicks off a turn and emits mAIvn events onto the
# per-session bridge. Your frontend subscribes to the mounted route
# GET /maivn/events/{session_id} using the same session_id.
@app.post("/start/{session_id}")
async def start(session_id: str) -> dict:
    bridge = get_event_bridge(session_id)
    await bridge.emit_status_message('orchestrator', 'Working...')
    await bridge.emit_final('Done!')
    return {"ok": True}

mount_events(app) wires up the streaming surface for you; get_event_bridge(session_id) returns (creating if needed) the per-session bridge you emit onto. The adapter creates frontend_safe bridges by default because the typical consumer is an end-user browser.

You can pass an auth dependency to gate connections and a custom bridge factory if you need session-scoped state. Both are standard FastAPI dependencies, so use your app's existing auth.

Other frameworks. mount_events is FastAPI-specific. For Flask, aiohttp, Django, raw ASGI, or anything else, use EventBridge directly. It produces the SSE frames and you pipe them into your framework's streaming response. The same bridge object also yields plain dicts, so you can adapt it to WebSockets if you prefer.

#Audiences: `frontend_safe` vs. `internal`

The bridge's audience sets how much sensitive detail is allowed onto history and the wire:

Python

from maivn.events import EventBridge

# End-user browsers: redacts injected private data and internal error details.
bridge = EventBridge('session-1', audience='frontend_safe')

# Trusted developer/admin surfaces (e.g. your own observability dashboard).
internal_bridge = EventBridge('session-1', audience='internal')

BridgeAudience is a public type with two values:

frontend_safe — preserves the mAIvn event contract but sanitizes sensitive bridge-bound values: injected private-data values are summarized to field names, redaction detail is masked, and internal error details are cleared. Unknown custom event types get a defensive injected-fields scrub so a stray emitter can't leak secrets. Use this for anything reaching an end user.
internal — passes payloads through unredacted. Use this only for trusted developer or admin tools.

EventBridgeSecurityPolicy exposes the same audience model if you need to inspect or reuse the sanitization decision directly. BridgeRegistry manages per-session bridge lifecycle (create / get / remove) when you wire the bridge by hand.

#Reading on the frontend

Once events are flowing through a bridge, the frontend reads them as standard SSE. Each frame carries a JSON envelope; your client dispatches on the event type and updates state. The envelope is the normalized AppEvent shape, so enrichment, tool, assistant, and terminal events all share one structure. Dedupe on the event id and key long-lived UI cards (tools, swarm members) off their stable descriptor ids rather than arrival order.

#Frontend Client Examples

The wire format is the same in every language, so the client is thin: open the stream, parse each SSE frame's JSON envelope, dispatch on the event type, and stop on a terminal event (final, error, session_end). Because every ecosystem has its own conventions, the SDK ships a stable wire contract rather than a per-language package. Copy the pattern that fits your stack.

Here is the browser pattern with the native EventSource API:

JavaScript

// Subscribe to the session's event stream.
const events = new EventSource(`/maivn/events/${sessionId}`);

const seen = new Set();
const eventTypes = [
  'tool_event', 'system_tool_start', 'system_tool_chunk', 'system_tool_complete',
  'assistant_chunk', 'status_message', 'agent_assignment', 'enrichment',
  'interrupt_required', 'hook_fired', 'session_start', 'final', 'error', 'session_end',
];
for (const type of eventTypes) {
  events.addEventListener(type, (e) => {
    // Network errors are plain Events; server error frames are MessageEvents.
    if (!(e instanceof MessageEvent)) return;
    if (e.lastEventId && seen.has(e.lastEventId)) return;
    if (e.lastEventId) seen.add(e.lastEventId);
    const frame = JSON.parse(e.data);
    updateUI(frame.data); // AppEvent is nested inside the bridge envelope.
    if (['final', 'error', 'session_end'].includes(type)) events.close();
  });
}
// EventSource reconnects after transport failures and sends Last-Event-ID.
// Close it yourself when leaving this screen. Bound `seen` for long sessions.

The bridge sends named SSE events, so onmessage alone does not receive them. Each JSON frame contains id, type, timestamp, and data; data holds the normalized payload. Add listeners for any custom event names your backend emits. For the payload fields, see the Events API reference.

#Next Steps

Events API reference — the full event-type and payload reference, plus normalization and replay helpers.
Sessions, Invocation, and Streaming — how invoke / stream and sessions fit together.
Token and Usage — read the token_usage dictionary from responses to surface usage in your UI.
Private Data — the redaction model behind frontend_safe audiences.
Multi-Agent — swarms, scope attribution, and per-agent cards.
mAIvn Studio — a local UI for inspecting live event streams while you build.
Versioning & Stability — what contract_version means and how to branch on it so a frontend survives backend upgrades.

#Enrichment Events and Streaming to a Frontend

#Why It Matters

#Reading Events From a Stream

#Filtering with events()

#The Enrichment Contract

#Reevaluate cycles

#Scope attribution (swarm cards)

#Normalizing Raw Events

#The AppEvent shape

#Versioning

#Building enrichment payloads directly

#Custom Reporters

#Status Messages vs. Enrichment

#Sending Events to a Browser

#One-line FastAPI mount

#Audiences: frontend_safe vs. internal

#Reading on the frontend

#Frontend Client Examples

#Next Steps

#Filtering with `events()`

#The `AppEvent` shape

#Audiences: `frontend_safe` vs. `internal`