Custom Events

AG-UI CUSTOM events let a backend node push arbitrary data to the Angular client while a run is in progress. The adapter accumulates these events into a customEvents signal on the AgUiAgent returned by injectAgent() — reachable directly, no cast required (shown in Reading Custom Events below).

interface CustomStreamEvent {
  name: string;
  data: unknown;
}

customEvents is a Signal<CustomStreamEvent[]>. The list is reset to [] when RUN_STARTED arrives, so it only ever contains events from the current run.

Where Custom Events Come From

Exactly one wire event feeds customEvents: an AG-UI CUSTOM frame whose name is anything other than on_interrupt. Every backend path below is judged by whether it produces that frame.

{
  "type": "CUSTOM",
  "name": "analysis_progress",
  "value": { "step": "scoring", "pct": 42 }
}

The adapter JSON-parses value when it arrives as a string, so consumers always receive the structured object. The event is appended to customEvents as { name: "analysis_progress", data: { step: "scoring", pct: 42 } }.

The working path under ag-ui-langgraph

The ag-ui-langgraph bridge consumes the graph through astream_events, and it forwards every on_custom_event it sees one-for-one as an AG-UI CUSTOM frame carrying the same name and payload. LangChain's adispatch_custom_event is what puts an on_custom_event on that stream, so it is the call a node (or a callback handler running inside one) makes to reach customEvents.

The threadplane-middleware Python package wraps that call as emit_custom_event, which is the recommended way to make it:

from langchain_core.runnables import RunnableConfig
from threadplane.middleware.langgraph import emit_custom_event
 
async def analysis_node(state: State, config: RunnableConfig) -> State:
    # Emit a partial result as the node runs
    await emit_custom_event(
        "analysis_progress", {"step": "scoring", "pct": 42}, config=config
    )
 
    # ... do more work ...
 
    await emit_custom_event(
        "analysis_progress", {"step": "scoring", "pct": 100}, config=config
    )
    return state

The signature is emit_custom_event(name, value, *, config=None). Pass config when the node already receives one; omit it and the ambient run context is used. Backends that do not depend on the middleware package can call adispatch_custom_event from langchain_core.callbacks directly — the helper adds no wire behavior of its own.

The event name becomes CustomStreamEvent.name and the payload becomes CustomStreamEvent.data. This is the mechanism the subagents example uses to stream child-agent tokens from a callback handler.

Graph state is a different signal

The other way for an ag-ui-langgraph node to push data mid-run is to return it as a top-level graph state field. The bridge auto-emits state as STATE_SNAPSHOT and STATE_DELTA frames, and the adapter reduces those into agent.state() — not into customEvents. Reach for state when the client needs the current value of something, and for a CUSTOM event when the client needs the individual occurrences. The Generative UI guide takes the state route for exactly this reason.

Reading Custom Events in Angular

injectAgent() returns an AgUiAgent, so the customEvents signal is available directly on the injected agent — no cast needed.

Reactive effect

Use an effect to react every time new events arrive:

import { Component, ChangeDetectionStrategy, effect, signal } from '@angular/core';
import { ChatComponent } from '@threadplane/chat';
import { injectAgent } from '@threadplane/ag-ui';
 
@Component({
  standalone: true,
  imports: [ChatComponent],
  changeDetection: ChangeDetectionStrategy.OnPush,
  template: `
    <chat [agent]="agent" />
    @if (progress() !== null) {
      <progress-bar [value]="progress()" />
    }
  `,
})
export class AnalysisComponent {
  protected readonly agent = injectAgent();
  protected readonly progress = signal<number | null>(null);
 
  constructor() {
    effect(() => {
      const events = this.agent.customEvents();
      const last = [...events]
        .reverse()
        .find((e) => e.name === 'analysis_progress');
      this.progress.set(
        last ? (last.data as { pct: number }).pct : null,
      );
    });
  }
}

Computed signal

When you only need to derive a value, computed is more concise:

import { Component, ChangeDetectionStrategy, computed } from '@angular/core';
import { ChatComponent } from '@threadplane/chat';
import { injectAgent } from '@threadplane/ag-ui';
 
@Component({
  standalone: true,
  imports: [ChatComponent],
  changeDetection: ChangeDetectionStrategy.OnPush,
  template: `
    <chat [agent]="agent" />
    <event-log [events]="progressEvents()" />
  `,
})
export class AnalysisComponent {
  protected readonly agent = injectAgent();
 
  protected readonly progressEvents = computed(() =>
    this.agent.customEvents().filter(
      (e) => e.name === 'analysis_progress',
    ),
  );
}

Both patterns are zoneless-safe: Angular's signal graph tracks the customEvents() read and re-evaluates the derived value automatically.

Relation to Interrupts

CUSTOM events named on_interrupt follow a separate path: the adapter routes them to agent.interrupt (a Signal<AgentInterrupt | undefined>) and they never enter customEvents. This keeps the two signals purpose-distinct — interrupt drives human-in-the-loop approval flows, while customEvents carries all other backend-pushed data.

See the Interrupts guide for the full interrupt lifecycle including <chat-approval-card> and submit({ resume }).

See Also

• Architecture — how the adapter reduces protocol events into Angular signals
• Event Mapping — full table of AG-UI event types and the agent fields they populate
• injectAgent() — the injection function that returns AgUiAgent