stream_events

Typed SSE event dataclasses for the async streaming agent loop.

AgentEvent = Union[TokenEvent, ToolCallEvent, ToolResultEvent, TruncatedEvent, SourceEvent, StatusEvent, DoneEvent, ErrorEvent] module-attribute

Union type for all events emitted by the agent's process_stream() generator.

SSEEvent = dict[str, str] module-attribute

Dictionary with 'event' (string name) and 'data' (JSON string) keys for SSE streaming.

DoneEvent dataclass

Emitted when the agent loop completes.

ErrorEvent dataclass

Yielded when an exception is caught inside the async generator.

SourceEvent dataclass

Emitted after RAG retrieval, before the tool loop.

StatusEvent dataclass

Emitted before long-running phases so clients can show progress.

TokenEvent dataclass

A single content delta from DeepSeek streaming.

ToolCallEvent dataclass

Emitted when the LLM requests a tool call.

ToolResultEvent dataclass

Emitted after synchronous tool execution completes.

TruncatedEvent dataclass

Emitted when the API stops a stream with finish_reason='length'.

Signals that the response was cut off by the max_tokens budget. The agent consumes this internally (never forwards it to SSE clients) to continue the reply with a follow-up completion instead of storing a partial answer.

agent_to_sse(event)

Convert an internal AgentEvent to an SSE dictionary for streaming.

Maps each event dataclass type to its corresponding SSE event name ("token", "source", "status", "tool_call", "tool_result", "done", "error") and serializes the event data as JSON in the data field.

Parameters:

Name	Type	Description	Default
event	AgentEvent	Any AgentEvent subclass (TokenEvent, SourceEvent, etc.).	required

Returns:

Type	Description
SSEEvent	Dictionary with "event" (string name) and "data" (JSON string) keys,
SSEEvent	suitable for use with sse_starlette's EventSourceResponse.

heartbeat_generator(interval=15) async

Async generator that yields heartbeat SSE events at a fixed interval.

Used to keep SSE connections alive during long-running agent processing. Yields {"event": "heartbeat", "data": {"t": timestamp}}.

Parameters:

Name	Type	Description	Default
interval	int	Seconds between heartbeat events. Defaults to 15.	15

Yields:

Type	Description
	Dict[str, str]: SSE heartbeat event dictionary.