Workflow YAML reference¶

A workflow file is a YAML document describing one WorkflowGraph. This page lists every key, exhaustively. For a guided introduction read Workflows; for the ideas behind the model read Workflows.

Load one with:

from kavalai import WorkflowEngine

engine = WorkflowEngine.from_yaml_path(
    "support_agent.yaml", agent_service=service
)
state = await engine.run({"user_message": "I want a refund"})

from_yaml (a string) and from_dict (an already-parsed dict) take the same keyword arguments.

Top level¶

Validation happens when the graph is loaded, not when it runs. A workflow is rejected if node names collide, if there is not exactly one start node, if there is no end node, if two end nodes return different data types, if a transition names a node that does not exist, if a node writes to an output that is not declared in data_types, or if an llm_kwargs key or value is not one LlmClientParameters accepts.

data_types¶

Each entry is a JSON-schema object compiled into a Pydantic model, so every value crossing a node boundary is validated.

Two names are special:

• input — the workflow’s own input type; what the caller passes to run().

• the type named by the end node’s output (output by convention) — what the caller gets back.

data_types:
  input:
    type: object
    properties:
      user_message: {type: string}
  classification:
    type: object
    properties:
      intent: {type: string}
      confidence: {type: number}
    required: [intent]
  output:
    type: object
    properties:
      agent_response: {type: string}

Nodes¶

Every node has a name (unique) and a type. The remaining keys depend on the type.

start¶

Entry point; receives the workflow input. Exactly one per graph.

- {name: begin, type: start, next: classify}

end¶

Exit point. A graph may have several, but they must all name the same output type.

- {name: finish, type: end, output: output}

llm¶

One structured LLM completion. The prompt is rendered, the model is called, and the validated result is stored under output.

- name: classify
  type: llm
  prompt: |
    Classify the villager's message as exactly one of: repair, permit, other.
    Respond with that single lowercase word in the `intent` field.
  inputs:
    message: {type: context, value: input}
  output: classification
  next: route
  use_history: false
  llm_model: openai/gpt-5.6-luna
  stream_output: true
  stream_delta: true

agent¶

A multi-step, tool-using Agent loop inside the graph. Use it when the model should decide which tools to call; use function when you already know.

- name: research
  type: agent
  prompt: "Research the company and summarise what it sells."
  inputs:
    company: {type: context, value: input.company}
  output: summary
  allowed_tools: ["python://web.crawl", "rest://crm.*"]
  max_steps: 6
  next: write_up

Takes every llm key above except use_history, history_limit and history_max_chars — an agent loop does not replay the chat history — plus:

allowed_tools: ["*"]                         # every tool
allowed_tools: ["mcp://github.*"]            # one server
allowed_tools: ["python://lookup_resident"]  # one tool
allowed_tools: []                            # none

function¶

Exactly one tool call through the FunctionKernel, addressed by URI.

- name: measure
  type: function
  tool: python://measure_pond
  inputs:
    name: {type: context, value: input.pond}
  output: reading
  next: phrase

rag_query¶

One retrieval against a RAG service. Read-only — the node calls query and nothing else, so no workflow document can write to an index.

query is a template, rendered exactly like an llm node’s prompt.

- name: retrieve
  type: rag_query
  query: "{{ context.input.question }}"
  output: facts
  next: answer

That is the whole node when there is one index: service and collection default to the workflow’s rag_service / rag_collection, and a single service passed to the engine is registered as "default".

Whatever store says, the node records its hits — each one’s 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. The query embedding is reported to the run’s token count and recorded as a model call of the run.

The service itself is supplied by the caller (WorkflowEngine(..., rag_services=my_service)) or registered with register_rag_service(); a node naming a service that is neither fails when the workflow loads, not when the branch first runs.

if¶

Branches on a boolean condition, evaluated by the safe expression language.

- name: check_confidence
  type: if
  condition: "classification.confidence >= 0.8"
  then: reply
  else: escalate

then and else are both required. Note else is a YAML key, not a Python keyword — write it plainly.

switch¶

Evaluates expr, converts the result to a string, and matches it against cases.

- name: route
  type: switch
  expr: classification.intent
  cases:
    repair: repair_reply
    permit: permit_reply
  default: general_reply

If no case matches and there is no default, the run fails with a WorkflowException.

parallel¶

Runs several independent branches concurrently and rejoins them at a single node.

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

Each name in branches is the entry node of a branch — an ordinary subgraph, walked exactly as the main graph is, up to but not including the join. All branches start together, and the run resumes at next once every branch has arrived there.

Each branch receives its own copy of the run context, so a node in one branch cannot observe a sibling’s output while both are running; outputs are merged into the parent context at the join. Because branches must therefore be independent, the graph validator rejects a workflow at load time if branch subgraphs overlap, if two branches write the same output variable, or if a branch contains an end node or re-enters the parallel node itself.

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

Three properties of a fan-out are worth knowing when reading a run back. Branch events interleave onto the single event stream as they are produced, each tagged with its own node name, so a streaming client can separate them. The recorded trace, by contrast, collects each branch’s nodes and appends them in declaration order, so state.trace is stable across runs even though the execution is not. And the first branch to raise cancels its siblings and propagates, so a failed run does not leave a long branch running behind it.

Note that tool calls made within a single agent node already execute concurrently; parallel concerns concurrency between nodes.

Node inputs¶

inputs maps a local name to a value resolved before the node runs:

inputs:
  message:  {type: context, value: input.user_message}
  tone:     {type: literal, value: "formal"}
  previous: {type: history, value: last_order_id}

In the WorkflowBuilder a bare string is shorthand for a context path — inputs={"message": "input"}.

Interpolation in prompts¶

A prompt may interpolate three prefixes:

prompt: |
  {{ templates.house_style }}

  The villager wrote: {{ context.input.user_message }}
  Their last order was {{ history.last_order_id }}.

• {{ context.PATH }} — the current run context.

• {{ templates.NAME }} — a fragment from the top-level templates list.

• {{ history.PATH }} — a value from an earlier run in the session.

Dicts and lists are inserted as JSON. An unresolvable reference raises rather than rendering empty, so a typo fails loudly instead of silently weakening the prompt.

Rendering is a single pass: a value that is inserted is never rendered again, so text containing {{ … }} arrives in the prompt literally. This is what makes per-run template values safe. A template declared in the document may be given another value for one run from Python — engine.run(data, templates={"house_style": text}) — and the values used are recorded with the run. Only declared templates may be overridden, and the override is a Python argument, never a field of the HTTP request: a request field would let a caller rewrite the author’s instructions.

Tool servers¶

Tools declared at the top level are registered on the engine’s kernel before the run, so function and agent nodes can address them.

python_functions:
  - {name: measure_pond, path: green_village.tools.measure_pond}

rest_servers:
  - name: crm
    url: https://crm.example.com/api
    username_env: CRM_USER
    password_env: CRM_PASSWORD

mcp_servers:
  - name: village
    command: python
    args: ["-m", "village_mcp"]

Use the *_env variants for anything secret: they keep credentials out of the workflow file, which is usually committed to source control.

MCP servers are started and asked for their tools when the engine connects, so an agent node sees them on the first run. Call await engine.connect() at startup if you want a misconfigured server to fail there rather than mid-run — see Tools.

REST tools themselves are declared in code with register_rest_tool() (they need input and output schemas); the YAML declares the server.

Execution limits¶

A run stops with a WorkflowException after max_node_visits node visits — 1000 by default, set on the engine, not in YAML. It is a backstop against a cycle that never terminates: loops in a graph are allowed, so this is what makes them safe.

engine = WorkflowEngine.from_yaml_path("workflow.yaml", max_node_visits=50)

A run can also be bounded in time. run_timeout on the engine, or timeout on a single run / run_stream call, is the number of seconds after which the run is cancelled — parallel branches included — and recorded as failed with a WorkflowTimeoutError. The agent server fills run_timeout from KAVALAI_AGENT_RUN_TIMEOUT_SECONDS. Neither is a YAML key: how long a run may take is a property of the deployment, not of the graph.

A complete example¶

# Example v2 workflow: a DAG / state machine.
#
# The user hands an input to the `start` node and reads the result from the
# `end` node. `if` / `switch` nodes route on simple string expressions
# evaluated against the run context (e.g. `classification.intent == 'refund'`).
#
# Run it with:
#   from kavalai.agent_service import AgentService
#   from kavalai.db import db_manager
#   from kavalai.workflow import WorkflowEngine, SqliteTaskLogger
#
#   await db_manager.init_sqlite()  # in-memory SQLite; point at Postgres in prod
#   engine = WorkflowEngine.from_yaml_path(
#       "examples/support_agent/support_agent.yaml",
#       agent_service=AgentService(db_manager.get_sqlite_sessionmaker()),
#       task_logger=SqliteTaskLogger(),
#   )
#   state = await engine.run({"user_message": "I want a refund"})
#   print(state.output_data)
name: Support agent
description: Routes a support request and produces a tailored response.
llm_model: openai/gpt-5.6-luna

data_types:
  input:
    type: object
    properties:
      user_message:
        type: string
  classification:
    type: object
    properties:
      intent:
        type: string
  output:
    type: object
    properties:
      agent_response:
        type: string

nodes:
  - name: begin
    type: start
    next: classify

  # LLM node: classify the user's intent into a structured output.
  - name: classify
    type: llm
    prompt: >
      Classify the user's intent as one of: refund, technical, other.
      Return it in the `intent` field.
    inputs:
      input:
        type: context
        value: input
    output: classification
    next: route

  # Switch node: branch on a simple string expression.
  - name: route
    type: switch
    expr: classification.intent
    cases:
      refund: handle_refund
      technical: handle_technical
    default: handle_general

  # Agent node: multi-step tool-using agent for technical questions.
  - name: handle_technical
    type: agent
    prompt: "Help the user resolve their technical problem."
    inputs:
      input:
        type: context
        value: input
    output: output
    max_steps: 5
    next: finish

  - name: handle_refund
    type: llm
    prompt: >-
      Acknowledge the refund request empathetically and explain
      next steps.
    inputs:
      input:
        type: context
        value: input
    output: output
    next: finish

  - name: handle_general
    type: llm
    prompt: "Answer the user's general question."
    inputs:
      input:
        type: context
        value: input
    output: output
    next: finish

  - name: finish
    type: end
    output: output