Workflow API¶

The workflow engine lives in kavalai.workflow. It turns a YAML graph (or a WorkflowBuilder chain) into an executable state machine, runs it as a stream of WorkflowStreamEvent events, and records a serialisable state through an AgentService and a pluggable task-logger backend.

Engine and builder¶

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software distributed under the License is distributed on an “AS IS” BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific language governing permissions and limitations under the License.

kavalai.workflow.engine.make_prompt(prompt: str, input_data: dict) → str[source]¶: Combine a rendered prompt with resolved input data into a system message.

class kavalai.workflow.engine.WorkflowEngine(graph: WorkflowGraph, *, agent_service: AgentService | None = None, task_logger: TaskLogger | None = None, client_factory: Callable[[...], BaseLlmClient] | None = None, data_models: dict[str, type[BaseModel]] | None = None, max_node_visits: int = 1000)[source]¶

Bases: object

Executes a v2 WorkflowGraph as a DAG / state machine.

The engine walks the graph from the start node, following transitions and evaluating branch nodes, until it reaches an end node. Each node’s result is stored in the run context; per-node debug data flows to task_logger.

Parameters:¶

graph : WorkflowGraph¶: The parsed workflow definition.
agent_service : Optional[AgentService]¶: Persistence for agents/sessions/runs/chat history. None runs the workflow without any persistence (no chat memory across turns).
task_logger : Optional[TaskLogger]¶: Backend for per-node debug data and model statistics.
client_factory : Optional[ClientFactory]¶: Factory (model, parameters, stats_receiver) -> BaseLlmClient used to build LLM clients. Defaults to the provider factory; inject a fake for offline testing.
max_node_visits : int¶: Safety cap on total node executions to guard against infinite loops.

classmethod from_yaml(yaml_string: str, **kwargs) → WorkflowEngine[source]¶: Build an engine from a YAML workflow definition string.

classmethod from_yaml_path(yaml_path: str, **kwargs) → WorkflowEngine[source]¶: Build an engine from a YAML workflow definition file.

classmethod from_dict(data: dict, **kwargs) → WorkflowEngine[source]¶: Build an engine from a parsed workflow definition dict.

get_data_type(name: str | None)[source]¶

async run(input_data: dict, *, session_id: str | None = None, external_id: str | None = None) → WorkflowState[source]¶

Execute the workflow for input_data and return the final state.

Drains run_stream() — the single execution path.

async run_stream(input_data: dict, *, session_id: str | None = None, external_id: str | None = None, state: WorkflowState | None = None) → AsyncGenerator[WorkflowStreamEvent, None][source]¶

Execute the workflow, yielding WorkflowStreamEvent events.

Lifecycle events (workflow_started, node_started / node_completed, workflow_completed / workflow_failed) frame the run; nodes with streaming enabled contribute partial / complete / restart content events in between.

Closing the generator early (e.g. the SSE client disconnected) aborts the run; the abort is recorded on the run row, best-effort. On failure a workflow_failed event is yielded before the WorkflowException is raised to the caller.

Parameters:¶

input_data: dict¶: The workflow input.
session_id: str | None = None¶: Optional session to continue.
external_id: str | None = None¶: Optional caller-supplied session key.
state: WorkflowState | None = None¶: Optional WorkflowState instance populated in place, so blocking callers can read the final state after draining the stream.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

class kavalai.workflow.builder.WorkflowBuilder(name: str, *, description: str = '', version: str = '2.0', llm_model: str | None = None, llm_kwargs: dict[str, Any] | None = None)[source]¶

Bases: object

A small fluent builder for constructing a WorkflowGraph in code.

Every method returns self so calls can be chained, and build() validates and returns the graph (build_engine() returns a ready WorkflowEngine). For example:

graph = (
    WorkflowBuilder("Greeter", llm_model="openai/gpt-4o-mini")
    .data_type("input", {"user_message": str})
    .data_type("output", {"agent_response": str})
    .start("reply")
    .llm("reply", prompt="Greet the user.",
         inputs={"input": "input"}, output="output", next="end")
    .end()
    .build()
)

Declare a data type.

