Class StreamingResponse

src/mcp_cli/chat/streaming_handler.py:44–83

Container for streaming response data. Pydantic model for type-safe streaming responses.

class StreamingResponse(BaseModel):
    """Container for streaming response data.

    Pydantic model for type-safe streaming responses.
    """

    content: str = Field(description="Response content")
    tool_calls: list[dict[str, Any]] = Field(
        default_factory=list, description="Tool calls from the model"
    )
    chunks_received: int = Field(default=0, description="Number of chunks processed")
    elapsed_time: float = Field(description="Time taken for the response")
    interrupted: bool = Field(
        default=False, description="Whether streaming was interrupted"
    )
    reasoning_content: str | None = Field(
        default=None, description="Reasoning content (for DeepSeek reasoner)"
    )
    streaming: bool = Field(
        default=True, description="Whether this was a streaming response"
    )
    usage: dict[str, int] | None = Field(
        default=None,
        description="Token usage from provider (input_tokens, output_tokens)",
    )

    model_config = {"frozen": False}

    def to_dict(self) -> dict[str, Any]:
        """Convert to dict for backwards compatibility using enums."""
        return {
            StreamingResponseField.RESPONSE: self.content,
            StreamingResponseField.TOOL_CALLS: self.tool_calls,
            StreamingResponseField.CHUNKS_RECEIVED: self.chunks_received,
            StreamingResponseField.ELAPSED_TIME: self.elapsed_time,
            StreamingResponseField.STREAMING: self.streaming,
            StreamingResponseField.INTERRUPTED: self.interrupted,
            StreamingResponseField.REASONING_CONTENT: self.reasoning_content,
            StreamingResponseField.USAGE: self.usage,
        }


class ToolCallAccumulator: