Pipelines API

A pipeline is a multi-step workflow that can be described in code or in JSON. See the pipelines guide for the narrative version; this page is the reference.

from ToolAgents.pipelines import Pipeline, SequentialProcess, ProcessStep

Pipeline

Pipeline

Pipeline(
    agent_configs: Sequence["AgentConfig"] | None = None,
    default_agent_name: str | None = None,
)

Main pipeline class that manages the execution of multiple processes.

The pipeline maintains a list of processes and executes them in sequence, passing results between processes.

Initialize an empty pipeline.

Parameters:

Name	Type	Description	Default
agent_configs	Sequence['AgentConfig'] | None	Optional declarative agent/endpoint configurations. These round-trip through JSON; the agents themselves are built at load time from environment-held API keys.	None
default_agent_name	str | None	Name of the declared agent used by processes that name none.	None

add_agent_config

add_agent_config(agent_config: 'AgentConfig') -> 'Pipeline'

Declare an agent/endpoint that processes can reference by name.

add_process

add_process(process: Process)

Add a new process to the pipeline.

add_processes

add_processes(processes: list[Process])

Add new processes to the pipeline.

run_pipeline

run_pipeline(**kwargs) -> PipelineResults

Execute all processes in the pipeline sequentially.

Results from each process are passed as input to the next process.

Keyword arguments become the inputs section; step results land in outputs. The returned object still reads like the flat dictionary it replaced — results["greeting"] resolves by scope order — so existing calling code is unaffected.

run

run(results: PipelineResults) -> PipelineResults

Execute every process against an existing results object.

run_pipeline is the usual entry point; this one is for a caller that needs to seed a section other than inputs -- shared prompt text, for instance -- before the run begins.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
    include_tool_plugins: bool = True,
) -> dict[str, Any]

Serialize this pipeline to a JSON-compatible dictionary.

Runtime agents are intentionally omitted. Pass agents back into from_dict/load_from_json when restoring a runnable pipeline.

from_dict classmethod

from_dict(
    data: Mapping[str, Any],
    tool_registry: PipelineToolRegistry | None = None,
    default_agent: BaseToolAgent | None = None,
    process_agents: (
        Mapping[str, BaseToolAgent] | None
    ) = None,
    step_agents: Mapping[str, BaseToolAgent] | None = None,
    load_tool_plugins: bool = True,
    build_agents: bool = True,
    allow_writes: bool = False,
) -> "Pipeline"

Restore a pipeline from a JSON-compatible dictionary.

If load_tool_plugins is true, plugin declarations in the JSON are imported with Python's import machinery. Only load JSON files that you trust, or pass a prebuilt tool_registry and disable plugin loading.

If build_agents is true, any agents block in the JSON is used to construct providers, reading API keys from the environment variables the config names. Pass build_agents=False to ignore the block entirely and supply every agent from Python instead.

allow_writes gates sinks that write files or make HTTP requests. It is false by default: loading a document should not let it reach outside the process unless the caller says so. Sources that only read are always permitted.

Agents injected here always win over names declared in the JSON at the same level of specificity; see :class:PipelineLoadContext.

to_json

to_json(
    tool_registry: PipelineToolRegistry | None = None,
    include_tool_plugins: bool = True,
    indent: int | None = 2,
) -> str

Serialize this pipeline to a JSON string.

from_json classmethod

from_json(
    json_text: str,
    tool_registry: PipelineToolRegistry | None = None,
    default_agent: BaseToolAgent | None = None,
    process_agents: (
        Mapping[str, BaseToolAgent] | None
    ) = None,
    step_agents: Mapping[str, BaseToolAgent] | None = None,
    load_tool_plugins: bool = True,
    build_agents: bool = True,
    allow_writes: bool = False,
) -> "Pipeline"

Restore a pipeline from a JSON string.

save_to_json

save_to_json(
    filepath: str | PathLike[str],
    tool_registry: PipelineToolRegistry | None = None,
    include_tool_plugins: bool = True,
    indent: int | None = 2,
) -> None

Write this pipeline to a JSON file.

load_from_json classmethod

load_from_json(
    filepath: str | PathLike[str],
    tool_registry: PipelineToolRegistry | None = None,
    default_agent: BaseToolAgent | None = None,
    process_agents: (
        Mapping[str, BaseToolAgent] | None
    ) = None,
    step_agents: Mapping[str, BaseToolAgent] | None = None,
    load_tool_plugins: bool = True,
    build_agents: bool = True,
    allow_writes: bool = False,
) -> "Pipeline"

Load a pipeline from a JSON file.

Results

Results are carried in named sections — inputs, outputs, vars — addressed by path in prompt templates ({outputs/draft}) and by subscript in conditions (outputs['draft']). A bare name still resolves innermost-first.

PipelineResults

PipelineResults(
    inputs: Mapping[str, Any] | None = None,
    outputs: Mapping[str, Any] | None = None,
    vars: Mapping[str, Any] | None = None,
    sections: Mapping[str, Mapping[str, Any]] | None = None,
)

Bases: MutableMapping

A sectioned results mapping that still behaves like the old flat dict.

Reading and writing a bare key works exactly as before, so existing pipelines, prompt templates and calling code need no changes::

results["draft"]            # resolves through vars -> outputs -> inputs
results["draft"] = "..."    # writes to outputs

