kaboo_workflows.converters¶

Stream converters that reshape agent events for external consumers.

kaboo_workflows.converters ¶

Stream converters for transforming StreamEvents into response formats.

StreamConverter ¶

Bases: ABC

Converts StreamEvent objects into protocol-specific output chunks.

Each converter is stateful across one completion stream (tracks message id, created timestamp, tool call index, etc.). Create a new instance per request — do not share across concurrent requests.

convert `abstractmethod` ¶

convert(event)

Convert one StreamEvent into zero or more output chunks.

Returns a list (possibly empty) of serializable dicts. The transport layer is responsible for serializing and framing (e.g. 'data: {json}\n\n' for SSE). This method returns data shapes only — never pre-serialized strings.

Parameters:

Name	Type	Description	Default
`event`	`StreamEvent`	The StreamEvent to convert.	required

Returns:

Type	Description
`list[dict[str, Any]]`	A list of serializable dicts representing output chunks.

Source code in src/kaboo_workflows/converters/base.py

@abstractmethod
def convert(self, event: StreamEvent) -> list[dict[str, Any]]:
    """Convert one StreamEvent into zero or more output chunks.

    Returns a list (possibly empty) of serializable dicts.
    The transport layer is responsible for serializing and framing
    (e.g. 'data: {json}\\n\\n' for SSE). This method returns data
    shapes only — never pre-serialized strings.

    Args:
        event: The StreamEvent to convert.

    Returns:
        A list of serializable dicts representing output chunks.
    """
    ...

done_marker `abstractmethod` ¶

done_marker()

Terminal sentinel to emit after the stream ends.

E.g. 'data: [DONE]\n\n' for OpenAI SSE. Return empty string if the protocol needs no terminator.

Returns:

Type	Description
`str`	The terminal string to emit, or empty string if none.

Source code in src/kaboo_workflows/converters/base.py

@abstractmethod
def done_marker(self) -> str:
    """Terminal sentinel to emit after the stream ends.

    E.g. 'data: [DONE]\\n\\n' for OpenAI SSE.
    Return empty string if the protocol needs no terminator.

    Returns:
        The terminal string to emit, or empty string if none.
    """
    ...

content_type ¶

content_type()

MIME type for the HTTP streaming response.

Returns:

Type	Description
`str`	The MIME type string.

Source code in src/kaboo_workflows/converters/base.py

def content_type(self) -> str:
    """MIME type for the HTTP streaming response.

    Returns:
        The MIME type string.
    """
    return "text/event-stream"

OpenAIStreamConverter ¶

OpenAIStreamConverter(
    *,
    entry_agent_name,
    model_label=None,
    completion_id=None,
    reasoning_field_mode="both",
    tool_result_render="details_block",
    emit_usage_chunk=False,
    verbosity="compact",
)

Bases: StreamConverter

Stateful converter from :class:~kaboo_workflows.wire.StreamEvent to OpenAI chat.completion.chunk dicts.

Targets Open WebUI and LibreChat. Translates strands events into the OpenAI Chat Completions streaming protocol (v1) with reasoning extensions from DeepSeek (reasoning_content) and OpenRouter (reasoning).

Single-use per stream. Call :meth:reset to reuse across requests.

Initialize the OpenAIStreamConverter.

Parameters:

