kaboo_workflows.testing

Scripted models and an in-process pipeline for testing your own workflows. See Testing your workflows for the guide.

kaboo_workflows.testing

Test your own workflows end to end, with no model provider or network.

:class:ScriptedModel stands in for an LLM with a fixed per-agent script, and :class:Pipeline drives a config through the same AG-UI path a deployed service uses, returning what a client would see as a :class:TurnResult.

Pipeline

Pipeline(config, *, session_config_key=None)

A resolved config that can be driven turn by turn in one event loop.

Resolve config once for every turn this pipeline drives.

By default one session serves every turn, as a process with a fixed config does. Pass session_config_key and each turn builds its own session from the config it submits in forwarded_props, exactly as the endpoint does — which is also how a test simulates a restart, since a rebuilt session shares nothing with the last one.

Parameters:

Name	Type	Description	Default
config	ConfigInput | list[ConfigInput]	A file path, a raw YAML string, or a list of either, as :func:~kaboo_workflows.config.parse_config_sources accepts. With session_config_key set this is the base config each submitted config layers over.	required
session_config_key	str | None	The forwarded_props key each turn's own config arrives under. None serves config alone.	None

Source code in src/kaboo_workflows/testing/pipeline.py

def __init__(
    self,
    config: ConfigInput | list[ConfigInput],
    *,
    session_config_key: str | None = None,
) -> None:
    """Resolve ``config`` once for every turn this pipeline drives.

    By default one session serves every turn, as a process with a fixed
    config does. Pass ``session_config_key`` and each turn builds its own
    session from the config it submits in ``forwarded_props``, exactly as
    the endpoint does — which is also how a test simulates a restart,
    since a rebuilt session shares nothing with the last one.

    Args:
        config: A file path, a raw YAML string, or a list of either, as
            :func:`~kaboo_workflows.config.parse_config_sources` accepts.
            With ``session_config_key`` set this is the base config each
            submitted config layers over.
        session_config_key: The ``forwarded_props`` key each turn's own
            config arrives under. ``None`` serves ``config`` alone.
    """
    self.base_raw = parse_config_sources(config)
    self.app_config = validate_raw_config(self.base_raw)
    self.infra = resolve_infra(self.app_config)
    self.infra.mcp_lifecycle.start(pin_clients=session_config_key is None)
    self._events: dict[str, list[Any]] = {}
    self._usage_by_run: dict[str, dict[str, int]] = {}
    self._resolver = (
        None
        if session_config_key is None
        else _make_session_resolver(
            self.base_raw, self.infra, session_config_key=session_config_key
        )
    )
    self._conversations = _ConversationSessions(self.app_config, self.infra)

turn async

turn(
    message=None,
    *,
    thread_id="t1",
    run_id="r1",
    state=None,
    messages=None,
    resume=None,
    forwarded_props=None,
    tools=None,
    context=None,
)

Drive one run and collect what a client would see.

Parameters:

Name	Type	Description	Default
message	str | None	The user's message. Ignored when messages is given.	None
thread_id	str	The conversation id.	't1'
run_id	str	This run's id.	'r1'
state	dict[str, Any] | None	The client state to send, usually the previous turn's :meth:TurnResult.state, which carries history and pending interrupts.	None
messages	list[Any] | None	A full AG-UI transcript to send instead of message.	None
resume	list[dict[str, Any]] | None	Answers to the pending interrupts, as a client sends them to resume a paused run.	None
forwarded_props	dict[str, Any] | None	The run's forwarded props, such as a run token or, with session_config_key, the run's own config.	None
tools	list[Any] | None	The browser tools the client declares, as AG-UI Tools.	None
context	list[Any] | None	The client's context items, as AG-UI Contexts.	None

Returns:

Type	Description
TurnResult	The events and activity groups the run produced.

Raises:

Type	Description
LookupError	resume was given but nothing is paused on this thread and the state carries no gate to restore.

Source code in src/kaboo_workflows/testing/pipeline.py