The structure is available whenever it is wanted::

results.outputs["draft"]
results["outputs/draft"]
results["outputs/news/draft"]

inputs property

inputs: dict[str, Any]

Arguments passed to run_pipeline.

outputs property

outputs: dict[str, Any]

Values produced by steps.

vars property

vars: dict[str, Any]

Flow-control scratch state, scoped to the body that owns it.

section_names property

section_names: list[str]

Return the names of all sections.

coerce classmethod

coerce(
    values: "PipelineResults | Mapping[str, Any] | None",
) -> "PipelineResults"

Return values as a PipelineResults.

A plain mapping is treated as inputs, so a caller that still hands over a flat dictionary keeps working.

copy

copy() -> 'PipelineResults'

Return a copy with independent section dictionaries.

Section dictionaries are copied; the values inside them are not. A body that mutates a nested list or dict still affects the original, which is why bodies should rebind rather than mutate.

section

section(name: str, create: bool = False) -> dict[str, Any]

Return a section by name, optionally creating it.

Adding a section is how a new kind of state joins the namespace — for example results.section("agent", create=True) for agent internals.

to_dict

to_dict() -> dict[str, dict[str, Any]]

Return the sectioned structure as plain dictionaries.

resolve_path

resolve_path(path: str) -> tuple[bool, Any]

Resolve path and return (found, value).

A path beginning with a section name is read from that section; anything else is treated as a bare name and resolved by scope order. Returning a flag rather than raising lets templates and conditions each decide what an absent value means.

set_path

set_path(path: str, value: Any) -> None

Write value at path, creating intermediate mappings.

flat

flat() -> dict[str, Any]

Return the flattened view a bare-name lookup would see.

Outer scopes first, so inner ones overwrite: this is the dictionary the pipeline behaved like before sections existed.

Processes

Every process implements run_process and serializes to JSON. Flow-control processes hold other processes, so they nest freely.

Process

Process(
    process_name: str = "Process",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
)

Bases: ABC

Abstract base class representing a process in the pipeline.

A process is a collection of steps that need to be executed in a specific way. The actual execution logic is defined by concrete implementations.

Initialize a new process.

Parameters:

Name	Type	Description	Default
process_name	str	Name identifier for the process	'Process'
agent	BaseToolAgent	Default agent to use for steps that don't have their own agent	None
agent_name	str | None	Optional name of an agent declared in the pipeline's agents block. The name round-trips to JSON; the agent object is what actually runs.	None

add_step

add_step(step: ProcessStep) -> 'Process'

Add a new step to the process and return self, for chaining.

add_steps

add_steps(steps: list[ProcessStep]) -> 'Process'

Add new steps to the process and return self, for chaining.

run_process abstractmethod

run_process(results: PipelineResults) -> PipelineResults

Execute the process steps according to the implementation logic.

Parameters:

Name	Type	Description	Default
results	PipelineResults	Sectioned results carried through the pipeline	required

Returns:

Name	Type	Description
PipelineResults	PipelineResults	Updated results after process execution

get_name

get_name() -> str

Return the process name.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this process to a JSON-compatible dictionary.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "Process"

Rebuild this process from JSON, using context for agents/tools.

Flow-control processes call :func:processes_from_config with the same context (or a nested one) to rebuild their children.

run_child_processes

run_child_processes(
    processes: Sequence["Process"], results: PipelineResults
) -> PipelineResults

Run child processes in order, threading the results mapping.

A child with no agent of its own inherits this process's agent, so a loop or branch built in Python needs the agent set only once, at the outermost level that has one.

lend_agent

lend_agent(process: 'Process') -> None

Give process this process's agent if it has none of its own.

ProcessStep

ProcessStep(
    step_name: str,
    system_message: str,
    prompt_template: str,
    tools: list[FunctionTool] = None,
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
)

Represents a single step in a process pipeline for LLM tool usage.

Each step contains the necessary configuration for the LLM to perform a specific task, including system message, prompt template, and available tools.

Attributes:

Name	Type	Description
step_name	str	The name identifier for the step. Can be used to reference results of previous steps in the prompt template, for example {outputs/step_name}.
system_message	str	The system message to provide context to the LLM
prompt_template	str	Template string for generating the actual prompt
tools	list[FunctionTool]	List of tools available for this step
agent	BaseToolAgent	The LLM agent responsible for executing this step

Initialize a new process step.

Parameters:

Name	Type	Description	Default
step_name	str	Unique identifier for the step. Its result is written to the outputs section, so a later prompt can reference it as {outputs/step_name}.	required
system_message	str	Context message for the LLM	required
prompt_template	str	Template for generating the actual prompt	required
tools	list[FunctionTool]	Optional list of tools available for this step	None
agent	BaseToolAgent	Optional specific agent for this step	None
agent_name	str | None	Optional name of an agent declared in the pipeline's agents block. The name is what round-trips to JSON; the agent object is what actually runs.	None

get_name

get_name() -> str

Return the step name.

get_system_message

get_system_message(
    fields: "PipelineResults | Mapping[str, Any] | None" = None,
) -> str

Return the system message, with any placeholders filled in.

A system message is a prompt as much as prompt_template is, so it is rendered against the same results. That is what lets a shared prompt file be addressed as {prompts/reviewer} in either field.

Parameters:

Name	Type	Description	Default
fields	'PipelineResults | Mapping[str, Any] | None'	A results mapping. Omit it to get the raw template.	None

get_prompt

get_prompt(
    fields: "PipelineResults | Mapping[str, Any] | None" = None,
    **kwargs: Any
) -> str

Generate the actual prompt using the template and provided parameters.

Parameters:

Name	Type	Description	Default
fields	'PipelineResults | Mapping[str, Any] | None'	A results mapping. Passing the mapping itself, rather than unpacking it, is what lets a template address a section: {outputs/draft} as well as a bare {draft}.	None
**kwargs	Any	Individual template fields, for callers not using a results mapping.	{}

Returns:

Name	Type	Description
str	str	The generated prompt

get_tools

get_tools() -> list[FunctionTool]

Return the list of tools available for this step.

get_agent

get_agent() -> BaseToolAgent

Return the agent assigned to this step.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this step to a JSON-compatible dictionary.

Agents are runtime objects and are intentionally not serialized.

from_dict classmethod

from_dict(
    data: Mapping[str, Any],
    tool_registry: PipelineToolRegistry | None = None,
    agent: BaseToolAgent | None = None,
) -> "ProcessStep"

Restore a process step from a JSON-compatible dictionary.

SequentialProcess

SequentialProcess(
    process_name: str = "SequentialProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
)

Bases: Process

Concrete implementation of Process that executes steps sequentially.

Each step is executed in order, with results from previous steps available to subsequent steps through the results dictionary.

Initialize a sequential process with optional default agent.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this sequential process to a JSON-compatible dictionary.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "SequentialProcess"

Restore a sequential process from its JSON representation.

run_process

run_process(results: PipelineResults) -> PipelineResults

Execute process steps in sequential order.

For each step: 1. Set up tool registry if tools are available 2. Prepare messages with system message and generated prompt 3. Execute step using appropriate agent 4. Store results in results dictionary

Parameters:

Name	Type	Description	Default
results	PipelineResults	Dictionary containing results from previous processes	required

Returns:

Type	Description
PipelineResults	dict[str, Any]: Updated results dictionary after all steps are executed

Raises:

Type	Description
Exception	If no agent is available for a step

TemplateProcess

TemplateProcess(
    template: str,
    result_key: str = "text",
    section: str = "outputs",
    process_name: str = "TemplateProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
)

Bases: Process

Compose a value from existing results, without calling a model.

Joining two results together is string work, not reasoning. Without this
the only way to do it is to ask a model to "return this unchanged", which
costs a request, adds latency, and is not guaranteed to comply.