Name	Type	Description	Default
`entry_agent_name`	`str`	Name of the agent that owns the user-visible turn. TOKEN and REASONING from all other agents are suppressed. NODE_START/NODE_STOP from this agent surface sub-agents as `<details>` blocks alongside tool calls.	required
`model_label`	`str \| None`	Value for every chunk's `model` field. Defaults to `entry_agent_name`. Pass the `model` from the incoming request body so the response echoes it faithfully.	`None`
`completion_id`	`str \| None`	Fixed id for the whole stream. Defaults to a fresh `chatcmpl-<hex24>` on each instantiation.	`None`
`reasoning_field_mode`	`Literal['both', 'deepseek', 'openrouter', 'none']`	Which `delta` field(s) carry reasoning. `"both"` (default): `reasoning_content` (Open WebUI / DeepSeek) and `reasoning` (LibreChat / OpenRouter). `"deepseek"`: only `reasoning_content`. `"openrouter"`: only `reasoning`. `"none"`: reasoning dropped from the stream.	`'both'`
`tool_result_render`	`Literal['details_block', 'none']`	How tool calls and results are surfaced. `"details_block"` (default): a single completed `<details type="tool_calls" done="true">` block emitted on TOOL_END / NODE_STOP, carrying the call inputs as HTML attributes and the result as the body. No native `delta.tool_calls` chunks are produced because strands has already executed the tool by the time the stream emits it; emitting native deltas without a closing `finish_reason: "tool_calls"` only confuses clients into looping on the response. `"none"`: tool calls and results are not surfaced at all.	`'details_block'`
`emit_usage_chunk`	`bool`	When `True`, deliver usage as a separate trailing chunk with `choices: []` matching `stream_options: {include_usage: true}`. When `False` (default), usage is stapled to the `finish_reason` chunk.	`False`
`verbosity`	`Literal['compact', 'narrate']`	`"compact"` (default) streams only entry-agent tokens. `"narrate"` also streams sub-agent tokens inside the active node's `<details>` body.	`'compact'`

Source code in src/kaboo_workflows/converters/openai.py

def __init__(
    self,
    *,
    entry_agent_name: str,
    model_label: str | None = None,
    completion_id: str | None = None,
    reasoning_field_mode: Literal["both", "deepseek", "openrouter", "none"] = "both",
    tool_result_render: Literal["details_block", "none"] = "details_block",
    emit_usage_chunk: bool = False,
    verbosity: Literal["compact", "narrate"] = "compact",
) -> None:
    """Initialize the OpenAIStreamConverter.

    Args:
        entry_agent_name: Name of the agent that owns the user-visible turn.
            TOKEN and REASONING from all other agents are suppressed.
            NODE_START/NODE_STOP from this agent surface sub-agents as
            ``<details>`` blocks alongside tool calls.
        model_label: Value for every chunk's ``model`` field.  Defaults to
            ``entry_agent_name``.  Pass the ``model`` from the incoming
            request body so the response echoes it faithfully.
        completion_id: Fixed id for the whole stream.  Defaults to a fresh
            ``chatcmpl-<hex24>`` on each instantiation.
        reasoning_field_mode: Which ``delta`` field(s) carry reasoning.

            - ``"both"`` (default): ``reasoning_content`` (Open WebUI /
              DeepSeek) **and** ``reasoning`` (LibreChat / OpenRouter).
            - ``"deepseek"``: only ``reasoning_content``.
            - ``"openrouter"``: only ``reasoning``.
            - ``"none"``: reasoning dropped from the stream.
        tool_result_render: How tool calls and results are surfaced.

            - ``"details_block"`` (default): a single completed
              ``<details type="tool_calls" done="true">`` block emitted on
              TOOL_END / NODE_STOP, carrying the call inputs as HTML
              attributes and the result as the body. No native
              ``delta.tool_calls`` chunks are produced because strands has
              already executed the tool by the time the stream emits it;
              emitting native deltas without a closing
              ``finish_reason: "tool_calls"`` only confuses clients into
              looping on the response.
            - ``"none"``: tool calls and results are not surfaced at all.
        emit_usage_chunk: When ``True``, deliver usage as a separate trailing
            chunk with ``choices: []`` matching ``stream_options: {include_usage: true}``.
            When ``False`` (default), usage is stapled to the ``finish_reason`` chunk.
        verbosity: ``"compact"`` (default) streams only entry-agent tokens.
            ``"narrate"`` also streams sub-agent tokens inside the active
            node's ``<details>`` body.
    """
    self._entry_agent_name = entry_agent_name
    self._model_label: str = model_label if model_label is not None else entry_agent_name
    self._completion_id = completion_id or f"chatcmpl-{uuid.uuid4().hex[:24]}"
    self._created = int(_time.time())
    self._reasoning_field_mode = reasoning_field_mode
    self._tool_result_render = tool_result_render
    self._emit_usage_chunk = emit_usage_chunk
    self._verbosity = verbosity

    # Mutable stream state — reset on each call to reset()
    self._sent_role = False
    self._open_tool_calls: dict[str, _ToolCallFrame] = {}
    self._open_node_frames: dict[str, _ToolCallFrame] = {}
    self._reasoning_tokens: int = 0