async def turn(
    self,
    message: str | None = None,
    *,
    thread_id: str = "t1",
    run_id: str = "r1",
    state: dict[str, Any] | None = None,
    messages: list[Any] | None = None,
    resume: list[dict[str, Any]] | None = None,
    forwarded_props: dict[str, Any] | None = None,
    tools: list[Any] | None = None,
    context: list[Any] | None = None,
) -> TurnResult:
    """Drive one run and collect what a client would see.

    Args:
        message: The user's message. Ignored when ``messages`` is given.
        thread_id: The conversation id.
        run_id: This run's id.
        state: The client state to send, usually the previous turn's
            :meth:`TurnResult.state`, which carries history and pending
            interrupts.
        messages: A full AG-UI transcript to send instead of ``message``.
        resume: Answers to the pending interrupts, as a client sends them
            to resume a paused run.
        forwarded_props: The run's forwarded props, such as a run token or,
            with ``session_config_key``, the run's own config.
        tools: The browser tools the client declares, as AG-UI ``Tool``s.
        context: The client's context items, as AG-UI ``Context``s.

    Returns:
        The events and activity groups the run produced.

    Raises:
        LookupError: ``resume`` was given but nothing is paused on this
            thread and the state carries no gate to restore.
    """
    set_activity_context(thread_id, run_id)
    history, paused = inbound_sessions(state)
    exchange = HistoryExchange(inbound=history)
    set_history_exchange(exchange)

    if messages is None:
        messages = [UserMessage(id="m1", role="user", content=message or "")]
    input_data = RunAgentInput(
        thread_id=thread_id,
        run_id=run_id,
        state=state or {},
        messages=messages,
        tools=tools or [],
        context=context or [],
        forwarded_props=forwarded_props or {},
        resume=[ResumeEntry.model_validate(entry) for entry in resume] if resume else None,
    )
    sent = list(input_data.messages)
    take_a2ui_action(input_data)
    carried = split_subagent_messages(input_data)
    browser_answers = take_browser_answers(input_data, carried)
    set_session_exchange(
        SessionExchange(inbound=paused, resuming=bool(resume or browser_answers))
    )
    set_forwarded_props(forwarded_props or {})
    set_references(_parse_references(input_data))

    session = self._resolve_session(input_data)
    agui = session.agui_agent
    browser_tools = prepare_run(input_data, session)
    merged: asyncio.Queue[Any] = asyncio.Queue()
    run_events = RunEvents(
        merged, carried, subagents=not session.orchestrated, browser_tools=browser_tools
    )
    session.event_queue.flush()
    session.event_queue.forward_to(run_events.sink)
    open_state_channel(input_data, session, run_events)

    try:
        if (resume or browser_answers) and not session.orchestrated:
            # Mirror kaboo_endpoint: ready the paused agent, then hand the
            # resume to ag-ui-strands as it came.
            if not await _prepare_resume(agui, input_data, browser_answers):
                # The endpoint answers RUN_ERROR / RESUME_NO_SESSION here;
                # in-process the equivalent is refusing to drive the run.
                raise LookupError("no resumable agent state for thread")
        await equip_entry_agent(input_data, session)
        await _consume_run(agui, input_data, merged, exchange)
    finally:
        session.event_queue.forward_to(None)
        if session.transient:
            await session.aclose()

    events: list[Any] = []
    while not merged.empty():
        item = merged.get_nowait()
        if item is not _DONE:
            events.extend(run_events.emit(item))
    conversation = self._events.setdefault(thread_id, [])
    conversation.extend(events)
    for finished in (e for e in events if e.type == AGUIEventType.RUN_FINISHED):
        totals: dict[str, int] = {}
        for entry in getattr(finished, "usage", None) or []:
            for key, value in entry.model_dump(by_alias=True, exclude_none=True).items():
                if isinstance(value, int):
                    totals[key] = totals.get(key, 0) + value
        self._usage_by_run[finished.run_id] = totals
    return TurnResult(
        events=events,
        groups=fold_subagents(conversation),
        exchange=exchange,
        usage_by_run=dict(self._usage_by_run),
        sent=sent,
    )

forget

forget(thread_id)

Drop a conversation's session and events, as a process that never saw it.

Parameters:

Name	Type	Description	Default
thread_id	str	The conversation id.	required

Source code in src/kaboo_workflows/testing/pipeline.py

def forget(self, thread_id: str) -> None:
    """Drop a conversation's session and events, as a process that never saw it.

    Args:
        thread_id: The conversation id.
    """
    self._conversations.forget(thread_id)
    self._events.pop(thread_id, None)

close

close()

Stop the MCP servers and clients the config started.

Source code in src/kaboo_workflows/testing/pipeline.py

def close(self) -> None:
    """Stop the MCP servers and clients the config started."""
    self.infra.mcp_lifecycle.stop()

TurnResult dataclass

TurnResult(
    events,
    groups,
    exchange=HistoryExchange(),
    usage_by_run=dict(),
    sent=list(),
)

Outcome of one run: the AG-UI events and the sub-agent groups.

Attributes:

Name	Type	Description
events	list[Any]	Every AG-UI event the run streamed, in order.
groups	dict[str, dict[str, Any]]	One group per sub-agent invocation of the thread so far, keyed by its subagentRunId, as a client folds them (see :func:~kaboo_workflows.adapters._subagents.fold_subagents).
exchange	HistoryExchange	The history carried in and out of the run.
usage_by_run	dict[str, dict[str, int]]	The run's token totals per run id, from RUN_FINISHED.
sent	list[Any]	The messages the client sent with the run.

types

types()

Return the AG-UI event types, in order.

Returns:

Type	Description
list[str]	Each event's type as a string, such as "RUN_STARTED".

Source code in src/kaboo_workflows/testing/pipeline.py

def types(self) -> list[str]:
    """Return the AG-UI event types, in order.

    Returns:
        Each event's type as a string, such as ``"RUN_STARTED"``.
    """
    return [getattr(e.type, "value", str(e.type)) for e in self.events]

of_type

of_type(agui_type)

Return the events of one AG-UI type.

Parameters:

Name	Type	Description	Default
agui_type	EventType	The ag_ui.core.EventType to keep.	required

Returns:

Type	Description
list[Any]	The matching events, in order.

Source code in src/kaboo_workflows/testing/pipeline.py

def of_type(self, agui_type: AGUIEventType) -> list[Any]:
    """Return the events of one AG-UI type.

    Args:
        agui_type: The ``ag_ui.core.EventType`` to keep.

    Returns:
        The matching events, in order.
    """
    return [e for e in self.events if e.type == agui_type]

text

text()

Return the chat reply the run streamed.

Returns:

Type	Description
str	Every TEXT_MESSAGE_CONTENT delta of the entry agent,
str	concatenated; sub-agents' text belongs to their cards.

Source code in src/kaboo_workflows/testing/pipeline.py

def text(self) -> str:
    """Return the chat reply the run streamed.

    Returns:
        Every ``TEXT_MESSAGE_CONTENT`` delta of the entry agent,
        concatenated; sub-agents' text belongs to their cards.
    """
    return "".join(
        getattr(e, "delta", "")
        for e in self.events
        if e.type == AGUIEventType.TEXT_MESSAGE_CONTENT
        and not getattr(e, "subagent_run_id", None)
    )

finished

finished()

Return whether the run reached RUN_FINISHED.

Returns:

Type	Description
bool	True when a RUN_FINISHED event was streamed.

Source code in src/kaboo_workflows/testing/pipeline.py

def finished(self) -> bool:
    """Return whether the run reached ``RUN_FINISHED``.

    Returns:
        True when a ``RUN_FINISHED`` event was streamed.
    """
    return any(e.type == AGUIEventType.RUN_FINISHED for e in self.events)

errored

errored()

Return whether the run ended in RUN_ERROR.

Returns:

Type	Description
bool	True when a RUN_ERROR event was streamed.

Source code in src/kaboo_workflows/testing/pipeline.py

def errored(self) -> bool:
    """Return whether the run ended in ``RUN_ERROR``.

    Returns:
        True when a ``RUN_ERROR`` event was streamed.
    """
    return any(e.type == AGUIEventType.RUN_ERROR for e in self.events)

run_outcome

run_outcome()

Return the outcome carried by RUN_FINISHED.

Returns:

Type	Description
Any	The outcome, which holds the pending interrupts when the run
Any	paused, or None when the run did not finish.

Source code in src/kaboo_workflows/testing/pipeline.py

def run_outcome(self) -> Any:
    """Return the outcome carried by ``RUN_FINISHED``.

    Returns:
        The outcome, which holds the pending interrupts when the run
        paused, or ``None`` when the run did not finish.
    """
    for e in self.of_type(AGUIEventType.RUN_FINISHED):
        return getattr(e, "outcome", None)
    return None

messages

messages()

Return the transcript a client holds after the run.

The messages it sent, followed by the ones the run's text, tool-call and tool-result events build, sub-agents' included. Pass it back as messages= on the next turn, the way a client does.

Returns:

Type	Description
list[Any]	The messages, in order.

Source code in src/kaboo_workflows/testing/pipeline.py

def messages(self) -> list[Any]:
    """Return the transcript a client holds after the run.

    The messages it sent, followed by the ones the run's text, tool-call
    and tool-result events build, sub-agents' included. Pass it back as
    ``messages=`` on the next turn, the way a client does.

    Returns:
        The messages, in order.
    """
    messages = [message.model_copy(deep=True) for message in self.sent]
    by_id: dict[str, Any] = {message.id: message for message in messages}
    calls: dict[str, ToolCall] = {}

    def assistant(message_id: str, subagent_run_id: str | None) -> Any:
        if message_id not in by_id:
            by_id[message_id] = AssistantMessage(
                id=message_id, role="assistant", subagent_run_id=subagent_run_id
            )
            messages.append(by_id[message_id])
        return by_id[message_id]

    for event in self.events:
        if event.type == AGUIEventType.TEXT_MESSAGE_START:
            assistant(event.message_id, event.subagent_run_id)
        elif event.type == AGUIEventType.TEXT_MESSAGE_CONTENT:
            message = by_id[event.message_id]
            message.content = (message.content or "") + event.delta
        elif event.type == AGUIEventType.TOOL_CALL_START:
            message = assistant(
                event.parent_message_id or event.tool_call_id, event.subagent_run_id
            )
            call = ToolCall(
                id=event.tool_call_id,
                type="function",
                function=FunctionCall(name=event.tool_call_name, arguments=""),
            )
            message.tool_calls = [*(message.tool_calls or []), call]
            calls[event.tool_call_id] = call
        elif event.type == AGUIEventType.TOOL_CALL_ARGS and event.tool_call_id in calls:
            calls[event.tool_call_id].function.arguments += event.delta
        elif event.type == AGUIEventType.TOOL_CALL_RESULT and event.message_id not in by_id:
            by_id[event.message_id] = ToolMessage(
                id=event.message_id,
                role="tool",
                content=event.content,
                tool_call_id=event.tool_call_id,
                subagent_run_id=event.subagent_run_id,
            )
            messages.append(by_id[event.message_id])
    return messages

state

state()

Return the state the run left, as a host would persist it.

The last STATE_SNAPSHOT with every STATE_DELTA after it applied. Pass it back as state= on the next turn to carry history and pending interrupts forward, the way a client does.

Returns:

Type	Description
dict[str, Any]	The state, or an empty dict when none was streamed.

Source code in src/kaboo_workflows/testing/pipeline.py

def state(self) -> dict[str, Any]:
    """Return the state the run left, as a host would persist it.

    The last ``STATE_SNAPSHOT`` with every ``STATE_DELTA`` after it applied.
    Pass it back as ``state=`` on the next turn to carry history and pending
    interrupts forward, the way a client does.

    Returns:
        The state, or an empty dict when none was streamed.
    """
    state: dict[str, Any] = {}
    for event in self.events:
        if event.type == AGUIEventType.STATE_SNAPSHOT:
            snapshot = getattr(event, "snapshot", None)
            state = copy.deepcopy(snapshot) if isinstance(snapshot, dict) else {}
        elif event.type == AGUIEventType.STATE_DELTA:
            delta = event.model_dump(by_alias=True, exclude_none=True)["delta"]
            state = jsonpatch.apply_patch(state, delta)
    return state

group_ids

group_ids()

Return the sub-agent group ids.

Returns:

Type	Description
list[str]	The ids, sorted.

Source code in src/kaboo_workflows/testing/pipeline.py

def group_ids(self) -> list[str]:
    """Return the sub-agent group ids.

    Returns:
        The ids, sorted.
    """
    return sorted(self.groups.keys())

agent_names

agent_names()

Return the names of the sub-agents that ran.

Returns:

Type	Description
set[str]	The agent names across all groups.

Source code in src/kaboo_workflows/testing/pipeline.py

def agent_names(self) -> set[str]:
    """Return the names of the sub-agents that ran.

    Returns:
        The agent names across all groups.
    """
    return {str(g["agentName"]) for g in self.groups.values() if g.get("agentName") is not None}

group

group(group_id)

Return one sub-agent group.

Parameters:

Name	Type	Description	Default
group_id	str	The group's id.	required

Returns:

Type	Description
dict[str, Any]	The group as the UI would hold it.

Raises:

Type	Description
KeyError	No group has that id.

Source code in src/kaboo_workflows/testing/pipeline.py

def group(self, group_id: str) -> dict[str, Any]:
    """Return one sub-agent group.

    Args:
        group_id: The group's id.

    Returns:
        The group as the UI would hold it.

    Raises:
        KeyError: No group has that id.
    """
    return self.groups[group_id]

children_of

