Class ServerSentEvent

src/anthropic/_streaming.py:361–410

class ServerSentEvent:
    def __init__(
        self,
        *,
        event: str | None = None,
        data: str | None = None,
        id: str | None = None,
        retry: int | None = None,
        raw: list[str] | None = None,
    ) -> None:
        if data is None:
            data = ""

        self._id = id
        self._data = data
        self._event = event or None
        self._retry = retry
        self._raw = raw if raw is not None else []

    @property
    def event(self) -> str | None:
        return self._event

    @property
    def id(self) -> str | None:
        return self._id

    @property
    def retry(self) -> int | None:
        return self._retry

    @property
    def data(self) -> str:
        return self._data

    @property
    def raw(self) -> list[str]:
        """The original wire lines this event was decoded from, without trailing newlines.

        Includes SSE fields the decoder does not otherwise model (comment lines,
        unknown fields). Empty for events that were constructed rather than decoded.
        """
        return self._raw

    def json(self) -> Any:
        return json.loads(self.data)

    @override
    def __repr__(self) -> str:
        return f"ServerSentEvent(event={self.event}, data={self.data}, id={self.id}, retry={self.retry})"


class SSEDecoder: