Server-Sent Event Tool Stream Parsing Prompt Template

This prompt is for real-time agent developers who need to consume, classify, and act on Server-Sent Event (SSE) streams from external tools. The core job is to turn a raw, potentially unreliable stream of text events into structured, actionable tool-call lifecycle states—such as tool_start, tool_progress, tool_result, and tool_error—while maintaining connection health and ordering guarantees. The ideal user is an AI platform engineer or integration developer building a production agent harness that must survive connection drops, out-of-order events, and malformed payloads without corrupting the agent's downstream state.

Use this prompt when your agent directly consumes SSE from an API, an MCP server, or a long-running tool endpoint, and you need the model to act as a programmable stream parser. This is appropriate when the stream protocol is known but the event types, reconnection logic, and error-recovery rules must be enforced by the agent itself rather than by a separate middleware service. Do not use this prompt for simple request-response tool calls, for WebSocket-based streams that require binary frame handling, or for streams where the tool's event schema is completely unknown and must be discovered at runtime. The prompt assumes you can provide the expected event type catalog, the Last-Event-ID tracking rules, and the reconnection policy as part of the input context.

Before deploying, validate the prompt against at least three failure scenarios: a mid-stream connection drop with partial event delivery, an event with a malformed data field that violates the expected schema, and a stream that resumes after a reconnect with an out-of-order id. If the agent's downstream actions include state mutations, database writes, or user-visible conclusions, require a human review gate or a strict idempotency check before acting on the parsed stream output. This prompt is a parsing and classification layer—it does not replace the application logic that consumes the parsed events.

This prompt template is designed for real-time agent developers who need to consume and interpret Server-Sent Event (SSE) streams from tool calls. It instructs the model to act as a stream parser that classifies event types, tracks last-event-id for reconnection state, and handles multiplexed streams where multiple logical event sequences arrive over a single connection. The template is structured to produce a deterministic JSON output that your application can consume directly, making it suitable for wiring into an AI harness that processes live tool streams.

textCopy
You are an SSE stream parser for a real-time agent system. Your job is to consume a raw Server-Sent Event stream from a tool call and produce a structured analysis of the stream state.

## INPUT
Raw SSE stream text:
[SSE_STREAM_INPUT]

## CONTEXT
- The stream may contain multiple event types: `data`, `event`, `id`, `retry`, and comment lines starting with `:`.
- The stream may be multiplexed, containing events from multiple logical streams identified by an `event` field prefix (e.g., `stream1:data`, `stream2:data`).
- The `id` field carries the last-event-id for reconnection. You must track the most recent `id` value seen.
- The `retry` field specifies the reconnection time in milliseconds.
- The stream may be truncated mid-chunk. You must identify incomplete events.

## OUTPUT SCHEMA
Produce a valid JSON object with the following structure:
{
  "events": [
    {
      "event_type": "string", // The event type, or "unknown" if not specified
      "data": "string", // The data payload
      "id": "string | null", // The event ID if present
      "stream_id": "string | null" // The logical stream identifier if multiplexed
    }
  ],
  "last_event_id": "string | null", // The most recent event ID for reconnection
  "retry_ms": "number | null", // The reconnection interval if specified
  "truncated": "boolean", // True if the input ended mid-event
  "multiplexed_streams": ["string"] // List of unique logical stream IDs detected
}

## CONSTRAINTS
- Do not invent events not present in the input.
- If the input is empty or only contains comments, return an empty events array.
- For multiplexed streams, extract the stream ID from the event type prefix before the colon.
- If a data field spans multiple lines, join them with newlines.
- If the input ends with an incomplete event (no blank line terminator), set truncated to true and include the partial event if possible.