children_of(parent)

Return the groups nested directly under parent.

Parameters:

Name	Type	Description	Default
parent	str | None	A group id, or None for the top-level groups.	required

Returns:

Type	Description
list[str]	The child group ids, sorted.

Source code in src/kaboo_workflows/testing/pipeline.py

def children_of(self, parent: str | None) -> list[str]:
    """Return the groups nested directly under ``parent``.

    Args:
        parent: A group id, or ``None`` for the top-level groups.

    Returns:
        The child group ids, sorted.
    """
    return sorted(gid for gid, g in self.groups.items() if g.get("parentGroup") == parent)

ScriptedModel

ScriptedModel(
    model_id="scripted",
    *,
    script=None,
    turns=None,
    final_text="Done.",
    stream_chunk=0,
    stream_delay_ms=0,
    **_,
)

Bases: Model

A strands model that replays a fixed script instead of calling a provider.

Initialize the scripted model.

Parameters:

Name	Type	Description	Default
model_id	str	Reported as the model id; any value works.	'scripted'
script	list[dict[str, Any]] | None	The steps to replay, one per model turn. See the module docstring for the step shapes.	None
turns	list[list[dict[str, Any]]] | None	One script per human turn; the last repeats. Overrides script when given.	None
final_text	str	The reply once the script is exhausted.	'Done.'
stream_chunk	int	Characters per streamed delta; 0 sends each text and each tool's arguments as one delta.	0
stream_delay_ms	int	Pause between deltas, for watching a UI render.	0
**_	Any	Extra model params from the config, accepted and ignored so a config written for a real provider still loads.	{}

Source code in src/kaboo_workflows/testing/scripted_model.py

def __init__(
    self,
    model_id: str = "scripted",
    *,
    script: list[dict[str, Any]] | None = None,
    turns: list[list[dict[str, Any]]] | None = None,
    final_text: str = "Done.",
    stream_chunk: int = 0,
    stream_delay_ms: int = 0,
    **_: Any,
) -> None:
    """Initialize the scripted model.

    Args:
        model_id: Reported as the model id; any value works.
        script: The steps to replay, one per model turn. See the module
            docstring for the step shapes.
        turns: One script per human turn; the last repeats. Overrides
            ``script`` when given.
        final_text: The reply once the script is exhausted.
        stream_chunk: Characters per streamed delta; 0 sends each text and
            each tool's arguments as one delta.
        stream_delay_ms: Pause between deltas, for watching a UI render.
        **_: Extra model params from the config, accepted and ignored so a
            config written for a real provider still loads.
    """
    self._config: dict[str, Any] = {"model_id": model_id}
    self._script = script or []
    self._turns = [list(turn) for turn in turns] if turns else None
    self._final_text = final_text
    self._stream_chunk = max(0, int(stream_chunk))
    self._stream_delay = max(0, int(stream_delay_ms)) / 1000
    self._step = 0
    self.calls: list[list[dict[str, Any]]] = []
    self.system_prompts: list[Any] = []

update_config

update_config(**model_config)

Merge model_config into the model's config.

Parameters:

Name	Type	Description	Default
**model_config	Any	Config keys to set.	{}

Source code in src/kaboo_workflows/testing/scripted_model.py

def update_config(self, **model_config: Any) -> None:
    """Merge ``model_config`` into the model's config.

    Args:
        **model_config: Config keys to set.
    """
    self._config.update(model_config)

get_config

get_config()

Return the model's config.

Returns:

Type	Description
dict[str, Any]	The config dict, including model_id.

Source code in src/kaboo_workflows/testing/scripted_model.py

def get_config(self) -> dict[str, Any]:
    """Return the model's config.

    Returns:
        The config dict, including ``model_id``.
    """
    return self._config

structured_output async

structured_output(output_model, prompt=None, **kwargs)

Yield a default instance of output_model.

Parameters:

Name	Type	Description	Default
output_model	Any	The Pydantic model to instantiate.	required
prompt	Any	Ignored.	None
**kwargs	Any	Ignored.	{}

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	One {"output": instance} event.

Source code in src/kaboo_workflows/testing/scripted_model.py

