Python DSL

Building a workflow

from agentloom.core.dsl import workflow

wf = (
    workflow("support-reply", model="gpt-4o-mini", budget_usd=0.10)
    .set_state(message="My order arrived damaged")
    .add_llm_step(
        "classify",
        system_prompt="Reply with one word: billing, technical or general.",
        prompt="Classify: {state.message}",
        output="category",
    )
    .add_llm_step(
        "reply",
        depends_on=["classify"],
        prompt="Write a short {state.category} support reply to: {state.message}",
        output="reply",
        temperature=0.3,
        retry={"max_retries": 2},
    )
    .build()
)

workflow(name, description="", **config) starts a builder. Its keyword arguments are the keys of a file’s config, such as model, budget_usd, timeout, on_step_failure or sandbox. Every method returns the builder, and build() returns the workflow.

.add_router_step(
    "route",
    depends_on=["classify"],
    conditions=[
        ("state.category.strip().lower() == 'billing'", "handle_billing"),
        ("state.category.strip().lower() == 'technical'", "handle_technical"),
    ],
    default="handle_general",
)

• A wrong field fails at once, at the workflow() or add_… call that sets it: an unknown key, a value of the wrong type, tools and response_schema on one step.
• A Pydantic response_schema names its class by path, as in YAML. In a script, "__main__:Answer" names a class defined in that script.

What the builder doesn’t cover

• version can’t be set.
• Only add_llm_step takes extra fields: the other steps get no timeout or retry, and a subworkflow no isolated_state, input or return_keys.

For those, write the workflow as a dictionary, the same shape as the YAML file, and load it with WorkflowParser.from_dict. Or add to a build and validate the whole again:

from agentloom.core.models import WorkflowDefinition

wf = WorkflowDefinition(**{
    **wf.model_dump(),
    "version": "2.0",
    "steps": [*wf.model_dump()["steps"],
              {"id": "ping", "type": "tool", "tool_name": "http_request", "depends_on": ["reply"],
               "retry": {"max_retries": 0}}],
})

Validating a build

build() checks each field, that step ids are unique, that dependencies, router targets and subworkflow files exist, that there are no cycles, and that steps of one layer don’t share a conversation. It raises ValidationError. Only parallel steps writing one output key aren’t checked.

To get that warning too, run the checks that agentloom validate runs:

from agentloom.core.parser import WorkflowParser

WorkflowParser.from_dict(wf.model_dump())   # raises ValidationError, warns on shared output keys

Loading YAML

from agentloom.core.parser import WorkflowParser

wf = WorkflowParser.from_yaml("pipeline.yaml")
wf = WorkflowParser.from_dict({"name": "inline", "steps": [...]})

Both run every check that agentloom validate runs, and raise agentloom.exceptions.ValidationError.

from_yaml takes a path or YAML text. A string with no file behind it is read as YAML, unless it looks like a path, one line without a :, which fails with Workflow file not found. A Path is always a path. Text too long to be a file name is read as YAML.

Running a workflow

import anyio

from agentloom.core.engine import WorkflowEngine
from agentloom.providers.gateway import ProviderGateway
from agentloom.providers.openai import OpenAIProvider

async def main() -> None:
    gateway = ProviderGateway()
    gateway.register(OpenAIProvider(), priority=0)   # reads OPENAI_API_KEY
    try:
        result = await WorkflowEngine(wf, provider_gateway=gateway).run()
    finally:
        await gateway.close()
    print(result.status, result.final_state["reply"])

anyio.run(main)

run() is asynchronous: call it from a coroutine, with anyio.run or asyncio.run.

• The gateway decides which providers answer. From Python, config.provider has no effect. config.model is still the model of every step that doesn’t name its own.
• A run that doesn’t succeed returns a result, with its status, as it does from the CLI. run() raises only for a cycle in a workflow that wasn’t validated, an observer that lacks a hook, or a cancellation.
• Every run writes a record to agentloom_runs/<run_id>.json, in the current directory, or under AGENTLOOM_RUNS_DIR. See Observability.
• The engine doesn’t close the providers: call await gateway.close() when done.

The provider gateway

gateway.register(provider, priority=0, is_fallback=False, models=None,
                 max_rpm=60, max_tpm=100_000,
                 circuit_fail_threshold=5, circuit_reset_timeout=60.0,
                 is_primary=False, model_aliases=None)

• A model goes to the providers that list it in models, or, if no provider lists it, to every provider that serves it by name; then to those registered with is_fallback=True. priority orders each group, lowest first.
• is_primary=True puts the provider ahead of the rest for any model it can serve. model_aliases maps a requested model name to this provider’s own, and makes it a candidate for the requested one.
• A streamed call waits for the first chunk inside gateway.stream(), so a provider that fails before sending one is replaced by the next. Iterate the stream it returns, as an abandoned one holds its connection open.
• max_rpm and max_tpm set the provider’s rate limits, and circuit_fail_threshold and circuit_reset_timeout its circuit breaker. See Provider failures.

Without an api_key, OpenAI, Anthropic and Google read OPENAI_API_KEY, ANTHROPIC_API_KEY and GOOGLE_API_KEY, and Ollama reads its address from OLLAMA_BASE_URL. The other *_BASE_URL variables apply only to the CLI: pass base_url instead. Providers covers each one.

Results

run() returns a WorkflowResult:

Redacted keys are redacted in final_state and step_results too.

As a function

WorkflowAdapter(workflow, provider_gateway, tool_registry=None, observer=None), from agentloom.core.handle, wraps a workflow as a call: await adapter.run(inputs) starts from the workflow’s state with inputs on top, and returns the final state, or raises WorkflowError when the run doesn’t succeed. It takes no checkpointer.

The state from Python

StateManager holds a run’s state. From a step or a tool of your own:

• await state.get("user.profile.email") and await state.set("report.summary", text) read and write dotted paths, by the rules of Writing state.
• await state.update("counter", lambda n: (n or 0) + 1) reads and writes under one lock, and returns the new value. A get followed by a set doesn’t: fifty parallel increments written that way leave 1, while fifty update calls leave 50.
• await state.get_state_snapshot() returns a copy of the whole state.

A write that the rules refuse raises StateWriteError, from agentloom.exceptions. When it comes from a step’s output, the step fails at once, without retries: the same write would be refused again.

Imports

agentloom itself exports Agent, AgentResult, WorkflowGraph, GraphNode and GraphEdge. The rest comes from its modules:

agentloom.contracts.stable follows semantic versioning: Attachment, Message, ProviderResponse, ResponseMeta, TokenUsage, ToolCall, ToolDefinition and WorkflowHandle change only in a major version. agentloom.contracts.experimental warns on import that it may change. No other module makes that promise.

What’s next

• Graph API

  Read, analyse and export a workflow’s DAG.

• Tools & sandbox

  Tools of your own, registered from Python.

• Testing & replay

  Running workflows offline against recordings.