Skip to content

Sse Models

src.api.sse_models.AgentStatus

Bases: str, Enum

Source code in src/api/sse_models.py 
11
12
13
14
15
16
class AgentStatus(str, Enum):
    STARTING = "starting_generation"
    TOOL_DETECTED = "tool_call_detected"
    TOOLS_EXECUTED = "tools_executed"
    MAX_DEPTH = "max_depth_reached"
    CONTINUING = "continuing_generation"

src.api.sse_models.SSEStatus

Bases: BaseModel

Source code in src/api/sse_models.py 
19
20
21
class SSEStatus(BaseModel):
    status: AgentStatus
    details: Optional[Dict[str, Any]] = None

src.api.sse_models.SSEFunction

Bases: BaseModel

Model for function calls in SSE responses, with support for streaming chunks.

Source code in src/api/sse_models.py 
24
25
26
27
class SSEFunction(BaseModel):
    """Model for function calls in SSE responses, with support for streaming chunks."""
    name: str = ""
    arguments: str = ""

src.api.sse_models.SSEToolCall

Bases: BaseModel

Model for tool calls in SSE responses.

Source code in src/api/sse_models.py 
30
31
32
33
34
35
class SSEToolCall(BaseModel):
    """Model for tool calls in SSE responses."""
    index: int = 0
    id: Optional[str] = None
    type: str = "function"
    function: Optional[SSEFunction] = None

src.api.sse_models.SSEDelta

Bases: BaseModel

Model for delta content in SSE responses.

Source code in src/api/sse_models.py 
38
39
40
41
42
43
44
45
class SSEDelta(BaseModel):
    """Model for delta content in SSE responses."""
    role: Optional[str] = None
    content: Optional[str] = None
    tool_calls: Optional[List[SSEToolCall]] = None
    refusal: Optional[str] = None
    status: Optional[str] = None
    metadata: Optional[dict] = None

src.api.sse_models.SSEChoice

Bases: BaseModel

Model for choices in SSE responses.

Source code in src/api/sse_models.py 
48
49
50
51
52
53
class SSEChoice(BaseModel):
    """Model for choices in SSE responses."""
    index: int
    delta: SSEDelta
    logprobs: Optional[dict] = None
    finish_reason: Optional[str] = None

src.api.sse_models.SSEChunk

Bases: BaseModel

Model for SSE chunks.

Source code in src/api/sse_models.py 
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
class SSEChunk(BaseModel):
    """Model for SSE chunks."""
    id: str
    object: str
    created: int
    model: str
    service_tier: Optional[str] = None
    system_fingerprint: Optional[str] = None
    choices: List[SSEChoice]
    thread_id: Optional[str] = None  # this is an IBM wxO thing

    @staticmethod
    def make_text_chunk(text: str) -> 'SSEChunk':
        """
        Utility to create a minimal SSEChunk that only has user-visible 'content'.
        This ensures we never leak partial function-call details back to the user.
        """
        return SSEChunk(
            id=f"chatcmpl-{time.time()}",
            object="chat.completion.chunk",
            created=int(time.time()),
            model="agent-01",
            choices=[
                SSEChoice(
                    index=0,
                    delta=SSEDelta(role="assistant", content=text),
                    finish_reason=None
                )
            ]
        )

    @staticmethod
    async def make_status_chunk(status: str, extra_info: Optional[Dict] = None) -> 'SSEChunk':
        metadata = {"status": status}
        if extra_info:
            metadata.update(extra_info)

        return SSEChunk(
            id=f"status_{time.time()}",
            object="chat.completion.chunk",
            created=int(time.time()),
            model="agent-01",
            choices=[
                SSEChoice(
                    index=0,
                    delta=SSEDelta(
                        role="system",
                        metadata=metadata
                    ),
                    finish_reason=None
                )
            ]
        )

    @staticmethod
    async def make_stop_chunk(content=None, refusal=None) -> 'SSEChunk':
        return SSEChunk(
            id=f"chatcmpl-{time.time()}",
            object="chat.completion.chunk",
            created=int(time.time()),
            model="agent-01",
            choices=[
                SSEChoice(
                    index=0,
                    delta=SSEDelta(role="assistant", content=content, refusal=refusal),
                    finish_reason="stop"
                )
            ]
        )

make_text_chunk(text) staticmethod

Utility to create a minimal SSEChunk that only has user-visible 'content'. This ensures we never leak partial function-call details back to the user.

Source code in src/api/sse_models.py 
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
@staticmethod
def make_text_chunk(text: str) -> 'SSEChunk':
    """
    Utility to create a minimal SSEChunk that only has user-visible 'content'.
    This ensures we never leak partial function-call details back to the user.
    """
    return SSEChunk(
        id=f"chatcmpl-{time.time()}",
        object="chat.completion.chunk",
        created=int(time.time()),
        model="agent-01",
        choices=[
            SSEChoice(
                index=0,
                delta=SSEDelta(role="assistant", content=text),
                finish_reason=None
            )
        ]
    )