How the Streaming Response Works in the OpenAPI Server

The OpenAPI server implements streaming responses using Spring's SseEmitter to maintain an open HTTP connection, asynchronously processing agent events through StreamAction while sending periodic ping keep-alives to prevent timeouts.

The junjiem/dat repository provides an AI data analysis tool that streams real-time responses through its OpenAPI server. Understanding how the streaming response works in the OpenAPI server is essential for building responsive client applications that consume agent outputs, SQL generation results, and incremental answers as they become available.

Core Architecture of the Streaming Endpoint

The streaming endpoint is implemented in AskController at /api/v1/ask/stream. When a client POSTs an AskRequest, the controller creates a SseEmitter and returns it immediately, allowing the HTTP connection to stay open while events are pushed to the client.

The AskController Entry Point

Located in dat-servers/dat-server-openapi/src/main/java/ai/dat/server/openapi/controller/AskController.java, the controller handles the initial request setup:

@PostMapping(value = "/stream", produces = MediaType.TEXT_EVENT_STREAM_VALUE)
public SseEmitter askStream(@Valid @RequestBody AskRequest request) {
    String conversationId = (request.getConversationId() == null || request.getConversationId().isBlank())
            ? UUID.randomUUID().toString()
            : request.getConversationId();

    SseEmitter emitter = new SseEmitter();
    // start ping & stream processing (see full implementation in AskController)
    …
    return emitter;
}

The method generates a unique conversationId if not provided, initializes the SseEmitter, and returns it immediately to the client while background processing continues.

Connection Keep-Alive with Scheduled Ping

To prevent HTTP timeouts during long-running agent operations, a ScheduledExecutorService (pingScheduler) sends a ping event every 10 seconds:

ScheduledFuture<?> pingTask = pingScheduler.scheduleAtFixedRate(() -> {
    emitter.send(SseEmitter.event()
        .name(PING_EVENT)
        .data(Map.of(TIMESTAMP, System.currentTimeMillis(),
                    CONVERSATION_ID, conversationId)));
}, 0, 10, TimeUnit.SECONDS);

Source: AskController – ping scheduler

If the ping fails (indicating the client has disconnected), the task is cancelled to prevent resource leaks.

Asynchronous Stream Processing

The controller delegates actual stream processing to a background thread pool so the request-handling thread can return immediately.

Background Thread Execution

A cached thread-pool (streamExecutor) runs processStreamEvents(...) asynchronously:

streamExecutor.execute(() ->
    processStreamEvents(emitter, conversationId, request, histories, pingTask));

Source: AskController – async processing

The StreamAction and Event Iteration

The processStreamEvents method obtains a StreamAction from ProjectService:

StreamAction action = runnerService.ask(conversationId,
    request.getAgentName(), request.getQuestion(), histories);

Source: AskController – get StreamAction

ProjectService creates or re-uses a ProjectRunner that holds the agent state. The StreamAction implements Iterable<StreamEvent>, allowing the controller to iterate over the agent's output:

for (StreamEvent event : action) {
    // translate each StreamEvent into an SSE payload
    sendStreamEvent(emitter, event, eventId, conversationId);
}

Source: StreamAction iterator

Mapping StreamEvents to SSE Messages

The sendStreamEvent method in AskController translates each StreamEvent into a specific SSE event name based on which data fields are present.

Event Type Detection Logic

Each StreamEvent carries optional pieces of data (incremental answer text, semantic SQL, query results, tool calls, etc.). The controller maps these to specific SSE event names:

SSE name	Triggered by
ping	Scheduled ping task
sql_generate	event.getSemanticSql()
semantic_to_sql	event.getQuerySql()
sql_execute	event.getQueryData()
agent_answer	event.getIncrementalContent()
agent_answer_end	End of incremental answer sequence
before_tool_execution	Tool request (no result yet)
tool_execution	Tool result present
hitl_ai_request	Human-in-the-loop AI request
hitl_tool_approval	Human-in-the-loop tool approval
error	Exceptions during processing
finished	Normal or error termination
other	Any additional custom event

Source: sendStreamEvent logic

Termination and Cleanup

After the loop finishes, a finished event is emitted with status succeeded (or failed if an exception occurred) and the SSE connection is closed via emitter.complete() or completeWithError(). The ping task is cancelled in the finally block to guarantee cleanup.

