stream_events

stream_events

Typed SSE event dataclasses for the async streaming agent loop.

AgentEvent module-attribute

AgentEvent = Union[
    TokenEvent,
    ToolCallEvent,
    ToolResultEvent,
    SourceEvent,
    DoneEvent,
    ErrorEvent,
]

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

SSEEvent module-attribute

SSEEvent = dict[str, str]

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

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.

SourceEvent dataclass

Emitted after RAG retrieval, before the tool loop.

DoneEvent dataclass

Emitted when the agent loop completes.

ErrorEvent dataclass

Yielded when an exception is caught inside the async generator.

agent_to_sse

agent_to_sse(event: AgentEvent) -> SSEEvent

Convert an internal AgentEvent to an SSE dictionary for streaming.

Maps each event dataclass type to its corresponding SSE event name ("token", "source", "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 async

heartbeat_generator(interval: int = 15)

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.