Pass fields for an object of named scalars ({"intent": str}), schema for a full JSON-schema fragment, or ref to alias another type.

data_model(name: str, model: type[BaseModel]) → WorkflowBuilder[source]¶

Declare a data type from a Pydantic model class.

The model is used directly at run time (its fields validate the matching node input/output and drive structured LLM output), so you get full Pydantic expressiveness — Literal enums, defaults, validators — without writing a JSON-schema fragment. Its JSON schema is also recorded on the graph for storage and inspection.

Build the engine with build_engine() (which forwards the models) or pass data_models= to WorkflowEngine yourself.

start(next: str, *, name: str = 'start') → WorkflowBuilder[source]¶

end(*, name: str = 'end', output: str = 'output') → WorkflowBuilder[source]¶

llm(name: str, *, prompt: str, output: str, next: str, inputs: dict[str, ArgumentInfo | str | dict] | None = None, use_history: bool = True, llm_model: str | None = None, llm_kwargs: dict[str, Any] | None = None, stream_output: bool = False, stream_delta: bool = False) → WorkflowBuilder[source]¶

agent(name: str, *, prompt: str, output: str, next: str, inputs: dict[str, ArgumentInfo | str | dict] | None = None, allowed_tools: list[str] | None = None, max_steps: int = 10, llm_model: str | None = None, llm_kwargs: dict[str, Any] | None = None, stream_output: bool = False, stream_delta: bool = False, stream_instructions: bool = False, stream_partials: bool = False) → WorkflowBuilder[source]¶

function(name: str, *, tool: str, output: str, next: str, inputs: dict[str, ArgumentInfo | str | dict] | None = None, method: str = 'get') → WorkflowBuilder[source]¶

if_(name: str, *, condition: str, then: str, else_: str) → WorkflowBuilder[source]¶

switch(name: str, *, expr: str, cases: dict[str, str] | None = None, default: str | None = None) → WorkflowBuilder[source]¶

python_function(name: str, path: str) → WorkflowBuilder[source]¶: Register a Python tool by import path (e.g. pkg.mod.func).

rest_server(server: RestServer | dict) → WorkflowBuilder[source]¶

mcp_server(server: McpServer | dict) → WorkflowBuilder[source]¶

template(name: str, value: str) → WorkflowBuilder[source]¶

build() → WorkflowGraph[source]¶: Validate and return the WorkflowGraph.

build_engine(**kwargs)[source]¶

Build the graph and wrap it in a ready-to-run WorkflowEngine.

Any Pydantic models registered via data_model() are forwarded to the engine. Keyword arguments are passed through (agent_service, task_logger, client_factory, data_models, max_node_visits).

Graph models¶

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Workflow data models: the shared building blocks (input wiring, server/tool declarations and the workflow exception) and the v2 workflow graph (nodes and WorkflowGraph).

exception kavalai.workflow.models.WorkflowException[source]¶

Bases: Exception

Base exception for errors building, validating or running a workflow.

Bases: BaseModel

Describes input arguments in workflow YAML files.

The ‘type’ field describes where the input argument should be retrieved from. ‘literal’ - use value as specified ‘context’ - retrieve from agent run context ‘history’ - retrieve from previous agent run contexts.

type : Literal['literal', 'context', 'history']¶

value : BaseModel | str | int | float | bool | None¶

name : str | None¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.RestServer(*, name: str, url: str | None = None, url_env: str | None = None, username_env: str | None = None, password_env: str | None = None)[source]¶

Bases: BaseModel

Defines a REST server.

We also support HTTP Basic Auth for REST server endpoints, which are defined via environment variables username_env and password_env.

Note that url_env can also be read from the env file.

name : str¶

url : str | None¶

url_env : str | None¶

username_env : str | None¶

password_env : str | None¶

check_url_configs() → RestServer[source]¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.McpServer(*, name: str, command: str | None = None, command_env: str | None = None, args: list[str] = [], env: dict[str, str] = {}, url: str | None = None, url_env: str | None = None)[source]¶

Bases: BaseModel

Defines an MCP server.

name : str¶

command : str | None¶

command_env : str | None¶

args : list[str]¶