Source: processStreamEvents – finalisation

Client-Side Implementation Example

Clients consume the stream using standard SSE protocols. The following JavaScript example demonstrates how to handle the various event types:

const payload = {
  conversationId: '',
  agentName: 'my-agent',
  question: 'Show the total sales per month',
  // … other fields if needed
};

fetch('/api/v1/ask/stream', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify(payload)
}).then(res => {
  const evtSource = new EventSource(res.url); // SSE URL returned by the server

  evtSource.addEventListener('ping', e => console.log('ping', JSON.parse(e.data)));
  evtSource.addEventListener('agent_answer', e => console.log('chunk', JSON.parse(e.data).answer));
  evtSource.addEventListener('agent_answer_end', e => console.log('answer finished'));
  evtSource.addEventListener('sql_execute', e => console.log('query result', JSON.parse(e.data).query_data));
  evtSource.addEventListener('error', e => console.error('stream error', JSON.parse(e.data)));
  evtSource.addEventListener('finished', e => {
    console.log('stream done', JSON.parse(e.data).status);
    evtSource.close();
  });
});

The client reconstructs the full answer from incremental agent_answer messages and handles HITL prompts (hitl_ai_request, hitl_tool_approval) similarly.

A typical SSE payload for an agent answer chunk looks like this:

event: agent_answer
data: {"conversation_id":"c1f2…","timestamp":1719915600000,"answer_id":"a3d9…","answer":"We are analyzing the data for you …"}

Key Source Files and Implementation Details

File	Role	GitHub link
AskController.java	Exposes /stream endpoint, creates SseEmitter, schedules ping, dispatches StreamEvents.	AskController
ProjectService.java	Looks up or creates a ProjectRunner per conversation, forwards the ask request.	ProjectService (openapi)
StreamAction.java	Implements Iterable<StreamEvent>; wraps the agent's processing pipeline.	StreamAction
StreamEvent.java	Holds the optional data pieces (incremental answer, SQL, tool calls, etc.) that are turned into SSE messages.	StreamEvent
ProjectRunner.java (used indirectly)	Executes the agent logic (ask, userResponse, userApproval).	ProjectRunner

Summary

• SseEmitter and ScheduledExecutorService maintain the HTTP connection through periodic ping events every 10 seconds.
• ProjectService and ProjectRunner drive the agent's execution, returning a StreamAction that implements Iterable<StreamEvent>.
• The controller iterates over StreamEvent objects and maps them to specific SSE event names based on which data fields are present.
• A cached thread pool processes the stream asynchronously, allowing the request thread to return the SseEmitter immediately.
• The connection closes with a finished event, and the ping task is cancelled in a finally block to prevent resource leaks.

Frequently Asked Questions

What protocol does the OpenAPI server use for streaming?

The server uses Server-Sent Events (SSE), a standard HTTP-based protocol where the client opens a persistent connection and the server pushes events as they occur. This is implemented through Spring's SseEmitter class with MediaType.TEXT_EVENT_STREAM_VALUE, allowing real-time delivery of agent outputs without WebSocket complexity.

How does the server prevent connection timeouts during long-running agent operations?

A ScheduledExecutorService sends a ping event every 10 seconds to keep the HTTP connection alive. If a ping fails (indicating client disconnection), the scheduled task is cancelled immediately. Additionally, all stream processing occurs in a background thread pool, ensuring the initial request thread returns quickly while the connection remains open for the duration of agent execution.

What types of events can clients expect from the stream?

Clients receive various event types including agent_answer (incremental text chunks), agent_answer_end (completion signal), sql_generate (semantic SQL), semantic_to_sql (translated SQL), sql_execute (query results), tool_execution (tool results), hitl_ai_request and hitl_tool_approval (human-in-the-loop prompts), error (exceptions), and finished (stream termination).

How is resource cleanup handled when a client disconnects?

The ping scheduler detects failed send attempts and cancels the keep-alive task. Additionally, the processStreamEvents method includes a finally block that guarantees the ping task is cancelled and the SseEmitter is completed (either normally or with error) when the stream ends, encounters an exception, or the client disconnects.

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