Class OpenAIFormatter

examples/server/server.py:7815–10256

class OpenAIFormatter:
    @dataclass
    class ReturnedToken:
        index: int
        text_bytes: bytes
        token: Token
        text_offset: int

    @dataclass
    class ResponsesOutputItem:
        output_index: int
        item: Dict[str, Any]
        content_index: Optional[int] = None

    @dataclass
    class ResponsesStream:
        body: CreateResponseRequest
        response_id: str
        created_at: float
        model: str
        output: List[Dict[str, Any]] = field(default_factory=list)
        sequence_number: int = 0
        started: bool = False
        assistant_meta: Dict[str, Any] = field(default_factory=dict)
        reasoning_item: Optional["OpenAIFormatter.ResponsesOutputItem"] = None
        message_item: Optional["OpenAIFormatter.ResponsesOutputItem"] = None
        tool_items: Dict[int, "OpenAIFormatter.ResponsesOutputItem"] = field(
            default_factory=dict
        )
        final_status: Optional[str] = None
        incomplete_details: Optional[Dict[str, Any]] = None

    def __init__(self, model: Model) -> None:
        self.model = model

    @staticmethod
    def decode_text(data: bytes) -> str:
        return data.decode("utf-8", errors="ignore")

    @staticmethod
    def encode_sse_payload(payload: BaseModel | Dict[str, Any]) -> bytes:
        data = (
            payload.model_dump(mode="json", exclude_none=True)
            if isinstance(payload, BaseModel)
            else payload
        )
        return (
            "data: "
            f"{json.dumps(data, ensure_ascii=False, separators=(',', ':'))}\n\n"
        ).encode("utf-8")

    @staticmethod
    def next_stream_chunk(
        stream: Iterator[CompletionChunk],
    ) -> Tuple[bool, Optional[CompletionChunk]]:
        try:
            return False, next(stream)
        except StopIteration: