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¶

Copyright 2026 OÜ KAVAL AI (registry code 17393877)

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.

DEFAULT_RAG_SERVICE is the name a rag_query node resolves to when neither it nor the workflow names a service. It matches BaseRagService’s own default collection and source identifiers, so the simple case never has to name anything.

class kavalai.workflow.engine.BranchDecision(expr: str, value: Any, taken: str | None, matched: bool)[source]¶

Bases: object

Why an if or switch node routed where it did.

Returned by WorkflowEngine._next_node() so the routing itself stays a pure function and the recording happens at its single call site.

expr is the condition or expression verbatim, as written in the YAML — kept as text so an old recorded run stays interpretable after the graph changes. value is what it evaluated to: a bool for if, the switched value for switch. This is the diagnostic — a mis-route is usually an upstream classifier emitting "Refund" where the graph expects "refund". taken is the node routed to, or None when nothing matched. matched says, for a switch, whether an explicit case matched; False means the value fell through to default — almost always an upstream bug.

expr : str¶

value : Any¶

taken : str | None¶

matched : bool¶

as_inputs() → dict[source]¶

as_output() → dict[source]¶

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, rag_services: Any | dict[str, Any] | None = None, max_node_visits: int = 1000, default_llm_model: str | None = None, default_llm_parameters: dict | None = None, record_context: bool = True, run_timeout: float | None = None)[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.

rag_services : Optional[BaseRagService | dict[str, BaseRagService]]¶

Services available to rag_query nodes. A bare service is registered under "default", which is what a node resolves to when neither it nor the workflow names one. Anything not passed here is looked up among the services registered with register_rag_service(), so a workflow served by python -m kavalai.server — where nobody constructs the engine — still works.

max_node_visits : int¶

Safety cap on total node executions to guard against infinite loops.

default_llm_model : Optional[str]¶

Model used when neither a node nor the graph names one, as provider/model. The engine reads no environment variable for it: python -m kavalai.server passes KAVALAI_DEFAULT_LLM_MODEL here.

default_llm_parameters : Optional[dict]¶

Fleet-wide llm_kwargs applied to every model call, below the graph’s and the node’s own: node > graph > these > provider defaults. python -m kavalai.server fills it from the KAVALAI_LLM_* variables.

record_context : bool¶

Whether runs.context keeps every node’s data (the default) or only the run’s input and output. The full context is what makes a run explainable afterwards; it is also a second copy of everything the run saw, which a deployer bound by data-protection rules may not want to keep. Off is an explicit choice, never a silent one.

run_timeout : Optional[float]¶

Seconds after which a run is cancelled and recorded as failed, unless run_stream(timeout=...) says otherwise. None (the default) sets no limit. python -m kavalai.server fills it from KAVALAI_AGENT_RUN_TIMEOUT_SECONDS.

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.

async connect() → WorkflowEngine[source]¶

Open the connections the workflow’s tool servers need.

Only MCP servers need this: they are separate processes (or HTTP endpoints) whose tool lists are discovered on connect. Calling it up front means a misconfigured server fails before any tokens are spent, and — more importantly — that an agent node is told about the MCP tools it has, instead of being handed an empty list because nothing has called one yet.

Safe to call more than once; already-connected servers are skipped. The engine still connects lazily on first use if you never call this, so it is an optimisation and a fail-fast check rather than a requirement.

Returns:¶

The engine, so it can be used as engine = await WorkflowEngine.from_yaml_path(...).connect().

async aclose() → None[source]¶

Release the tool servers this engine owns.

Call once when the engine is discarded — from a FastAPI lifespan, or at the end of a script. Runs must not do this: the kernel is engine state, shared by every run, so closing it mid-flight would tear down MCP sessions other runs are still using.

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

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

Execute the workflow for input_data and return the final state.

Drains run_stream() — the single execution path.

Parameters:¶

task_logger: TaskLogger | None = None¶

Records this run only, instead of the engine’s own logger. Pass a MemoryTaskLogger to get the run’s trajectory back without a database — this is what the evaluation runner does, and what makes one engine safe to share across concurrent runs that each want their own trace.

templates: dict[str, str] | None = None¶

Template values for this run only. See run_stream().

timeout: float | None = None¶

Seconds before the run is cancelled. See run_stream().

async run_stream(input_data: dict, *, session_id: str | None = None, external_id: str | None = None, state: WorkflowState | None = None, task_logger: TaskLogger | None = None, templates: dict[str, str] | None = None, timeout: float | 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.

task_logger: TaskLogger | None = None¶

Records this run only, instead of the engine’s own logger. See run().

templates: dict[str, str] | None = None¶

Values for templates the workflow declares, for this run only. They are rendered in one pass like every template value, so text passed here is never interpreted as template syntax, and they are recorded with the run. This is a Python argument and deliberately not a field of the HTTP request: a request field would let a caller rewrite the author’s instructions.

timeout: float | None = None¶

Seconds after which the run is cancelled — parallel branches included — and recorded as failed with a WorkflowTimeoutError. Defaults to the engine’s run_timeout.

Copyright 2026 OÜ KAVAL AI (registry code 17393877)

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.

class kavalai.workflow.builder.WorkflowBuilder(name: str, *, description: str = '', version: str = '1.0', llm_model: str | None = None, llm_kwargs: dict[str, Any] | None = None, rag_service: str | None = None, rag_collection: str | 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()
)

data_type(name: str, fields: dict[str, str | type | dict] | None = None, *, schema: dict | None = None, ref: str | None = None) → WorkflowBuilder[source]¶

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, history_limit: int = 50, history_max_chars: int | None = None, 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, use_history: bool = True, history_limit: int = 50, history_max_chars: int | None = None, 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]¶

rag_query(name: str, *, query: str, output: str, next: str, service: str | None = None, collection: str | None = None, top_k: int = 5, source_ids: list[str] | None = None, match: dict | None = None, keep_best: bool = False, min_similarity: float | None = None, store: str = 'results') → WorkflowBuilder[source]¶

Add a read-only retrieval node.

query is a template, like an llm node’s prompt. service and collection default to the workflow’s rag_service / rag_collection; leave both unset when there is only one index. source_ids=[] matches nothing; None does not filter. match keeps only entries whose metadata has every given key equal, inside the nearest-neighbour search; its string values are templates. min_similarity drops weaker hits. store="content" keeps just the hit texts, which is what a following llm node’s prompt usually wants.

parallel(name: str, *, branches: list[str], next: str, max_concurrency: int | None = None) → WorkflowBuilder[source]¶

Fan out to independent branches and rejoin at next.

Each entry in branches names the first node of a branch; every branch is walked concurrently up to next. Branch subgraphs must be disjoint and must not write the same output variable — both are checked by build().

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¶

Copyright 2026 OÜ KAVAL AI (registry code 17393877)

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.

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.

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

Bases: WorkflowException

A run exceeded its time limit and was cancelled.

Raised by run_stream() when timeout (or the engine’s run_timeout) elapses. The run is recorded as failed with this message, and its parallel branches are cancelled with it.

class kavalai.workflow.models.ArgumentInfo(*, type: 'literal' | 'context' | 'history', value: BaseModel | str | int | float | bool | None = None, name: str | None = None)[source]¶

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, history_limit: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 50, history_max_chars: ~typing.Annotated[int | None, ~annotated_types.Ge(ge=1)] = None, 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).

Chat history (read only when use_history is true and the engine has an AgentService):

history_limit

The most recent messages of the session sent with the prompt, the current user message included. 0 sends none.

history_max_chars

A ceiling on the characters of those messages. Whole messages are dropped from the oldest end until the rest fits; a message is never cut. Characters rather than tokens, so the budget means the same for every provider and needs no tokenizer; it is a cost ceiling, not an exact token count.

type : Literal['llm']¶

prompt : str¶

inputs : dict[str, ArgumentInfo]¶

output : str¶

next : str¶

use_history : bool¶

history_limit : int¶

history_max_chars : int | None¶

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] | None = None, max_steps: int = 10, use_history: bool = True, history_limit: ~typing.Annotated[int, ~annotated_types.Ge(ge=0)] = 50, history_max_chars: ~typing.Annotated[int | None, ~annotated_types.Ge(ge=1)] = None, 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.

Chat history (read only when use_history is true and the engine has an AgentService): the earlier turns of the session are sent with every step, between the system prompt and the step message. The current user message is left out — the node’s prompt already carries it, and the model would read it twice.

history_limit

The most recent earlier messages sent. 0 sends none.

history_max_chars

A ceiling on the characters of those messages, applied as on LLMNode.

allowed_tools restricts what this node may call, and means the same here as it does in the Python API:

allowed_tools: ["*"]                        # every registered tool
allowed_tools: ["mcp://github.*"]           # one server's tools
allowed_tools: ["python://lookup_resident"] # one tool
allowed_tools: []                           # no tools at all
# key omitted                               # every registered tool

type : Literal['agent']¶

prompt : str¶

inputs : dict[str, ArgumentInfo]¶

output : str¶

next : str¶

allowed_tools : list[str] | None¶

max_steps : int¶

use_history : bool¶

history_limit : int¶

history_max_chars : int | None¶

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].

class kavalai.workflow.models.ParallelNode(*, name: str, type: 'parallel' = 'parallel', branches: list[str], next: str, max_concurrency: int | None = None)[source]¶

Bases: BaseNode

Fan-out node: runs several independent branches concurrently.

Each name in branches is the entry node of a branch — a subgraph walked exactly like the main graph, up to (and not including) the join node named by next. All branches start together, and the run continues at next once every branch has reached it.

Branches each get their own copy of the run context, so a node in one branch cannot see another branch’s outputs while they run; the outputs are merged back into the parent context at the join. Because of that, branches must be independent, which the graph validator enforces at load time:

• branch subgraphs must be disjoint from each other,

• no two branches may write the same output variable,

• a branch may not contain an end node or re-enter the parallel node itself.

