How AgentProcessEvent Streaming Enables Real-Time Agent Progress Monitoring in Embabel

Embabel's agent runtime publishes lifecycle events to the Spring ApplicationContext, where EmbabelSpanEventListener captures them in a thread-local EmbabelObservationContext and streams them to clients via Server-Sent Events (SSE) for real-time monitoring.

The embabel/embabel-agent repository implements a sophisticated event-driven architecture that decouples agent execution from progress observation. By leveraging Spring's application event infrastructure alongside Server-Sent Events (SSE), the framework enables true real-time monitoring of agent runs without blocking execution threads. This article examines the internal mechanics of AgentProcessEvent streaming, from event generation in the agent core to consumption via HTTP streaming endpoints.

The AgentProcessEvent Hierarchy

The streaming system centers on AgentProcessEvent, an abstract base class representing discrete moments in an agent's lifecycle. Concrete implementations include AgentProcessCreationEvent, AgentProcessCompletedEvent, AgentProcessWaitingEvent, ToolResponseEvent, and LlmInvocationEvent.

Each event type carries contextual metadata about the specific lifecycle phase. The agent runtime—particularly components like StreamingPromptRunnerBuilder—emits these events via Spring's ApplicationEventPublisher, broadcasting them to registered listeners within the application context.

Event Capture and Storage

EmbabelSpanEventListener

The [EmbabelSpanEventListener.java](https://github.com/embabel/embabel-agent/blob/main/embabel-agent-observability/src/main/java/com/embabel/agent/observability/tracing/EmbabelSpanEventListener.java) class serves as the central interception point for all process events. Located in the embabel-agent-observability module, this listener implements the onProcessEvent(Event) handler method.

When the agent runtime publishes an event, this listener enriches it with tracing and metric tags, then appends it to the current EmbabelObservationContext. This design ensures that every tool invocation, LLM call, and state transition is recorded without impacting the agent's execution semantics.

EmbabelObservationContext

The [EmbabelObservationContext.java](https://github.com/embabel/embabel-agent/blob/main/embabel-agent-observability/src/main/java/com/embabel/agent/observability/tracing/EmbabelObservationContext.java) maintains a thread-local store that accumulates events for the active agent run. Because it uses thread-local storage, the context isolate events between concurrent agent executions, preventing cross-contamination of process state.

This context provides the source data for SSE streaming, holding the complete event history until the agent run completes or the client disconnects.

SSE Streaming Implementation

ProcessEventsEndpoint

The [ProcessEventsEndpoint.java](https://github.com/embabel/embabel-agent/blob/main/embabel-agent-common/embabel-agent-webmvc/src/main/java/com/embabel/agent/web/sse/ProcessEventsEndpoint.java) exposes the /process/events endpoint using Spring's SseEmitter. When a client initiates an HTTP connection to this endpoint, the controller reads the current EmbabelObservationContext and begins streaming.

The emitter remains open for the duration of the agent run, pushing each recorded event as an SSE message via SseEmitter.send(). Because Server-Sent Events provide a unidirectional, push-based stream over standard HTTP, clients receive updates instantaneously as the EmbabelSpanEventListener records them.

Implementing Real-Time Monitoring

Enabling Event Streaming in Prompt Runners

To emit process events from your agent, configure the runner using StreamingPromptRunnerBuilder:

// Build a streaming prompt runner that forwards process events
var runner = StreamingPromptRunnerBuilder.builder()
        .prompt("You are a helpful assistant…")
        .streamEvents(true)                 // enable event streaming
        .build()
        .run();                               // blocks until completion

Setting streamEvents(true) ensures the runner emits AgentProcessEvent subclasses to the application context, making them available to the SSE endpoint.

Consuming Events via JavaScript

Clients connect to the SSE endpoint to receive real-time updates:

const evtSource = new EventSource('/process/events?runId=abc123');

evtSource.onmessage = (e) => {
  const ev = JSON.parse(e.data);
  console.log('Agent event:', ev.type, ev.payload);
};

evtSource.onerror = (e) => {
  console.error('SSE error', e);
};

The EventSource API maintains a persistent connection, invoking the onmessage handler each time the server pushes a new event from the EmbabelObservationContext.

Custom Event Listeners

You can extend monitoring capabilities by implementing custom listeners that intercept events before they reach the SSE stream:

@Component
public class MyProcessListener implements EmbabelSpanEventListener {

    @Override
    public void onProcessEvent(AgentProcessEvent event) {
        // add custom tag
        event.addAttribute("myTag", "value");
        // still delegate to the default listener so the SSE stream works
        EmbabelSpanEventListener.super.onProcessEvent(event);
    }
}

This pattern allows you to enrich events with domain-specific metadata while preserving the default streaming behavior to the /process/events endpoint.

Summary

• AgentProcessEvent subclasses represent discrete lifecycle moments (creation, tool calls, LLM invocations, completion) emitted by the agent runtime via Spring's event bus.
• EmbabelSpanEventListener intercepts all events in EmbabelObservationContext, enabling accumulation of execution history in thread-local storage.
• ProcessEventsEndpoint exposes an SSE stream that pushes events from the context to HTTP clients in real-time as they occur.
• The architecture decouples event generation from consumption, allowing multiple observers to monitor agent progress without impacting execution performance.

Frequently Asked Questions

What specific event types are available in AgentProcessEvent streaming?

The framework provides concrete event types including AgentProcessCreationEvent, AgentProcessCompletedEvent, AgentProcessWaitingEvent, ToolResponseEvent, and LlmInvocationEvent. Each type extends the base AgentProcessEvent class and carries payload specific to that lifecycle phase, such as tool outputs or LLM token counts.

How does the system handle concurrent agent runs without event mixing?

EmbabelObservationContext uses thread-local storage to isolate event collections per execution thread. This ensures that events from one agent run never contaminate another's event stream, even when multiple agents execute simultaneously within the same JVM.

Can multiple clients monitor the same agent run simultaneously?

Yes. Because events are stored in the thread-local EmbabelObservationContext and served through stateless SSE connections, any number of HTTP clients can connect to /process/events for the same run ID. Each client receives an independent stream of events from the shared context without affecting the agent's execution or other observers.

Is it possible to filter events before they reach the SSE stream?

While the default EmbabelSpanEventListener captures all events, you can implement custom filtering by extending the listener pattern. Create a component that implements EmbabelSpanEventListener, inspect the event type in onProcessEvent(), and conditionally call super.onProcessEvent() or apply custom logic. However, for client-side filtering, implement the logic in your JavaScript event handler to avoid losing observability data on the server.

{
  "mcpServers": {
    "instagit": {
      "command": "npx",
      "args": ["-y", "instagit@latest"]
    }
  }
}