async def structured_output(  # ty: ignore[invalid-method-override]
    self, output_model: Any, prompt: Any = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Yield a default instance of ``output_model``.

    Args:
        output_model: The Pydantic model to instantiate.
        prompt: Ignored.
        **kwargs: Ignored.

    Yields:
        One ``{"output": instance}`` event.
    """
    yield {"output": output_model()}

stream async

stream(
    messages, tool_specs=None, system_prompt=None, **kwargs
)

Replay the next script step as a strands model event stream.

Parameters:

Name	Type	Description	Default
messages	Any	The conversation strands sends; recorded on calls.	required
tool_specs	Any	The tools available this turn, used to resolve tool names and fill tool inputs.	None
system_prompt	Any	Recorded on system_prompts.	None
**kwargs	Any	Ignored.	{}

Yields:

Type	Description
AsyncIterator[dict[str, Any]]	strands model stream events for one assistant message.

Raises:

Type	Description
RuntimeError	The step is an error step, or the conversation breaks Bedrock's tool-use pairing rule.
ValueError	A tool step names a tool that is not available.

Source code in src/kaboo_workflows/testing/scripted_model.py

async def stream(  # ty: ignore[invalid-method-override]
    self, messages: Any, tool_specs: Any = None, system_prompt: Any = None, **kwargs: Any
) -> AsyncIterator[dict[str, Any]]:
    """Replay the next script step as a strands model event stream.

    Args:
        messages: The conversation strands sends; recorded on ``calls``.
        tool_specs: The tools available this turn, used to resolve tool
            names and fill tool inputs.
        system_prompt: Recorded on ``system_prompts``.
        **kwargs: Ignored.

    Yields:
        strands model stream events for one assistant message.

    Raises:
        RuntimeError: The step is an ``error`` step, or the conversation
            breaks Bedrock's tool-use pairing rule.
        ValueError: A tool step names a tool that is not available.
    """
    self.calls.append(copy.deepcopy(messages))
    self.system_prompts.append(system_prompt)
    _validate_tool_pairing(messages)

    script = self._script_for(messages)
    step = _steps_since_human_turn(messages)
    action = script[step] if step < len(script) else {"text": self._final_text}
    self._step = step + 1

    if "error" in action:
        raise RuntimeError(action["error"])

    yield {"messageStart": {"role": "assistant"}}

    if "text" in action:
        yield {"contentBlockStart": {"start": {}}}
        for piece in _pieces(action["text"], self._stream_chunk):
            yield {"contentBlockDelta": {"delta": {"text": piece}}}
            await self._pause()
        yield {"contentBlockStop": {}}
        yield {"messageStop": {"stopReason": "end_turn"}}
    elif "tools" in action:
        for sub in action["tools"]:
            spec = _resolve_tool_spec(sub, tool_specs or [])
            for event in _tool_use_events(
                spec, _build_tool_input(sub, spec), self._stream_chunk
            ):
                yield event
                await self._pause()
        yield {"messageStop": {"stopReason": "tool_use"}}
    else:
        spec = _resolve_tool_spec(action, tool_specs or [])
        for event in _tool_use_events(
            spec, _build_tool_input(action, spec), self._stream_chunk
        ):
            yield event
            await self._pause()
        yield {"messageStop": {"stopReason": "tool_use"}}

    yield {
        "metadata": {
            "usage": {"inputTokens": 1, "outputTokens": 1, "totalTokens": 2},
            "metrics": {"latencyMs": 1},
        }
    }

run_pipeline async

run_pipeline(
    config,
    message,
    *,
    thread_id="t1",
    run_id="r1",
    state=None,
)

Build a fresh pipeline, drive one turn, and close it.

Parameters:

Name	Type	Description	Default
config	ConfigInput | list[ConfigInput]	A file path, a raw YAML string, or a list of either.	required
message	str	The user's message.	required
thread_id	str	The conversation id.	't1'
run_id	str	The run's id.	'r1'
state	dict[str, Any] | None	The client state to send.	None

Returns:

Type	Description
TurnResult	The events and activity groups the run produced.

Source code in src/kaboo_workflows/testing/pipeline.py

async def run_pipeline(
    config: ConfigInput | list[ConfigInput],
    message: str,
    *,
    thread_id: str = "t1",
    run_id: str = "r1",
    state: dict[str, Any] | None = None,
) -> TurnResult:
    """Build a fresh pipeline, drive one turn, and close it.

    Args:
        config: A file path, a raw YAML string, or a list of either.
        message: The user's message.
        thread_id: The conversation id.
        run_id: The run's id.
        state: The client state to send.

    Returns:
        The events and activity groups the run produced.
    """
    pipe = Pipeline(config)
    try:
        return await pipe.turn(message, thread_id=thread_id, run_id=run_id, state=state)
    finally:
        pipe.close()