Loops, if/switch routing and nested parallel nodes inside a branch are all fine.

- name: gather
  type: parallel
  branches: [fetch_weather, fetch_news]
  next: summarise
  max_concurrency: 4

Variables:¶

branches : list[str]¶

Entry node of each branch. Branches run concurrently.

next : str¶

The join node; every branch ends by transitioning to it, and the run resumes there once all branches have arrived.

max_concurrency : int | None¶

Cap on branches running at once (None = all of them). Useful when the branches hit a rate-limited provider.

type : Literal['parallel']¶

branches : list[str]¶

next : str¶

max_concurrency : int | None¶

check_branches() → ParallelNode[source]¶

model_config : ClassVar[ConfigDict] = {}¶

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.RagQueryNode(*, name: str, type: 'rag_query' = 'rag_query', query: str, output: str, next: str, service: str | None = None, collection: str | None = None, top_k: int = 5, source_ids: list[str] | None = None, match: Annotated[dict[str, str | int | float | bool] | None, MinLen(min_length=1)] = None, keep_best: bool = False, min_similarity: Annotated[float | None, Ge(ge=-1.0), Le(le=1.0)] = None, store: 'results' | 'content' = 'results')[source]¶

Bases: BaseNode

Retrieval node: queries a RAG service and stores the hits.

Read-only by construction. The node calls exactly one method, BaseRagService.query; nothing that writes to an index is reachable from a workflow document. Retrieval is idempotent, so a cycle revisiting the node is harmless, whereas indexing is a side effect whose blast radius is a corrupted index – that belongs in a script or a tool, where the author is holding the service object.

query is a template rendered exactly like an llm node’s prompt, so it may reference {{ context.* }}, {{ templates.* }} and {{ history.* }}.

service and collection follow the llm_model pattern: a workflow-level default that a node may override. The common case is one service and one collection, and that case needs neither field:

- name: retrieve
  type: rag_query
  query: "{{ context.user_message }}"
  output: docs
  next: answer

The service is resolved as node → workflow rag_service → "default", looked up first among the services passed to the engine and then among those registered with register_rag_service(). A workflow names a service; it never carries a connection string, because the document is served by GET /workflow and edited in the backoffice.

Variables:¶

query : str¶

Query text, rendered as a template.

output : str¶

Context variable the hits are stored under.

next : str¶

The node to continue at.

service : str | None¶

Registered service name. Defaults to the workflow’s rag_service, then "default".

collection : str | None¶

Collection to search. Defaults to the workflow’s rag_collection, then the backend’s own default.

top_k : int¶

Maximum number of hits.

source_ids : list[str] | None¶

Restrict the search to these source identifiers. Absent means no restriction; an empty list matches nothing, so a filter computed to be empty never widens into a search of everything.

match : dict[str, str | int | float | bool] | None¶

Return only entries whose metadata has every key of match equal — top-level keys and scalar values, the condition query() takes as match. Each string value is a template: one that consists of a single {{ context.* }} placeholder takes the referenced value as it is, number or boolean included, so a filter extracted by an earlier node keeps its type; any other string is rendered as text. The filter is a condition of the nearest-neighbour search itself, so top_k matching entries come back when the collection holds that many.

keep_best : bool¶

Keep only the best hit per source_id. Useful when one document was indexed as many chunks.

min_similarity : float | None¶

Drop hits whose similarity is below this value. similarity is higher-is-better cosine on every backend, so one number means the same thing on each; a useful threshold still depends on the embedding model and on whether a normaliser is used.

store : Literal['results', 'content']¶

"results" (default) stores the full RagServiceResult list, so scores and metadata stay available for routing. "content" stores just the hit texts joined by blank lines, which is what a following llm node’s prompt usually wants.

Whatever store says, the node records its hits — id, source_id, similarity and metadata, without the text — on its task row and in the output_data of its node_completed event, so the passages behind an answer can be cited and audited.

type : Literal['rag_query']¶

query : str¶

output : str¶

next : str¶

service : str | None¶

collection : str | None¶

top_k : int¶

source_ids : list[str] | None¶

match : dict[str, str | int | float | bool] | None¶

keep_best : bool¶

min_similarity : float | None¶

store : Literal['results', 'content']¶

model_config : ClassVar[ConfigDict] = {}¶

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class kavalai.workflow.models.WorkflowStreamEvent(*, type: 'partial' | 'complete' | 'restart' | 'node_started' | 'node_completed' | 'workflow_started' | 'workflow_completed' | 'workflow_failed', name: str, value: str | None = None, session_id: str | None = None, run_id: str | None = None, output_data: dict | None = None, token_usage: dict | None = None)[source]¶

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. A rag_query node’s node_completed carries its hits in output_data ({"hits": [...]}, without the text).

• 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. Only a node that streams emits it: a restart exists to tell a client to discard partial output, and a node that streams nothing has none to discard.

The engine’s events are complete, because in-process consumers (tests, notebooks, the evaluation runner) want the full error text and token counts. What reaches an HTTP client is decided at the server boundary; see kavalai.server.public_events.

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].