Skip to content

symfonic.agent.fastapi.typed_event_codec

typed_event_codec

Frozen class-name and JSON wire encoding for compatibility typed SSE.

encode_typed_event 

encode_typed_event(stream_event: Any) -> dict[str, str]

Encode one StreamEvent exactly as the legacy typed route does.

Source code in src/symfonic/agent/fastapi/typed_event_codec.py 
 9
10
11
12
13
14
15
16
def encode_typed_event(stream_event: Any) -> dict[str, str]:
    """Encode one StreamEvent exactly as the legacy typed route does."""
    from symfonic.core.streaming.serialization import _event_to_dict

    return {
        "event": type(stream_event).__name__,
        "data": json.dumps(_event_to_dict(stream_event), default=str),
    }