JSON::

    {
      "process_type": "template",
      "process_name": "assemble",
      "template": "# {outputs/title}

{outputs/body}", "result_key": "digest" }

Initialize a template process.

Parameters:

Name	Type	Description	Default
template	str	Text with {section/key} placeholders.	required
result_key	str	Key the rendered text is written to.	'text'
section	str	Results section written to. Defaults to outputs.	'outputs'
process_name	str	Name identifier for the process.	'TemplateProcess'
agent	BaseToolAgent	Unused; this process calls no model. Accepted so it can sit anywhere a process can.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None

run_process

run_process(results: PipelineResults) -> PipelineResults

Render the template and store the result.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this template process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "TemplateProcess"

Restore a template process from JSON.

Flow control

FlowProcess

FlowProcess(
    process_name: str = "Process",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
)

Bases: Process

Base class for processes whose body is a list of other processes.

add_step is kept working as a convenience: a step added to a flow process is appended to a trailing :class:SequentialProcess in its primary body, created on demand. Without this, add_step on a loop would append to the unused steps list inherited from Process and silently never run.

primary_body

primary_body() -> list[Process]

Return the process list that add_step/add_process extend.

child_bodies

child_bodies() -> tuple[list[Process], ...]

Return every child process list, for traversal.

add_process

add_process(process: Process) -> 'FlowProcess'

Append a child process to this process's primary body.

add_processes

add_processes(
    processes: Sequence[Process],
) -> "FlowProcess"

Append several child processes to this process's primary body.

add_step

add_step(step: ProcessStep) -> 'FlowProcess'

Append a step to the trailing sequence of this process's body.

add_steps

add_steps(steps: Sequence[ProcessStep]) -> 'FlowProcess'

Append several steps to the trailing sequence of this body.

ConditionalProcess

ConditionalProcess(
    condition: Condition | Mapping[str, Any] | str,
    process_name: str = "ConditionalProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
    then_processes: Sequence[Process] | None = None,
    else_processes: Sequence[Process] | None = None,
    then_steps: Sequence[ProcessStep] | None = None,
    else_steps: Sequence[ProcessStep] | None = None,
    record_as: str | None = None,
)

Bases: FlowProcess

Run one branch or the other depending on a condition.

Example::

ConditionalProcess(
    condition="score < 0.7",
    then_steps=[revise_step],
    else_steps=[publish_step],
)

JSON::

{
  "process_type": "conditional",
  "process_name": "gate",
  "condition": "score < 0.7",
  "then": [ ...processes... ],
  "else": [ ...processes... ],
  "record_as": "needed_revision"
}

Initialize a conditional process.

Parameters:

Name	Type	Description	Default
condition	Condition | Mapping[str, Any] | str	A condition object, config mapping, or bare expression string evaluated against the results mapping.	required
process_name	str	Name identifier for the process.	'ConditionalProcess'
agent	BaseToolAgent	Default agent lent to children that have none.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None
then_processes	Sequence[Process] | None	Processes run when the condition holds.	None
else_processes	Sequence[Process] | None	Processes run when it does not.	None
then_steps	Sequence[ProcessStep] | None	Convenience steps wrapped into a sequential process.	None
else_steps	Sequence[ProcessStep] | None	Convenience steps for the else branch.	None
record_as	str | None	Optional results key recording which branch was taken.	None

run_process

run_process(results: PipelineResults) -> PipelineResults

Evaluate the condition and run the matching branch.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this conditional process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "ConditionalProcess"

Restore a conditional process from JSON.

LoopProcess

LoopProcess(
    condition: (
        Condition | Mapping[str, Any] | str | None
    ) = None,
    mode: str = "until",
    max_iterations: int = DEFAULT_MAX_ITERATIONS,
    process_name: str = "LoopProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
    processes: Sequence[Process] | None = None,
    steps: Sequence[ProcessStep] | None = None,
    iteration_var: str = "iteration",
    iterations_key: str | None = None,
    on_max_iterations: str = "stop",
)

Bases: FlowProcess

Repeat a body until a condition is satisfied or a cap is reached.

The two modes differ in when the condition is tested and in what it means. They are inverses, as in most languages, so the same expression cannot simply be moved from one to the other:

"until" (the default) The condition is a stop condition. Run the body, then test; stop when it becomes true. The body always runs at least once, which is what a refine-until-good-enough loop needs, because the condition usually reads a value the body produces.

"while" The condition is a continue condition. Test before each iteration; stop when it becomes false. The body may run zero times, and the condition must only reference values that already exist.

So mode="until", condition="approved" and mode="while", condition="not approved" express the same loop.

max_iterations is always enforced, so a condition that never becomes true cannot spin forever burning API credit.

JSON::

{
  "process_type": "loop",
  "process_name": "refine",
  "mode": "until",
  "condition": "contains(lower(review), 'approved')",
  "max_iterations": 4,
  "on_max_iterations": "stop",
  "processes": [ ...body... ]
}

Initialize a loop process.

Parameters:

Name	Type	Description	Default
condition	Condition | Mapping[str, Any] | str | None	Stop condition. When omitted the loop runs exactly max_iterations times.	None
mode	str	"until" (post-test, body runs at least once) or "while" (pre-test, body may not run at all).	'until'
max_iterations	int	Hard cap on iterations. Must be positive.	DEFAULT_MAX_ITERATIONS
process_name	str	Name identifier for the process.	'LoopProcess'
agent	BaseToolAgent	Default agent lent to children that have none.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None
processes	Sequence[Process] | None	Body processes.	None
steps	Sequence[ProcessStep] | None	Convenience steps wrapped into a sequential body process.	None
iteration_var	str	Results key holding the zero-based iteration index, so prompt templates can reference {iteration}.	'iteration'
iterations_key	str | None	Results key receiving the total iteration count. Defaults to "<process_name>_iterations".	None
on_max_iterations	str	"stop" to exit quietly at the cap, or "error" to raise when the condition was never satisfied.	'stop'

run_process

run_process(results: PipelineResults) -> PipelineResults

Run the body repeatedly according to mode and condition.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this loop process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "LoopProcess"

Restore a loop process from JSON.

MapProcess

MapProcess(
    items: str | SafeExpression,
    process_name: str = "MapProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
    processes: Sequence[Process] | None = None,
    steps: Sequence[ProcessStep] | None = None,
    item_var: str = "item",
    index_var: str = "index",
    collect: str | None = None,
    result_key: str | None = None,
)

Bases: FlowProcess

Run a body once per item of a list, collecting the outputs.

Each iteration runs against its own copy of the results mapping, so an iteration rebinding a key cannot affect the next one, and only the collected list is written back to the outer results. The copy is shallow: a body that mutates a nested list or dict in place still affects the outer value and later iterations. Rebind rather than mutate.

JSON::

{
  "process_type": "map",
  "process_name": "per_topic",
  "items": "topics",
  "item_var": "topic",
  "collect": "draft",
  "result_key": "drafts",
  "processes": [ ...body... ]
}

items is a sandboxed expression, so "topics" and "topics[:3]" are both valid. When collect names a results key, the output is the list of that key's value per iteration. When it is omitted, each entry is a dict of whatever keys that iteration added.

Initialize a map process.

Parameters:

Name	Type	Description	Default
items	str | SafeExpression	Sandboxed expression selecting the list to iterate.	required
process_name	str	Name identifier for the process.	'MapProcess'
agent	BaseToolAgent	Default agent lent to children that have none.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None
processes	Sequence[Process] | None	Body processes run per item.	None
steps	Sequence[ProcessStep] | None	Convenience steps wrapped into a sequential body process.	None
item_var	str	Results key holding the current item.	'item'
index_var	str	Results key holding the zero-based index.	'index'
collect	str | None	Results key gathered from each iteration.	None
result_key	str | None	Where the collected list lands. Defaults to "<process_name>_results".	None

run_process

run_process(results: PipelineResults) -> PipelineResults

Run the body once per item, collecting outputs into a list.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this map process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "MapProcess"

Restore a map process from JSON.

ParallelProcess

ParallelProcess(
    branches: Sequence[Process] | None = None,
    process_name: str = "ParallelProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
    max_workers: int | None = None,
    on_conflict: str = "error",
)

Bases: FlowProcess

Run independent branches concurrently and merge their results.

Each branch runs against its own copy of the results mapping in a worker thread; afterwards the keys each branch added or changed are merged back.

.. warning::

An agent instance is not thread-safe: ``ChatToolAgent`` keeps a
``last_messages_buffer`` on ``self``, so two branches sharing one agent
will interleave their transcripts. The ``.response`` text each step
stores stays correct, but ``ChatResponse.messages`` does not. Give each
branch its own agent when the transcript matters; this process warns
once if branches would share one.

JSON::

{
  "process_type": "parallel",
  "process_name": "research",
  "branches": [ ...one process per branch... ],
  "max_workers": 4,
  "on_conflict": "error"
}

Under on_conflict="section" a contested output moves into a sub-section named for its branch, so it is addressed by structure rather than by a mangled name::

{outputs/news/draft}        in a prompt template
outputs['news']['draft']    in a condition

Any value the key already held at the top of outputs is untouched.

Initialize a parallel process.

Parameters:

Name	Type	Description	Default
branches	Sequence[Process] | None	Processes run concurrently, one per branch.	None
process_name	str	Name identifier for the process.	'ParallelProcess'
agent	BaseToolAgent	Default agent lent to branches that have none. See the thread-safety warning above.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None
max_workers	int | None	Thread pool size. Defaults to the branch count.	None
on_conflict	str	What to do when two branches write the same output with different values: "error", "last_wins", or "section" (give each branch its own sub-section of outputs).	'error'

run_process

run_process(results: PipelineResults) -> PipelineResults

Run every branch concurrently and merge the results.

branch_labels

branch_labels() -> list[str]

Return a unique, template-safe label for each branch.

Labels are used to qualify keys under on_conflict="prefix". They must contain only word characters, because MessageTemplate substitutes on \w+ and the condition sandbox rejects anything with a dot as attribute access — a qualified key that no prompt or condition could read would be worse than useless.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this parallel process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "ParallelProcess"

Restore a parallel process from JSON.

Conditions

Conditions are compiled from a whitelisted subset of Python's grammar, never eval'd, so pipeline JSON from an untrusted source cannot execute arbitrary code.

SafeExpression

SafeExpression(source: str)

A compiled, sandboxed expression evaluated against a results mapping.

The expression is validated once at construction time, so an invalid or unsafe expression fails when the pipeline is loaded rather than midway through an expensive run.

referenced_names

referenced_names() -> set[str]

Return the result keys this expression reads.

Only names used as values count. A name used solely as a call target (len in len(draft)) is a helper, not a result key.

evaluate

evaluate(results: Mapping[str, Any]) -> Any

Evaluate the expression against results and return its value.

Names resolve lazily, so and / or and conditional expressions short-circuit properly: "has_score and score > 3" is well defined even when score does not exist. Reaching a name that is genuinely absent raises an error naming the results that do exist.

evaluate_bool

evaluate_bool(results: Mapping[str, Any]) -> bool

Evaluate the expression and coerce the result to bool.

Condition

Bases: ABC

A serializable boolean test over the pipeline results mapping.

evaluate abstractmethod

evaluate(results: Mapping[str, Any]) -> bool

Return the truth value of this condition for results.

to_dict abstractmethod

to_dict() -> dict[str, Any]

Return a JSON-compatible representation of this condition.

from_dict abstractmethod classmethod

from_dict(data: Mapping[str, Any]) -> 'Condition'

Restore a condition of this kind from its JSON representation.

describe

describe() -> str

Return a short human-readable form, used in error messages.

ExpressionCondition

ExpressionCondition(expression: str | SafeExpression)

Bases: Condition

A condition backed by a sandboxed expression over the results mapping.

Example::

ExpressionCondition("score > 0.8 and not is_empty(draft)")

source property

source: str

Return the original expression source.

condition_from_config

condition_from_config(
    config: Condition | Mapping[str, Any] | str,
) -> Condition

Build a Condition from JSON, a bare expression string, or itself.

A plain string is treated as an expression, so "score > 0.8" and {"kind": "expression", "expression": "score > 0.8"} are equivalent.

register_condition_kind

register_condition_kind(
    condition_cls: type[Condition],
) -> type[Condition]

Register a Condition subclass so JSON can dispatch to it.

Usable as a decorator. The class must set a non-empty kind.

Declaring agents and endpoints

A pipeline document can name the providers it runs against. API keys are never serialized — a config names the environment variable holding one.

ProviderConfig

ProviderConfig dataclass

ProviderConfig(
    provider_type: str,
    model: str,
    base_url: str | None = None,
    api_key_env: str | None = None,
    provider_identifier: str | None = None,
    settings: dict[str, Any] = dict(),
    extra_settings: dict[str, Any] = dict(),
    env_file: str | None = None,
    timeout: float | None = None,
    max_retries: int | None = None,
)

A serializable description of a chat provider endpoint.

Attributes:

Name	Type	Description
provider_type	str	One of :data:PROVIDER_SPECS or an OpenAI-compatible alias such as "openrouter" or "vllm".
model	str	Model identifier passed to the provider.
base_url	str | None	Optional endpoint override. This is what makes an OpenAI-shaped or Anthropic-shaped API reachable at a custom address (a gateway, a proxy, or a self-hosted server).
api_key_env	str | None	Environment variable holding the API key. Defaults to the provider's conventional variable.
provider_identifier	str | None	Optional override for the provider identifier string, where the provider supports one.
settings	dict[str, Any]	Sampling settings applied to the provider's defaults. Names must already exist on the provider, so typos are caught loudly.
extra_settings	dict[str, Any]	Additional request settings to add, for parameters a given endpoint understands but the provider does not declare.
env_file	str | None	Optional .env file read before the API key is looked up. Variables already set in the environment win, so the file supplies a default rather than an override.
timeout	float | None	Seconds to wait for a response before giving up. The SDKs default to 600 with two retries, so a stalled request can hang for half an hour looking like a crash; set this for anything running unattended.
max_retries	int | None	How many times to retry a failed request.

resolve_spec

resolve_spec() -> (
    tuple[ProviderSpec, str | None, str, bool]
)

Return the spec, base URL, API key env var, and key-optional flag.

resolve_api_key

resolve_api_key(
    api_key_env: str, key_optional: bool = False
) -> str

Read the API key from the environment, or fail with a clear message.

build

build() -> Any

Construct and return the configured chat provider.

apply_settings

apply_settings(provider: Any) -> None

Apply configured sampling settings to provider's defaults.

ProviderSettings.__setattr__ silently creates a dead attribute for an unknown name, so settings are applied through set_value and an unknown name is reported rather than quietly ignored.

to_dict

to_dict() -> dict[str, Any]

Return a JSON-compatible representation. Never contains a secret.

from_dict classmethod

from_dict(data: Mapping[str, Any]) -> 'ProviderConfig'

Restore a provider config from its JSON representation.

AgentConfig dataclass

AgentConfig(
    name: str,
    provider: ProviderConfig,
    agent_type: str = "chat_tool_agent",
)

A named agent declared in pipeline JSON.

build

build() -> Any

Construct the configured agent.

to_dict

to_dict() -> dict[str, Any]

Return a JSON-compatible representation.

from_dict classmethod

from_dict(data: Mapping[str, Any]) -> 'AgentConfig'

Restore an agent config from its JSON representation.

ProviderSpec dataclass

ProviderSpec(
    name: str,
    module: str,
    class_name: str,
    default_api_key_env: str,
    supports_base_url: bool = True,
    supports_provider_identifier: bool = False,
)

How to construct one kind of chat provider from configuration.

load_class

load_class() -> type

Import and return the provider class.

LazyAgentRegistry

LazyAgentRegistry(configs: Sequence[AgentConfig])

Bases: Mapping

Build declared agents on first reference, not on load.

Building eagerly means a document that merely declares an unused Anthropic agent fails to load with "ANTHROPIC_API_KEY is not set", even when every process references the OpenAI one. Constructing on demand keeps the failure attached to the agent actually being used.

register_provider_spec

register_provider_spec(spec: ProviderSpec) -> ProviderSpec

Register a provider kind so pipeline JSON can name it.

Sources and sinks

A source loads data into the inputs section; a sink emits a value out of the results. Both are processes, so they nest inside flow control. Writing is gated behind allow_writes — see the sources and sinks guide.

SourceProcess

SourceProcess(
    source: Source | Mapping[str, Any],
    result_key: str = "documents",
    splitter: Mapping[str, Any] | str | None = None,
    section: str = "inputs",
    process_name: str = "SourceProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
)

Bases: Process

Load data into the pipeline.

The loaded value lands in inputs by default, because it is input -- outputs stays for what the model produced.

JSON::

{
  "process_type": "source",
  "process_name": "load",
  "source": {"type": "folder", "path": "./notes", "glob": "*.md"},
  "splitter": {"type": "recursive_character", "chunk_size": 800},
  "result_key": "documents"
}

Initialize a source process.

Parameters:

Name	Type	Description	Default
source	Source | Mapping[str, Any]	Source object or config describing what to load.	required
result_key	str	Key the loaded value is written to.	'documents'
splitter	Mapping[str, Any] | str | None	Optional text splitter config. A source yielding text becomes a list of chunks; a source yielding records becomes more records, each with a chunk_index.	None
section	str	Results section to write into. Defaults to inputs.	'inputs'
process_name	str	Name identifier for the process.	'SourceProcess'
agent	BaseToolAgent	Unused; sources call no model. Accepted so a source can sit anywhere a process can.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None

run_process

run_process(results: PipelineResults) -> PipelineResults

Load the source and write it into the configured section.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this source process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "SourceProcess"

Restore a source process from JSON.

SinkProcess

SinkProcess(
    sink: Sink | Mapping[str, Any] | str,
    source_key: str = "outputs",
    process_name: str = "SinkProcess",
    agent: BaseToolAgent = None,
    agent_name: str | None = None,
    record_as: str | None = None,
    allow_writes: bool = True,
)

Bases: Process

Send a value out of the pipeline.

JSON::

{
  "process_type": "sink",
  "process_name": "save",
  "sink": {"type": "file", "path": "out/{inputs/name}.md"},
  "from": "outputs/draft"
}

Sinks that write a file or make a request are refused unless the pipeline was loaded with allow_writes=True. A sink built in Python is permitted by default; the gate is on loading a document that writes.

Initialize a sink process.

Parameters:

Name	Type	Description	Default
sink	Sink | Mapping[str, Any] | str	Sink object or config describing where the value goes.	required
source_key	str	Results path to read, such as outputs/draft. Defaults to the whole outputs section.	'outputs'
process_name	str	Name identifier for the process.	'SinkProcess'
agent	BaseToolAgent	Unused; sinks call no model.	None
agent_name	str | None	Name of a declared agent, for JSON round-tripping.	None
record_as	str | None	Optional output key recording what the sink returned -- the path written, or the HTTP status.	None
allow_writes	bool	Whether this sink may touch the filesystem or the network. Defaults to true in Python and false when loaded from JSON without allow_writes=True.	True

run_process

run_process(results: PipelineResults) -> PipelineResults

Read the configured value and emit it.

to_dict

to_dict(
    tool_registry: PipelineToolRegistry | None = None,
) -> dict[str, Any]

Serialize this sink process.

from_dict classmethod

from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "SinkProcess"

Restore a sink process from JSON.

Source types

Source

Bases: ABC

Loads data into a pipeline.

load abstractmethod

load(results: PipelineResults) -> Any

Return the loaded value.

to_dict abstractmethod

to_dict() -> dict[str, Any]

Return a JSON-compatible representation.

from_dict abstractmethod classmethod

from_dict(data: Mapping[str, Any]) -> 'Source'

Restore a source of this kind from JSON.

TextSource

TextSource(text: str)

Bases: Source

Inline text, useful for tests and small fixed prompts.

FileSource

FileSource(path: str, encoding: str = 'utf-8')

Bases: Source

One file, read as text.

The path may contain placeholders: {inputs/report_path}.

FilesSource

FilesSource(
    paths: Sequence[str],
    encoding: str = "utf-8",
    max_files: int = DEFAULT_MAX_FILES,
)

Bases: Source

An explicit list of files, loaded as records.

FolderSource

FolderSource(
    path: str,
    glob: str = "*",
    encoding: str = "utf-8",
    max_files: int = DEFAULT_MAX_FILES,
)

Bases: Source

Every file in a folder matching a glob, loaded as records.

Results are sorted by path so a run is reproducible.

register_source_type

register_source_type(
    source_cls: type[Source],
) -> type[Source]

Register a Source subclass so pipeline JSON can name it.

Sink types

Sink

Bases: ABC

Emits a value out of a pipeline.

emit abstractmethod

emit(value: Any, results: PipelineResults) -> Any

Emit value. Returns anything worth recording, or None.

to_dict abstractmethod

to_dict() -> dict[str, Any]

Return a JSON-compatible representation.

from_dict abstractmethod classmethod

from_dict(data: Mapping[str, Any]) -> 'Sink'

Restore a sink of this kind from JSON.

StreamSink

StreamSink(stream: str = 'stdout', prefix: str = '')

Bases: Sink

Print to stdout or stderr. Not gated: printing writes nothing.

FileSink

FileSink(
    path: str,
    mode: str = "write",
    encoding: str = "utf-8",
    create_parents: bool = True,
)

Bases: Sink

Write the value to one file. The path may contain placeholders.

FilesSink

FilesSink(
    path: str,
    content_key: str | None = None,
    encoding: str = "utf-8",
    create_parents: bool = True,
)

Bases: Sink

Write one file per item of a list.

The path template sees {index} and, when the item is a mapping, its own keys — so "out/{name}.md" works over records from a folder source.

HttpSink

HttpSink(
    url: str,
    method: str = "POST",
    headers: Mapping[str, str] | None = None,
    headers_from_env: Mapping[str, str] | None = None,
    as_json: bool = True,
    field: str = "content",
    timeout: float = 30.0,
)

Bases: Sink

Send the value to an HTTP endpoint.

Secrets are never serialized: headers_from_env maps a header name to the environment variable holding its value, the same rule provider configs follow.

resolve_headers

resolve_headers() -> dict[str, str]

Return the headers, reading secret values from the environment.

register_sink_type

register_sink_type(sink_cls: type[Sink]) -> type[Sink]

Register a Sink subclass so pipeline JSON can name it.

Path placeholders

render_path

render_path(
    template: str, results: Mapping[str, Any]
) -> str

Fill {section/key} placeholders in a path or URL.

Unresolved placeholders are an error rather than a silent gap: a path with a hole in it would write to the wrong place, which is exactly the mistake worth being loud about.

normalize_placeholders

normalize_placeholders(template: str) -> str

Repair placeholders whose separator was rewritten by pathlib.

Path("out") / "{vars/index}.md" becomes out\{vars\index}.md on Windows: joining a path normalizes every /, including the one inside the placeholder, which then no longer resolves. Since no section or key name contains a backslash, one inside braces is always this accident. Only the inside of {...} is touched, so real Windows paths around it are left alone.

Chunking

build_splitter

build_splitter(
    config: Mapping[str, Any] | str | None,
) -> Any

Build a text splitter from a config mapping, or a bare type name.

Returns None when config is None, meaning "do not split".

SplitterSpec dataclass

SplitterSpec(
    name: str,
    factory: Callable[..., Any],
    fields: tuple[str, ...] = (),
)

How to construct one kind of text splitter from configuration.

register_splitter_spec

register_splitter_spec(spec: SplitterSpec) -> SplitterSpec

Register a splitter kind so pipeline JSON can name it.

Tools

PipelineToolRegistry

PipelineToolRegistry()

Resolve pipeline JSON tool references to live FunctionTool objects.

register_plugin

register_plugin(
    name: str,
    tools: (
        Iterable[FunctionTool | type[BaseModel]]
        | ToolRegistry
    ),
    source: str | None = None,
) -> "PipelineToolRegistry"

Register tools under a plugin name and optional import source.

load_plugin

load_plugin(
    name: str, source: str
) -> "PipelineToolRegistry"

Import and register tools from a "module:attribute" source.

load_plugins

load_plugins(
    plugin_configs: Iterable[Mapping[str, Any]],
) -> "PipelineToolRegistry"

Load plugin declarations from pipeline JSON metadata.

resolve_tool

resolve_tool(
    reference: Mapping[str, Any] | str,
) -> FunctionTool

Resolve a serialized tool reference.

reference_for_tool

reference_for_tool(tool: FunctionTool) -> dict[str, str]

Return the JSON reference for a registered tool.

to_plugin_configs

to_plugin_configs() -> list[dict[str, str]]

Return JSON-compatible plugin declarations for importable plugins.

get_tool

get_tool(
    plugin_name: str, tool_name: str
) -> FunctionTool | None

Return a registered tool by plugin and public tool name.

get_tools

get_tools() -> list[FunctionTool]

Return all registered tools.

PipelineToolPlugin dataclass

PipelineToolPlugin(
    name: str,
    tools: Sequence[FunctionTool],
    source: str | None = None,
)

A named source of tools that can be referenced from pipeline JSON.

The optional source uses "module:attribute" syntax. The attribute can be an iterable of tools, a ToolRegistry, a Pydantic model class, or a zero-argument factory returning any of those values.

from_spec classmethod

from_spec(name: str, source: str) -> 'PipelineToolPlugin'

Load a plugin from a "module:attribute" tool source.

to_dict

to_dict() -> dict[str, str]

Return a JSON-compatible plugin declaration.

load_pipeline_tools_from_spec

load_pipeline_tools_from_spec(
    spec: str,
) -> list[FunctionTool]

Load tools from "module:attribute" for pipeline plugins.

Loading and extension

PipelineLoadContext dataclass

PipelineLoadContext(
    tool_registry: "PipelineToolRegistry | None" = None,
    default_agent: BaseToolAgent | None = None,
    process_agents: Mapping[str, BaseToolAgent] = dict(),
    step_agents: Mapping[str, BaseToolAgent] = dict(),
    load_tool_plugins: bool = True,
    named_agents: Mapping[str, BaseToolAgent] = dict(),
    json_default_agent_name: str | None = None,
    allow_writes: bool = False,
    ignore_agent_names: bool = False,
    process_path: tuple[str, ...] = (),
    parent_agent: BaseToolAgent | None = None,
)

Everything a process needs in order to rebuild itself from JSON.

Flow-control processes contain other processes, so deserialization is recursive. Rather than thread five separate keyword arguments down every level, the loader carries this single context and hands it to each child.

Agent resolution follows one rule: an agent injected from Python wins over a name declared in JSON at the same level of specificity, and a more specific source wins over a less specific one. In descending priority:

1. step_agents entry for this step
2. the step's own JSON agent name
3. process_agents entry for this process
4. the process's own JSON agent name
5. the agent resolved for an enclosing flow-control process
6. default_agent passed from Python
7. the JSON default_agent name

nested

nested(
    process_name: str, agent: BaseToolAgent | None = None
) -> "PipelineLoadContext"

Return a copy of this context scoped inside process_name.

named_agent

named_agent(
    name: str | None, *, referenced_by: str
) -> BaseToolAgent | None

Resolve a JSON agent reference, or None when name is None.

agent_for_process

agent_for_process(
    process_name: str, agent_name: str | None = None
) -> BaseToolAgent | None

Return the agent a process should use, honouring the priority rule.

agent_for_step

agent_for_step(
    process_name: str,
    step_name: str,
    agent_name: str | None = None,
) -> BaseToolAgent | None

Return the agent a step should use, or None to inherit.

None means "no step-specific agent"; the process-level agent is used at run time instead.

register_process_type

register_process_type(
    process_cls: type["Process"],
) -> type["Process"]

Register a Process subclass so pipeline JSON can dispatch to it.

Usable as a decorator. The class must define a non-empty process_type.

get_process_type

get_process_type(process_type: str) -> type['Process']

Return the registered Process subclass for process_type.

process_from_dict

process_from_dict(
    data: Mapping[str, Any], context: PipelineLoadContext
) -> "Process"

Rebuild a single process from its JSON representation.

Errors

PipelineSerializationError

Bases: ValueError

Raised when a pipeline cannot be serialized or restored.

PipelineExecutionError

Bases: RuntimeError

Raised when a pipeline fails while running, not while loading.

Kept distinct from :class:PipelineSerializationError so that code wrapping from_dict and code wrapping run_pipeline can catch the failures that actually belong to each.

PipelineConditionError

Bases: ValueError

Raised when a condition cannot be compiled or evaluated.

AgentConfigurationError

Bases: ValueError

Raised when an agent or provider cannot be built from configuration.