Skip to content

symfonic.platform.transport_sse_response

transport_sse_response

ASGI-owned status choice and cleanup for native SSE turns.

The first agent event is pulled while the disconnect watcher is active and before response.start. A refusal on that pull can therefore use a real HTTP status. Once headers are sent, the existing stream projector owns the wire shape and terminal error frame.

NativeStreamResponse 

NativeStreamResponse(agent: Any, query: str, history: tuple[Any, ...], session_id: str, prepared: Any, *, typed: bool)

Bases: StreamingResponse

Choose status before headers and close a primed stream on every exit.

Source code in src/symfonic/platform/transport_sse_response.py 
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
def __init__(
    self,
    agent: Any,
    query: str,
    history: tuple[Any, ...],
    session_id: str,
    prepared: Any,
    *,
    typed: bool,
) -> None:
    super().__init__(iter(()), media_type="text/event-stream")
    # Exact type only: subclasses and duck-typed hosts may override stream
    # with cleanup that has no kernel-owned finalizer budget.
    from symfonic.agent.facade import Agent

    self._trusted_source = type(agent) is Agent
    self._agent = agent
    self._query = query
    self._history = history
    self._session_id = session_id
    self._prepared = prepared
    self._typed = typed