convert ¶

convert(event)

Convert one :class:~kaboo_workflows.wire.StreamEvent into OpenAI chunk(s).

Parameters:

Name	Type	Description	Default
`event`	`StreamEvent`	The event to convert.	required

Returns:

Type	Description
`list[dict[str, Any]]`	A list of `chat.completion.chunk` dicts (possibly empty).
`list[dict[str, Any]]`	The transport layer is responsible for `data: {json}\n\n` framing.

Source code in src/kaboo_workflows/converters/openai.py

def convert(self, event: StreamEvent) -> list[dict[str, Any]]:
    """Convert one :class:`~kaboo_workflows.wire.StreamEvent` into OpenAI chunk(s).

    Args:
        event: The event to convert.

    Returns:
        A list of ``chat.completion.chunk`` dicts (possibly empty).
        The transport layer is responsible for ``data: {json}\\n\\n`` framing.
    """
    is_entry = event.agent_name == self._entry_agent_name

    dispatch: dict[str, Any] = {
        EventType.TOKEN: self._handle_token,
        EventType.REASONING: self._handle_reasoning,
        EventType.TOOL_START: self._handle_tool_start,
        EventType.TOOL_END: self._handle_tool_end,
        EventType.AGENT_COMPLETE: self._handle_agent_complete,
        EventType.MULTIAGENT_COMPLETE: self._handle_multiagent_complete,
        EventType.ERROR: self._handle_error,
        EventType.NODE_START: self._handle_node_start,
        EventType.NODE_STOP: self._handle_node_stop,
    }

    handler = dispatch.get(event.type)
    if handler is not None:
        return handler(event, is_entry)
    return []

done_marker ¶

done_marker()

Return the OpenAI SSE stream terminator.

Returns:

Type	Description
`str`	The `data: [DONE]\n\n` sentinel string.

Source code in src/kaboo_workflows/converters/openai.py

def done_marker(self) -> str:
    """Return the OpenAI SSE stream terminator.

    Returns:
        The ``data: [DONE]\\n\\n`` sentinel string.
    """
    return "data: [DONE]\n\n"

reset ¶

reset()

Reset all mutable stream state for reuse across requests.

Preserves constructor configuration and regenerates completion_id and created for the new stream.

Source code in src/kaboo_workflows/converters/openai.py

def reset(self) -> None:
    """Reset all mutable stream state for reuse across requests.

    Preserves constructor configuration and regenerates ``completion_id``
    and ``created`` for the new stream.
    """
    self._completion_id = f"chatcmpl-{uuid.uuid4().hex[:24]}"
    self._created = int(_time.time())
    self._sent_role = False
    self._open_tool_calls = {}
    self._open_node_frames = {}
    self._reasoning_tokens = 0

RawStreamConverter ¶

Bases: StreamConverter

Converts StreamEvents to raw JSON dicts (newline-delimited).

convert ¶

convert(event)

Pass through as dict.

Parameters:

Name	Type	Description	Default
`event`	`StreamEvent`	The StreamEvent to convert.	required

Returns:

Type	Description
`list[dict[str, Any]]`	A single-element list containing the event's dict representation.

Source code in src/kaboo_workflows/converters/raw.py

def convert(self, event: StreamEvent) -> list[dict[str, Any]]:
    """Pass through as dict.

    Args:
        event: The StreamEvent to convert.

    Returns:
        A single-element list containing the event's dict representation.
    """
    return [event.asdict()]

done_marker ¶

done_marker()

No terminator needed for raw streams.

Returns:

Type	Description
`str`	An empty string.

Source code in src/kaboo_workflows/converters/raw.py

def done_marker(self) -> str:
    """No terminator needed for raw streams.

    Returns:
        An empty string.
    """
    return ""

kaboo_workflows.converters¶