env : dict[str, str]¶

url : str | None¶

url_env : str | None¶

check_configs() → McpServer[source]¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.PythonFunction(*, name: str, path: str)[source]¶

Bases: BaseModel

Declares a Python tool available to a workflow.

Variables:¶

name : str¶: Name the tool is registered and addressed under (python://<name>).
path : str¶: Import path to the @kavalai.pythontool decorated function, e.g. my_package.my_module.my_func.

name : str¶

path : str¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.TemplateModel(*, name: str, value: str)[source]¶

Bases: BaseModel

A named, reusable text template referenced within a workflow.

Variables:¶

name : str¶: Identifier the template is referenced by.
value : str¶: The template text (e.g. a prompt) to interpolate at run time.

name : str¶

value : str¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.BaseNode(*, name: str)[source]¶

Bases: BaseModel

Common fields shared by every node in a workflow graph.

A node is one vertex in the DAG/state-machine. name uniquely identifies the node and is the target referenced by transitions (next/then/ else/cases/default).

name : str¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.StartNode(*, name: str, type: 'start' = 'start', next: str)[source]¶

Bases: BaseNode

Interaction start node.

The caller hands an input to this node; execution begins here and proceeds to next.

type : Literal['start']¶

next : str¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.EndNode(*, name: str, type: 'end' = 'end', output: str = 'output')[source]¶

Bases: BaseNode

Interaction end node.

Reaching an end node terminates the interaction. output names the context variable whose value is returned to the caller.

type : Literal['end']¶

output : str¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.LLMNode(*, name: str, type: ~typing.Literal['llm'] = 'llm', prompt: str, inputs: dict[str, ~kavalai.workflow.models.ArgumentInfo] = {}, output: str, next: str, use_history: bool = True, llm_model: str | None = None, llm_kwargs: dict[str, ~typing.Any] = <factory>, stream_output: bool = False, stream_delta: bool = False)[source]¶

Bases: BaseNode

Single LLM completion node.

Resolves inputs from context, renders prompt and calls the LLM, storing the structured result in the output context variable, then transitions to next.

Streaming (see WorkflowStreamEvent for the event contract):

stream_output: Stream the completion as partial events named after this node while it is generated; auxiliary provider streams (e.g. Gemini thoughts) are streamed as <node>_<stream> (<node>_thought).
stream_delta: When True, each partial carries only the newly generated text and the client reassembles the value. When False (default), each partial carries the full accumulated, safe-parsed value so far — render-ready with no client-side assembly, at the cost of re-sending the whole buffer on every chunk (O(n^2) wire traffic over the stream; prefer stream_delta: true for long outputs).

type : Literal['llm']¶

prompt : str¶

inputs : dict[str, ArgumentInfo]¶

output : str¶

next : str¶

use_history : bool¶

llm_model : str | None¶

llm_kwargs : dict[str, Any]¶

stream_output : bool¶

stream_delta : bool¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.AgentNode(*, name: str, type: ~typing.Literal['agent'] = 'agent', prompt: str, inputs: dict[str, ~kavalai.workflow.models.ArgumentInfo] = {}, output: str, next: str, allowed_tools: list[str] = <factory>, max_steps: int = 10, llm_model: str | None = None, llm_kwargs: dict[str, ~typing.Any] = <factory>, stream_output: bool = False, stream_delta: bool = False, stream_instructions: bool = False, stream_partials: bool = False)[source]¶

Bases: BaseNode

Multi-step agent node.

Runs the v2 Agent loop (tool calling) up to max_steps and stores the final result in output.

Streaming (see WorkflowStreamEvent for the event contract):

stream_output: Stream the step’s output field as partial events named after this node while the model writes it. The streamed value is always the full safe-parsed JSON of the output so far (partial objects are not delta-able); the final complete event is authoritative — a provisional output produced alongside tool calls may be superseded.
stream_instructions: Stream each step’s instructions as <node>_instructions partials, completed once per step — an “ideating…” status line the UI replaces each step.
stream_partials: Stream each step’s raw model output as <node>_step<N> — a debug firehose including tool-call JSON.
stream_delta: As on LLMNode (including the O(n^2) full-buffer trade-off); applies to the _instructions and _step<N> streams, not to the structured output stream.

type : Literal['agent']¶

prompt : str¶

inputs : dict[str, ArgumentInfo]¶

output : str¶

next : str¶

allowed_tools : list[str]¶

max_steps : int¶

llm_model : str | None¶

llm_kwargs : dict[str, Any]¶

stream_output : bool¶

stream_delta : bool¶

stream_instructions : bool¶

stream_partials : bool¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.FunctionNode(*, name: str, type: 'function' = 'function', tool: str, inputs: dict[str, ArgumentInfo] = {}, output: str, next: str, method: str = 'get')[source]¶

Bases: BaseNode

Function-call node.

Invokes a single tool via the FunctionKernel (python:// / rest:// / mcp:// URIs) and stores the result in output.

type : Literal['function']¶

tool : str¶

inputs : dict[str, ArgumentInfo]¶

output : str¶

next : str¶

method : str¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.IfNode(*, name: str, type: 'if' = 'if', condition: str, then: str, else_: str)[source]¶

Bases: BaseNode

Boolean branch node.

Evaluates the condition string expression (e.g. state.count > 3) against the run context and transitions to then when truthy, otherwise to else_ (authored as else in YAML).

type : Literal['if']¶

condition : str¶

then : str¶

else_ : str¶

model_config : ClassVar[ConfigDict] = {'populate_by_name': True, 'validate_by_alias': True, 'validate_by_name': True}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.SwitchNode(*, name: str, type: 'switch' = 'switch', expr: str, cases: dict[str, str] = {}, default: str | None = None)[source]¶

Bases: BaseNode

Multi-way branch node.

Evaluates the expr string expression, stringifies the result and looks it up in cases; falls back to default when no case matches.

type : Literal['switch']¶

expr : str¶

cases : dict[str, str]¶

default : str | None¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Bases: BaseModel

One event in a streamed workflow run (the SSE payload of POST /stream_agent, yielded by WorkflowEngine.run_stream).

Event types and their name:

workflow_started / workflow_completed / workflow_failed: run lifecycle; name is the workflow name. workflow_started carries session_id/run_id; workflow_completed carries output_data and token_usage; workflow_failed carries the error message in value.
node_started / node_completed: node lifecycle; name is the node name.
partial / complete: streamed content. The node’s own output streams under the node name; auxiliary streams are prefixed with it (<node>_thought, <node>_instructions, <node>_step<N>). Whether value is a delta or the full accumulated content depends on the node’s stream_delta setting.
restart: the named stream is starting over (an LLM call was retried after a transient error; value describes the attempt). Clients must discard content accumulated for streams under this name — it will be re-sent.

type : Literal['partial', 'complete', 'restart', 'node_started', 'node_completed', 'workflow_started', 'workflow_completed', 'workflow_failed']¶

name : str¶

value : str | None¶

session_id : str | None¶

run_id : str | None¶

output_data : dict | None¶

token_usage : dict | None¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.WorkflowGraph(*, name: str, description: str = '', version: str = '2.0', llm_model: str | None = None, llm_kwargs: dict[str, ~typing.Any] = <factory>, data_types: dict[str, dict], rest_servers: list[~kavalai.workflow.models.RestServer] = [], mcp_servers: list[~kavalai.workflow.models.McpServer] = [], templates: list[~kavalai.workflow.models.TemplateModel] = [], python_functions: list[~kavalai.workflow.models.PythonFunction] = [], nodes: list[~typing.Annotated[~kavalai.workflow.models.StartNode | ~kavalai.workflow.models.EndNode | ~kavalai.workflow.models.LLMNode | ~kavalai.workflow.models.AgentNode | ~kavalai.workflow.models.FunctionNode | ~kavalai.workflow.models.IfNode | ~kavalai.workflow.models.SwitchNode, FieldInfo(annotation=NoneType, required=True, discriminator='type')]])[source]¶

Bases: BaseModel

A workflow: a directed graph of nodes forming a state machine.

Variables:¶

name : str: Workflow / agent name.
description : str: Human-readable description.
version : str: Schema version.
llm_model : str | None: Default LLM model (provider/model); nodes may override.
llm_kwargs : dict[str, Any]: Default LLM kwargs; nodes may override.
data_types : dict[str, dict]: JSON-schema data type definitions (parsed by SchemaParser).
nodes : list[kavalai.workflow.models.StartNode | kavalai.workflow.models.EndNode | kavalai.workflow.models.LLMNode | kavalai.workflow.models.AgentNode | kavalai.workflow.models.FunctionNode | kavalai.workflow.models.IfNode | kavalai.workflow.models.SwitchNode]: The graph vertices; exactly one start node, and every end node returns the same output data type.

name : str¶

description : str¶

version : str¶

llm_model : str | None¶

llm_kwargs : dict[str, Any]¶

data_types : dict[str, dict]¶

rest_servers : list[RestServer]¶

mcp_servers : list[McpServer]¶

templates : list[TemplateModel]¶

python_functions : list[PythonFunction]¶

nodes : list[Annotated[StartNode | EndNode | LLMNode | AgentNode | FunctionNode | IfNode | SwitchNode, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]¶

validate_graph() → WorkflowGraph[source]¶

property start : str¶: Name of the workflow’s entry point (its single start node).

property output_type : str¶: Name of the data type every end node returns.

property node_map : dict[str, Annotated[StartNode | EndNode | LLMNode | AgentNode | FunctionNode | IfNode | SwitchNode, FieldInfo(annotation=NoneType, required=True, discriminator='type')]]¶

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Run state¶

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

class kavalai.workflow.state.WorkflowState(*, workflow_name: str, status: ~typing.Literal['pending', 'running', 'completed', 'failed'] = 'pending', current_node: str | None = None, trace: list[str] = <factory>, data: dict = <factory>, input_data: dict = <factory>, output_data: dict | None = None, error: str | None = None, invocation_id: str | None = None, token_usage: dict | None = None, run_id: str | None = None, session_id: str | None = None, agent_id: str | None = None)[source]¶

Bases: BaseModel

Serializable runtime state of a single workflow interaction.

The state is JSON round-trippable (to_json / from_json) and is returned by WorkflowEngine.run() so a run can be inspected.

workflow_name: name of the workflow being executed. status: lifecycle status of the run. current_node: name of the node about to run / last run. trace: ordered list of executed node names. data: the run context data (RunContext.data) passed through to_plain. input_data: the original interaction input. output_data: the value of the end node’s output variable (when finished). error: error message when status == 'failed'. invocation_id: short id shared by every log line of this run (for scanning). token_usage: aggregate model token counts for the run. run_id / session_id / agent_id: persistence identifiers (string UUIDs).

workflow_name : str¶

status : Literal['pending', 'running', 'completed', 'failed']¶

current_node : str | None¶

trace : list[str]¶

data : dict¶

input_data : dict¶

output_data : dict | None¶

error : str | None¶

invocation_id : str | None¶

token_usage : dict | None¶

run_id : str | None¶

session_id : str | None¶

agent_id : str | None¶

to_json() → str[source]¶: Serialize the state to a JSON string.

classmethod from_json(data: str) → WorkflowState[source]¶: Deserialize a WorkflowState from a JSON string.

model_config : ClassVar[ConfigDict] = {}¶: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Expressions¶

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

exception kavalai.workflow.expressions.ExpressionError[source]¶

Bases: ValueError

Raised when an expression cannot be parsed or safely evaluated.

Signals an invalid or empty expression, a syntax error, use of an unsupported/disallowed construct (function calls, comprehensions, imports, etc.) or an error encountered while evaluating the expression.

kavalai.workflow.expressions.evaluate_expression(expr: str, context: dict) → Any[source]¶

Safely evaluate a simple string expression against context.

Supports comparisons (==, !=, <, <=, >, >=, in, not in, is, is not), boolean logic (and, or, not), arithmetic (+, -, *, /, //, %), literals, and list/tuple/dict displays. Names and attribute/subscript chains (state.count, input.user_message, items[0].title) are resolved from context via resolve_path(); unknown names resolve to None.

Arbitrary code is rejected: function calls, lambdas, comprehensions, imports, attribute writes, etc. all raise ExpressionError.

kavalai.workflow.expressions.evaluate_bool(expr: str, context: dict) → bool[source]¶: Evaluate expr and coerce the result to a bool (for if nodes).

kavalai.workflow.expressions.evaluate_value(expr: str, context: dict) → str[source]¶: Evaluate expr and stringify the result (for switch case lookup).

Client factory¶

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

kavalai.workflow.clients.build_parameters(llm_kwargs: dict[str, Any] | None) → LlmClientParameters[source]¶

Build LlmClientParameters from a node’s llm_kwargs.

Recognised keys (temperature, top_p, reasoning_effort, service_tier, timeout_seconds) are mapped onto the parameters model; unknown keys are ignored so authors can keep provider-specific extras without breaking.

kavalai.workflow.clients.make_client(model: str, parameters: LlmClientParameters | None = None, stats_receiver: ModelStatsReceiver | None = None) → BaseLlmClient[source]¶

Construct a v2 LLM client from a provider/model string.

Supported providers: openai, gemini, anthropic, ollama, browser. The browser provider runs inference client-side via a WebLLM bridge (Pyodide only) and needs no API key — see BrowserLLMClient.

Task logging backends¶

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

class kavalai.workflow.tasklog.base.TaskLogger[source]¶

Bases: ABC

Common interface for storing per-node debugging data and model stats.

Logging is fire-and-forget: the public log_* methods schedule a background task and return immediately so they never block workflow execution. Call flush() (e.g. at the end of a run or in tests) to await all pending writes.

log_node(*, run_id: str | None, session_id: str | None, agent_id: str | None, node_name: str, node_type: str, inputs: dict | None, output: Any, prompt: str | None = None, duration: float = 0.0, errors: list[str] | None = None) → None[source]¶: Record the execution of a single node (fire-and-forget).

log_model_call(stats: ModelCallStat, agent_id: str | None = None) → None[source]¶: Record an LLM / embedding model call (fire-and-forget).

async flush() → None[source]¶: Await all pending background writes.

async close() → None[source]¶: Flush and release backend resources. Override to add cleanup.

class kavalai.workflow.tasklog.base.StatsBridge(task_logger: TaskLogger, agent_id: str | None = None)[source]¶

Bases: ModelStatsReceiver

Adapter forwarding LLM ModelCallStat events to a TaskLogger.

Wired into v2 LLM clients via their model_stats_receiver so every model call made during a workflow is logged against the run’s agent.

receive_model_stats(stats: ModelCallStat) → None[source]¶

class kavalai.workflow.tasklog.base.TokenAccumulator(task_logger: TaskLogger | None = None, agent_id: str | None = None)[source]¶

Bases: ModelStatsReceiver

Aggregates token usage across a workflow run and optionally forwards each ModelCallStat to a TaskLogger.

The engine wires one accumulator into every LLM client built during a run so that, when the run ends, it can report the total token spend. When a task_logger is supplied each individual call is still logged through it, so this fully subsumes StatsBridge.

receive_model_stats(stats: ModelCallStat) → None[source]¶

summary() → dict[source]¶: Return the aggregated token counts as a plain dict.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

class kavalai.workflow.tasklog.sqlite.SqliteTaskLogger(path: str = ':memory:')[source]¶

Bases: TaskLogger

Task logger storing node executions and model stats in SQLite.

Defaults to a private :memory: database. Pass a file path to keep the debugging data across runs.

async close() → None[source]¶: Flush and release backend resources. Override to add cleanup.

Licensed under the Apache License, Version 2.0 (the “License”); you may not use this file except in compliance with the License. You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

class kavalai.workflow.tasklog.postgres.PostgresTaskLogger(agent_service: AgentService)[source]¶

Bases: TaskLogger

Postgres-backed TaskLogger delegating to AgentService.

Node executions become tasks rows (with their node_type) and model calls become model_call_stats rows, keeping the backoffice dashboards populated for v2 runs.

classmethod from_session_maker(session_maker: async_sessionmaker[AsyncSession]) → PostgresTaskLogger[source]¶