## EXAMPLES
Input:
id: 1\nevent: stream1:data\ndata: {\"value\": 42}\n\nid: 2\nevent: stream2:data\ndata: {\"value\": 99}\n\n
Output:
{
  "events": [
    {"event_type": "stream1:data", "data": "{\"value\": 42}", "id": "1", "stream_id": "stream1"},
    {"event_type": "stream2:data", "data": "{\"value\": 99}", "id": "2", "stream_id": "stream2"}
  ],
  "last_event_id": "2",
  "retry_ms": null,
  "truncated": false,
  "multiplexed_streams": ["stream1", "stream2"]
}

## RISK LEVEL
[RISK_LEVEL]

Parse the input stream and return only the JSON object.

To adapt this template for your application, replace the [SSE_STREAM_INPUT] placeholder with the raw text from your SSE connection. Set [RISK_LEVEL] to guide the model's handling of ambiguous or malformed events—use low for best-effort parsing in non-critical paths, or high to instruct the model to flag anomalies explicitly and avoid guessing. For production use, always validate the output JSON against your expected schema before passing parsed events to downstream agent logic. If your stream contains sensitive data, ensure the SSE input is sanitized before reaching the model, or run this prompt in a local deployment where data does not leave your environment.

This prompt is designed to sit inside a real-time agent's event ingestion pipeline, not as a standalone chat interaction. The model receives raw Server-Sent Event text—potentially interleaved from multiple streams—and must produce structured classification, state tracking, and multiplexing decisions that downstream code can act on. The implementation harness wraps this prompt in a thin application layer that handles connection lifecycle, buffers partial event text, calls the model when a decision boundary is reached, validates the structured output, and applies the resulting instructions to the active stream handlers.

The core integration pattern is a decision-point dispatcher. Your SSE client accumulates raw event lines in a per-stream buffer. When a decision boundary is reached—such as a dispatch event, a reconnection signal, an id field change, or a buffer-size threshold—the harness assembles the prompt with the current [RAW_EVENT_TEXT], [STREAM_STATE] (including lastEventId, reconnectStatus, and activeStreamIds), and [OUTPUT_SCHEMA] (a strict JSON schema for the expected classification and action payload). The model response is parsed and validated before any action is taken. If validation fails, the harness retries with the validation error injected into [PREVIOUS_ERRORS]. After a configurable retry limit (typically 2), the harness falls back to a safe default: log the raw event, increment a parse_failure metric, and either drop the event or route it to a dead-letter queue depending on the [RISK_LEVEL] parameter.

For connection-drop recovery, the harness must track lastEventId outside the prompt and feed it back in on reconnect. The prompt's reconnection classification output tells the harness whether to replay from the last acknowledged ID or start fresh. Implement this as a state machine with three modes: NORMAL, RECONNECTING, and DRAINING. The prompt's output transitions between these modes. Log every mode transition with the triggering event text and the model's classification for post-incident trace analysis. For stream multiplexing, maintain a map of streamId -> buffer and only pass the buffer for the stream that triggered the decision boundary, plus a summary of other active stream states. This prevents cross-stream contamination while giving the model enough context to detect ordering anomalies across streams.

Model choice matters here. This is a low-latency, high-reliability classification task. Use a fast model (e.g., GPT-4o-mini, Claude Haiku, or a fine-tuned small open-weight model) with response_format set to JSON mode and a low temperature (0.0–0.1). Do not use a large reasoning model for this hot path—the added latency will break real-time stream processing guarantees. Instrument every model call with a trace span that captures: prompt version, input token count, output token count, latency, validation pass/fail, retry count, and the resulting action. This trace data is essential for debugging silent stream corruption and for tuning buffer-size thresholds.

Before shipping, build an eval harness that replays recorded SSE streams—including connection drops, out-of-order events, duplicate IDs, and malformed event text—and asserts that the harness produces the correct classifications, state transitions, and multiplexing decisions. Add adversarial cases: empty event data, events with only comments, streams that never send a dispatch signal, and rapid-fire reconnections. The eval should also measure end-to-end latency from event arrival to action dispatch, with a p99 target appropriate for your real-time requirements. If the prompt or harness logic changes, rerun the full eval suite and compare trace-by-trace before deployment.