windlass.agent

agent

Agents — models that use tools to accomplish goals.

Start with :func:windlass.Windlass.agent, which returns an :class:~windlass.agent.builder.AgentBuilder::

from windlass import Windlass, tool

@tool
def search(query: str) -> list[str]:
    '''Search the knowledge base.'''
    return kb.search(query)

agent = Windlass.agent().llm("openai").tool(search).memory()
print(agent.run("What changed in the API last quarter?"))

The pieces:

• :class:~windlass.agent.builder.AgentBuilder — the fluent API.
• :class:~windlass.agent.runtime.AgentRuntime — the built-in reason/act loop, with no dependencies.
• :class:~windlass.agent.graph.LangGraphRuntime — LangGraph-backed execution, for conditional routing and subgraphs.
• :class:~windlass.agent.supervisor.Supervisor — multi-agent delegation.
• :mod:~windlass.agent.checkpoint — durable state for resume and human-in-the-loop.

AgentBuilder

AgentBuilder(container: Container | None = None)

Fluent builder for an agent.

Parameters:

Name	Type	Description	Default
container	Container | None	Dependency container. Defaults to a child of the process root.	None

Example

Level 1 — a working agent in two lines, no dependencies::

agent = Windlass.agent().llm("fake", responses=["Hello!"])
print(agent.run("Say hello"))

Level 2 — a real agent::

agent = (Windlass.agent()
         .llm("anthropic", model="claude-sonnet-4-5")
         .tool(search).tool(calculator)
         .system("You are a research assistant. Cite your sources.")
         .memory("summary")
         .guardrails(pii=True, on_violation="redact")
         .checkpoint("sqlite", path="./state.db")
         .max_iterations(15)
         .observe("langfuse"))

Level 3 — a graph you control::

agent = Windlass.agent().llm("openai").tool(search).graph()
graph = agent.native_graph()
graph.add_node("critic", critic_fn)
graph.add_edge("tools", "critic")
agent.recompile()

Source code in src\windlass\agent\builder.py

def __init__(self, container: Container | None = None) -> None:
    self.container = container or root_container().scope()
    self._specs: dict[str, tuple[Any, dict[str, Any]]] = {}
    self._tools: list[Any] = []
    self._mcp_specs: list[tuple[Any, dict[str, Any]]] = []
    self._sub_agents: dict[str, Any] = {}
    self._descriptions: dict[str, str] = {}
    self._options: dict[str, Any] = {
        "system_prompt": None,
        "max_iterations": 10,
        "parallel_tools": True,
        "require_approval": False,
        "max_tool_call_retries": 2,
        "name": "agent",
    }
    self._use_graph = False
    self._runtime: AgentRuntime | None = None

runtime property

runtime: AgentRuntime

The built runtime, constructing it on first access.

llm

llm(spec: Any = None, /, **config: Any) -> Self

Choose the reasoning model.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("openai", "anthropic", "gemini", "groq", "ollama", "fake"), an :class:~windlass.interfaces.llm.LLM, or a factory. A bare model id like "gpt-4o" is also accepted and mapped to its provider.	None
**config	Any	Provider options — model, temperature, api_key, ...	{}

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = Windlass.agent().llm("gpt-4o-mini", temperature=0)

Source code in src\windlass\agent\builder.py

def llm(self, spec: Any = None, /, **config: Any) -> Self:
    """Choose the reasoning model.

    Args:
        spec: Registry name (``"openai"``, ``"anthropic"``, ``"gemini"``,
            ``"groq"``, ``"ollama"``, ``"fake"``), an
            :class:`~windlass.interfaces.llm.LLM`, or a factory. A bare model
            id like ``"gpt-4o"`` is also accepted and mapped to its provider.
        **config: Provider options — ``model``, ``temperature``, ``api_key``, ...

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = Windlass.agent().llm("gpt-4o-mini", temperature=0)
    """
    if isinstance(spec, str):
        provider, model = _split_model(spec)
        if model and "model" not in config:
            config = {**config, "model": model}
        spec = provider
    return self._set("llm", spec, config)

tool

tool(*tools: Any) -> Self

Bind one or more tools.

Parameters:

Name	Type	Description	Default
*tools	Any	:class:~windlass.interfaces.tool.Tool instances, functions decorated with :func:~windlass.tools.tool, plain functions (wrapped automatically), or registry names.	()

Returns:

Type	Description
Self	self.

Raises:

Type	Description
ConfigurationError	When an argument cannot be interpreted as a tool.

Example

from windlass import Windlass, tool @tool ... def now() -> str: ... '''Return the current time.''' ... return "12:00" _ = Windlass.agent().tool(now)

Source code in src\windlass\agent\builder.py

def tool(self, *tools: Any) -> Self:
    """Bind one or more tools.

    Args:
        *tools: :class:`~windlass.interfaces.tool.Tool` instances, functions
            decorated with :func:`~windlass.tools.tool`, plain functions
            (wrapped automatically), or registry names.

    Returns:
        ``self``.

    Raises:
        ConfigurationError: When an argument cannot be interpreted as a tool.

    Example:
        >>> from windlass import Windlass, tool
        >>> @tool
        ... def now() -> str:
        ...     '''Return the current time.'''
        ...     return "12:00"
        >>> _ = Windlass.agent().tool(now)
    """
    from windlass.interfaces.tool import Tool

    for item in tools:
        if item is None:
            continue
        # A Tool subclass is not necessarily callable — only FunctionTool
        # forwards __call__ — so the isinstance check has to come first.
        if not (isinstance(item, Tool | str) or callable(item)):
            raise ConfigurationError(
                f"Cannot use {type(item).__name__} as a tool.",
                hint="Pass a @tool-decorated function, a Tool instance, a plain "
                "callable, or a registered tool name.",
            )
        self._tools.append(item)
    return self._invalidate()

tools

tools(tools: Sequence[Any]) -> Self

Bind a sequence of tools. See :meth:tool.

Source code in src\windlass\agent\builder.py

def tools(self, tools: Sequence[Any]) -> Self:
    """Bind a sequence of tools. See :meth:`tool`."""
    return self.tool(*tools)

mcp

mcp(spec: Any = None, /, **config: Any) -> Self

Connect an MCP server and bind its tools.

Call it repeatedly to connect several servers; their tools are namespaced by server name so identically-named tools do not collide.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("fastmcp", "static"), an :class:~windlass.interfaces.mcp.MCPClient, or a factory. Omit it for the FastMCP client.	None
**config	Any	Transport options — command, args, url, env, server.	{}

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = Windlass.agent().mcp(server="fs", command="npx", ... args=["-y", "@modelcontextprotocol/server-filesystem", "."])

Source code in src\windlass\agent\builder.py

def mcp(self, spec: Any = None, /, **config: Any) -> Self:
    """Connect an MCP server and bind its tools.

    Call it repeatedly to connect several servers; their tools are namespaced
    by server name so identically-named tools do not collide.

    Args:
        spec: Registry name (``"fastmcp"``, ``"static"``), an
            :class:`~windlass.interfaces.mcp.MCPClient`, or a factory. Omit it
            for the FastMCP client.
        **config: Transport options — ``command``, ``args``, ``url``, ``env``,
            ``server``.

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = Windlass.agent().mcp(server="fs", command="npx",
        ...     args=["-y", "@modelcontextprotocol/server-filesystem", "."])
    """
    self._mcp_specs.append((spec if spec is not None else "fastmcp", config))
    return self._invalidate()

memory

memory(spec: Any = None, /, **config: Any) -> Self

Attach conversation memory.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("window", "buffer", "summary", "vector", "composite"), a :class:~windlass.interfaces.memory.Memory, or a factory. Omit it for a sliding window.	None
**config	Any	Memory options.	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\agent\builder.py

def memory(self, spec: Any = None, /, **config: Any) -> Self:
    """Attach conversation memory.

    Args:
        spec: Registry name (``"window"``, ``"buffer"``, ``"summary"``,
            ``"vector"``, ``"composite"``), a
            :class:`~windlass.interfaces.memory.Memory`, or a factory. Omit it
            for a sliding window.
        **config: Memory options.

    Returns:
        ``self``.
    """
    return self._set("memory", spec if spec is not None else "window", config)

guardrails

guardrails(spec: Any = None, /, **config: Any) -> Self

Enable input and output guardrails.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("rules", "nemo"), a :class:~windlass.interfaces.guardrail.Guardrail, or a factory. Omit it for the dependency-free rule guardrail.	None
**config	Any	Policy options.	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\agent\builder.py

def guardrails(self, spec: Any = None, /, **config: Any) -> Self:
    """Enable input and output guardrails.

    Args:
        spec: Registry name (``"rules"``, ``"nemo"``), a
            :class:`~windlass.interfaces.guardrail.Guardrail`, or a factory.
            Omit it for the dependency-free rule guardrail.
        **config: Policy options.

    Returns:
        ``self``.
    """
    return self._set("guardrail", spec if spec is not None else "rules", config)

observe

observe(spec: Any = None, /, **config: Any) -> Self

Enable tracing.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("console", "langfuse", "langsmith", "memory"), a :class:~windlass.interfaces.tracer.Tracer, or a factory.	None
**config	Any	Tracer options.	{}

Returns:

Type	Description
Self	self.

Source code in src\windlass\agent\builder.py

def observe(self, spec: Any = None, /, **config: Any) -> Self:
    """Enable tracing.

    Args:
        spec: Registry name (``"console"``, ``"langfuse"``, ``"langsmith"``,
            ``"memory"``), a :class:`~windlass.interfaces.tracer.Tracer`, or a
            factory.
        **config: Tracer options.

    Returns:
        ``self``.
    """
    return self._set("tracer", spec if spec is not None else "console", config)

checkpoint

checkpoint(spec: Any = None, /, **config: Any) -> Self

Enable durable state, resume and human-in-the-loop approval.

Parameters:

Name	Type	Description	Default
spec	Any	Registry name ("memory", "sqlite"), a :class:~windlass.agent.checkpoint.Checkpointer, or a factory. Omit it for the in-process store.	None
**config	Any	Checkpointer options — path, max_history.	{}

Returns:

Type	Description
Self	self.

Note

Approval interrupts require a checkpointer, since a paused run has to be stored somewhere before it can be resumed.

Source code in src\windlass\agent\builder.py

def checkpoint(self, spec: Any = None, /, **config: Any) -> Self:
    """Enable durable state, resume and human-in-the-loop approval.

    Args:
        spec: Registry name (``"memory"``, ``"sqlite"``), a
            :class:`~windlass.agent.checkpoint.Checkpointer`, or a factory.
            Omit it for the in-process store.
        **config: Checkpointer options — ``path``, ``max_history``.

    Returns:
        ``self``.

    Note:
        Approval interrupts require a checkpointer, since a paused run has to
        be stored somewhere before it can be resumed.
    """
    return self._set("checkpointer", spec if spec is not None else "memory", config)

agent

agent(name: str, worker: Any, description: str = '') -> Self

Add a specialist, turning this into a supervisor.

Parameters:

Name	Type	Description	Default
name	str	The name the supervisor uses to delegate.	required
worker	Any	An agent, builder, or anything with arun.	required
description	str	What the specialist is good at. The supervisor routes on this text.	''

Returns:

Type	Description
Self	self.

Example

from windlass import Windlass _ = (Windlass.agent().llm("fake") ... .agent("researcher", Windlass.agent().llm("fake"), ... "Finds and summarises source material."))

Source code in src\windlass\agent\builder.py

def agent(self, name: str, worker: Any, description: str = "") -> Self:
    """Add a specialist, turning this into a supervisor.

    Args:
        name: The name the supervisor uses to delegate.
        worker: An agent, builder, or anything with ``arun``.
        description: What the specialist is good at. The supervisor routes on
            this text.

    Returns:
        ``self``.

    Example:
        >>> from windlass import Windlass
        >>> _ = (Windlass.agent().llm("fake")
        ...      .agent("researcher", Windlass.agent().llm("fake"),
        ...             "Finds and summarises source material."))
    """
    self._sub_agents[name] = worker
    if description:
        self._descriptions[name] = description
    return self._invalidate()

system

system(prompt: str) -> Self

Set the system prompt.

Parameters:

Name	Type	Description	Default
prompt	str	Instructions prepended to every run.	required

Returns:

Type	Description
Self	self.

Source code in src\windlass\agent\builder.py

def system(self, prompt: str) -> Self:
    """Set the system prompt.

    Args:
        prompt: Instructions prepended to every run.

    Returns:
        ``self``.
    """
    self._options["system_prompt"] = prompt
    return self._invalidate()

name

name(value: str) -> Self

Name the agent, for traces and multi-agent routing.

Source code in src\windlass\agent\builder.py

def name(self, value: str) -> Self:
    """Name the agent, for traces and multi-agent routing."""
    self._options["name"] = value
    return self._invalidate()

max_iterations

max_iterations(limit: int) -> Self

Set the reason/act step budget.

Parameters:

Name	Type	Description	Default
limit	int	Maximum cycles before :class:~windlass.core.exceptions.MaxIterationsExceeded.	required

Returns:

Type	Description
Self	self.

Raises:

Type	Description
ValueError	If limit is not positive.

Source code in src\windlass\agent\builder.py

def max_iterations(self, limit: int) -> Self:
    """Set the reason/act step budget.

    Args:
        limit: Maximum cycles before
            :class:`~windlass.core.exceptions.MaxIterationsExceeded`.

    Returns:
        ``self``.

    Raises:
        ValueError: If ``limit`` is not positive.
    """
    if limit <= 0:
        raise ValueError("max_iterations must be positive")
    self._options["max_iterations"] = limit
    return self._invalidate()

tool_call_retries

tool_call_retries(limit: int) -> Self

Set how often a run may recover from an unparseable tool call.

Models occasionally emit a tool call the provider cannot parse — most often by nesting one call inside another's arguments, which the tool-calling protocol has no way to express. Windlass shows the model the problem and lets it retry, the same way it handles an invented tool name.

Parameters:

Name	Type	Description	Default
limit	int	Maximum corrections per run. Each one costs an iteration from :meth:max_iterations, so a model that never recovers still terminates. 0 fails on the first malformed call.	required

Returns:

Type	Description
Self	self.

Raises:

Type	Description
ValueError	If limit is negative.

Example

from windlass import Windlass _ = Windlass.agent().llm("fake").tool_call_retries(3)

Source code in src\windlass\agent\builder.py

def tool_call_retries(self, limit: int) -> Self:
    """Set how often a run may recover from an unparseable tool call.

    Models occasionally emit a tool call the provider cannot parse — most
    often by nesting one call inside another's arguments, which the
    tool-calling protocol has no way to express. Windlass shows the model the
    problem and lets it retry, the same way it handles an invented tool name.

    Args:
        limit: Maximum corrections per run. Each one costs an iteration from
            :meth:`max_iterations`, so a model that never recovers still
            terminates. ``0`` fails on the first malformed call.

    Returns:
        ``self``.

    Raises:
        ValueError: If ``limit`` is negative.

    Example:
        >>> from windlass import Windlass
        >>> _ = Windlass.agent().llm("fake").tool_call_retries(3)
    """
    if limit < 0:
        raise ValueError("tool_call_retries must not be negative")
    self._options["max_tool_call_retries"] = limit
    return self._invalidate()

parallel_tools

parallel_tools(enabled: bool = True) -> Self

Control whether simultaneous tool calls run concurrently.

Parameters:

Name	Type	Description	Default
enabled	bool	False forces serial execution — occasionally necessary when tools share mutable state.	True

Returns:

Type	Description
Self	self.

Source code in src\windlass\agent\builder.py

def parallel_tools(self, enabled: bool = True) -> Self:
    """Control whether simultaneous tool calls run concurrently.

    Args:
        enabled: False forces serial execution — occasionally necessary when
            tools share mutable state.

    Returns:
        ``self``.
    """
    self._options["parallel_tools"] = enabled
    return self._invalidate()

human_in_the_loop

human_in_the_loop(enabled: bool = True) -> Self

Require human approval before every tool call.

Individual tools can require approval on their own with @tool(requires_approval=True); this turns it on globally.

Parameters:

Name	Type	Description	Default
enabled	bool	True to pause before every tool call.	True

Returns:

Type	Description
Self	self.

Note

Implies a checkpointer, since a paused run has to be stored. One is enabled automatically if you have not chosen one.

Source code in src\windlass\agent\builder.py

def human_in_the_loop(self, enabled: bool = True) -> Self:
    """Require human approval before *every* tool call.

    Individual tools can require approval on their own with
    ``@tool(requires_approval=True)``; this turns it on globally.

    Args:
        enabled: True to pause before every tool call.

    Returns:
        ``self``.

    Note:
        Implies a checkpointer, since a paused run has to be stored. One is
        enabled automatically if you have not chosen one.
    """
    self._options["require_approval"] = enabled
    if enabled and "checkpointer" not in self._specs:
        self.checkpoint()
    return self._invalidate()

graph

graph(enabled: bool = True) -> Self

Use the LangGraph runtime instead of the built-in loop.

Parameters:

Name	Type	Description	Default
enabled	bool	True to compile the agent into a LangGraph state machine.	True

Returns:

Type	Description
Self	self.

Raises:

Type	Description
MissingDependencyError	At build time, when langgraph is missing.

Source code in src\windlass\agent\builder.py

def graph(self, enabled: bool = True) -> Self:
    """Use the LangGraph runtime instead of the built-in loop.

    Args:
        enabled: True to compile the agent into a LangGraph state machine.

    Returns:
        ``self``.

    Raises:
        MissingDependencyError: At build time, when ``langgraph`` is missing.
    """
    self._use_graph = enabled
    return self._invalidate()

bind

bind(key: str, factory: Any) -> Self

Bind an arbitrary dependency into this builder's container.

Source code in src\windlass\agent\builder.py

def bind(self, key: str, factory: Any) -> Self:
    """Bind an arbitrary dependency into this builder's container."""
    if callable(factory):
        self.container.bind(key, factory)
    else:
        self.container.bind_instance(key, factory)
    return self._invalidate()

build

build() -> AgentRuntime

Resolve every component and construct the runtime.

Called automatically on first use.

Returns:

Name	Type	Description
An	AgentRuntime	class:~windlass.agent.runtime.AgentRuntime, a
	AgentRuntime	class:~windlass.agent.graph.LangGraphRuntime, or a
	AgentRuntime	class:~windlass.agent.supervisor.Supervisor — whichever the
	AgentRuntime	configuration describes.

Raises:

Type	Description
ConfigurationError	When a component cannot be constructed.
MissingDependencyError	When a chosen provider's extra is missing.

Source code in src\windlass\agent\builder.py

def build(self) -> AgentRuntime:
    """Resolve every component and construct the runtime.

    Called automatically on first use.

    Returns:
        An :class:`~windlass.agent.runtime.AgentRuntime`, a
        :class:`~windlass.agent.graph.LangGraphRuntime`, or a
        :class:`~windlass.agent.supervisor.Supervisor` — whichever the
        configuration describes.

    Raises:
        ConfigurationError: When a component cannot be constructed.
        MissingDependencyError: When a chosen provider's extra is missing.
    """
    if self._runtime is not None:
        return self._runtime

    from windlass.core.config import settings

    cfg = settings()
    llm = self._resolve("llm", cfg.default_llm)
    registry = ToolRegistry()

    for item in self._tools:
        registry.add(self.container.component("tool", item) if isinstance(item, str) else item)

    for spec, config in self._mcp_specs:
        for remote in self._connect_mcp(spec, config):
            registry.add(remote)

    common: dict[str, Any] = {
        "llm": llm,
        "memory": self._resolve_memory(),
        "guardrail": self._resolve_optional("guardrail"),
        "checkpointer": self._resolve_optional("checkpointer"),
        "tracer": self._resolve_optional("tracer"),
        **self._options,
    }

    if self._sub_agents:
        from windlass.agent.supervisor import Supervisor

        self._runtime = Supervisor(
            agents={k: _built(v) for k, v in self._sub_agents.items()},
            descriptions=self._descriptions,
            tools=list(registry),
            **common,
        )
    elif self._use_graph:
        from windlass.agent.graph import LangGraphRuntime

        self._runtime = LangGraphRuntime(tools=registry, **common)
    else:
        self._runtime = AgentRuntime(tools=registry, **common)

    return self._runtime

run

run(prompt: Any, **kwargs: Any) -> AgentResponse

Run the agent. See :meth:~windlass.agent.runtime.AgentRuntime.run.

Source code in src\windlass\agent\builder.py

def run(self, prompt: Any, **kwargs: Any) -> AgentResponse:
    """Run the agent. See :meth:`~windlass.agent.runtime.AgentRuntime.run`."""
    return self.build().run(prompt, **kwargs)

arun async

arun(prompt: Any, **kwargs: Any) -> AgentResponse

Async :meth:run.

Source code in src\windlass\agent\builder.py

async def arun(self, prompt: Any, **kwargs: Any) -> AgentResponse:
    """Async :meth:`run`."""
    return await self.build().arun(prompt, **kwargs)

stream

stream(prompt: Any, **kwargs: Any) -> Iterator[StreamEvent]

Stream the run. See :meth:~windlass.agent.runtime.AgentRuntime.stream.

Source code in src\windlass\agent\builder.py

def stream(self, prompt: Any, **kwargs: Any) -> Iterator[StreamEvent]:
    """Stream the run. See :meth:`~windlass.agent.runtime.AgentRuntime.stream`."""
    return self.build().stream(prompt, **kwargs)

astream

astream(prompt: Any, **kwargs: Any) -> AsyncIterator[StreamEvent]

Async :meth:stream.

Source code in src\windlass\agent\builder.py

def astream(self, prompt: Any, **kwargs: Any) -> AsyncIterator[StreamEvent]:
    """Async :meth:`stream`."""
    return self.build().astream(prompt, **kwargs)

astream_text

astream_text(prompt: Any, **kwargs: Any) -> AsyncIterator[str]

Stream only the assistant's text.

Source code in src\windlass\agent\builder.py

def astream_text(self, prompt: Any, **kwargs: Any) -> AsyncIterator[str]:
    """Stream only the assistant's text."""
    return self.build().astream_text(prompt, **kwargs)

resume

resume(thread_id: str, **kwargs: Any) -> AgentResponse

Resume an interrupted run.

Source code in src\windlass\agent\builder.py

def resume(self, thread_id: str, **kwargs: Any) -> AgentResponse:
    """Resume an interrupted run."""
    return self.build().resume(thread_id, **kwargs)

aresume async

aresume(thread_id: str, **kwargs: Any) -> AgentResponse

Async :meth:resume.

Source code in src\windlass\agent\builder.py

async def aresume(self, thread_id: str, **kwargs: Any) -> AgentResponse:
    """Async :meth:`resume`."""
    return await self.build().aresume(thread_id, **kwargs)

pending_approvals

pending_approvals(thread_id: str) -> list[Any]

Return the tool calls awaiting approval on a thread.

Source code in src\windlass\agent\builder.py

def pending_approvals(self, thread_id: str) -> list[Any]:
    """Return the tool calls awaiting approval on a thread."""
    return self.build().pending_approvals(thread_id)

state

state(thread_id: str) -> dict[str, Any] | None

Return the latest checkpoint for a thread.

Source code in src\windlass\agent\builder.py

def state(self, thread_id: str) -> dict[str, Any] | None:
    """Return the latest checkpoint for a thread."""
    return self.build().state(thread_id)

history

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent checkpoints for a thread.

Source code in src\windlass\agent\builder.py

def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent checkpoints for a thread."""
    return self.build().history(thread_id, limit)

broadcast

broadcast(task: str) -> dict[str, AgentResponse]

Run every specialist on the same task (supervisor only).

Raises:

Type	Description
ConfigurationError	When no specialists are configured.

Source code in src\windlass\agent\builder.py

def broadcast(self, task: str) -> dict[str, AgentResponse]:
    """Run every specialist on the same task (supervisor only).

    Raises:
        ConfigurationError: When no specialists are configured.
    """
    runtime = self.build()
    caller = getattr(runtime, "broadcast", None)
    if caller is None:
        raise ConfigurationError(
            "broadcast() needs specialist agents.",
            hint="Add them with .agent('name', worker).",
        )
    return caller(task)

pipeline

pipeline(task: str, order: list[str]) -> AgentResponse

Chain specialists in sequence (supervisor only).

Raises:

Type	Description
ConfigurationError	When no specialists are configured.

Source code in src\windlass\agent\builder.py

def pipeline(self, task: str, order: list[str]) -> AgentResponse:
    """Chain specialists in sequence (supervisor only).

    Raises:
        ConfigurationError: When no specialists are configured.
    """
    runtime = self.build()
    caller = getattr(runtime, "pipeline", None)
    if caller is None:
        raise ConfigurationError(
            "pipeline() needs specialist agents.",
            hint="Add them with .agent('name', worker).",
        )
    return caller(task, order)

native_llm

native_llm() -> Any

Return the model provider's own client.

Source code in src\windlass\agent\builder.py

def native_llm(self) -> Any:
    """Return the model provider's own client."""
    return self.build().native_llm()

native_graph

native_graph() -> Any

Return the LangGraph StateGraph.

Raises:

Type	Description
AgentError	When the agent was not built with :meth:graph.

Source code in src\windlass\agent\builder.py

def native_graph(self) -> Any:
    """Return the LangGraph ``StateGraph``.

    Raises:
        AgentError: When the agent was not built with :meth:`graph`.
    """
    return self.build().native_graph()

recompile

recompile() -> Any

Recompile the graph after editing it.

Raises:

Type	Description
ConfigurationError	When the agent is not graph-backed.

Source code in src\windlass\agent\builder.py

def recompile(self) -> Any:
    """Recompile the graph after editing it.

    Raises:
        ConfigurationError: When the agent is not graph-backed.
    """
    runtime = self.build()
    recompiler = getattr(runtime, "recompile", None)
    if recompiler is None:
        raise ConfigurationError(
            "recompile() only applies to graph-backed agents.",
            hint="Build the agent with .graph().",
        )
    return recompiler()

describe

describe() -> dict[str, Any]

Return a JSON-safe description of the agent.

Source code in src\windlass\agent\builder.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe description of the agent."""
    return self.build().describe()

close

close() -> None

Release the model's and MCP clients' resources.

Source code in src\windlass\agent\builder.py

def close(self) -> None:
    """Release the model's and MCP clients' resources."""
    if self._runtime is not None:
        self._runtime.close()

Checkpointer

Bases: ABC

Stores and retrieves agent state by thread.

Implementations must be safe to use from multiple threads, since one agent instance typically serves many concurrent conversations.

put abstractmethod

put(thread_id: str, state: dict[str, Any]) -> None

Save a state snapshot.

Parameters:

Name	Type	Description	Default
thread_id	str	Conversation / run identifier.	required
state	dict[str, Any]	JSON-serialisable state.	required

Raises:

Type	Description
SerializationError	When the state cannot be stored.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def put(self, thread_id: str, state: dict[str, Any]) -> None:
    """Save a state snapshot.

    Args:
        thread_id: Conversation / run identifier.
        state: JSON-serialisable state.

    Raises:
        SerializationError: When the state cannot be stored.
    """

get abstractmethod

get(thread_id: str) -> dict[str, Any] | None

Return the latest snapshot for a thread, or None.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def get(self, thread_id: str) -> dict[str, Any] | None:
    """Return the latest snapshot for a thread, or ``None``."""

history abstractmethod

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent snapshots, newest first.

Parameters:

Name	Type	Description	Default
thread_id	str	Conversation identifier.	required
limit	int	Maximum snapshots to return.	20

Returns:

Type	Description
list[dict[str, Any]]	Snapshots, newest first.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent snapshots, newest first.

    Args:
        thread_id: Conversation identifier.
        limit: Maximum snapshots to return.

    Returns:
        Snapshots, newest first.
    """

delete abstractmethod

delete(thread_id: str) -> None

Discard every snapshot for a thread.

Source code in src\windlass\agent\checkpoint.py

@abc.abstractmethod
def delete(self, thread_id: str) -> None:
    """Discard every snapshot for a thread."""

threads

threads() -> list[str]

Return the known thread ids.

Source code in src\windlass\agent\checkpoint.py

def threads(self) -> list[str]:
    """Return the known thread ids."""
    return []

MemoryCheckpointer

MemoryCheckpointer(max_history: int = 50)

Bases: Checkpointer

Keeps checkpoints in a dict.

Fast and dependency-free, but per-process and lost on restart. Right for tests, notebooks and single-process services; wrong for anything that needs a resumed run to survive a deploy.

Parameters:

Name	Type	Description	Default
max_history	int	Snapshots retained per thread. Older ones are dropped.	50

Example

saver = MemoryCheckpointer() saver.put("t", {"step": 1}) saver.put("t", {"step": 2}) saver.get("t")["step"], len(saver.history("t")) (2, 2)

Source code in src\windlass\agent\checkpoint.py

def __init__(self, max_history: int = 50) -> None:
    self.max_history = max_history
    self._data: dict[str, list[dict[str, Any]]] = {}
    self._lock = threading.RLock()

put

put(thread_id: str, state: dict[str, Any]) -> None

Append a snapshot, trimming the oldest beyond max_history.

Source code in src\windlass\agent\checkpoint.py

def put(self, thread_id: str, state: dict[str, Any]) -> None:
    """Append a snapshot, trimming the oldest beyond ``max_history``."""
    with self._lock:
        bucket = self._data.setdefault(thread_id, [])
        bucket.append({**state, "_saved_at": time.time()})
        if len(bucket) > self.max_history:
            del bucket[: len(bucket) - self.max_history]

get

get(thread_id: str) -> dict[str, Any] | None

Return the newest snapshot, or None.

Source code in src\windlass\agent\checkpoint.py

def get(self, thread_id: str) -> dict[str, Any] | None:
    """Return the newest snapshot, or ``None``."""
    with self._lock:
        bucket = self._data.get(thread_id)
        return dict(bucket[-1]) if bucket else None

history

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent snapshots, newest first.

Source code in src\windlass\agent\checkpoint.py

def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent snapshots, newest first."""
    with self._lock:
        bucket = self._data.get(thread_id, [])
        return [dict(s) for s in reversed(bucket[-limit:])]

delete

delete(thread_id: str) -> None

Discard a thread's snapshots.

Source code in src\windlass\agent\checkpoint.py

def delete(self, thread_id: str) -> None:
    """Discard a thread's snapshots."""
    with self._lock:
        self._data.pop(thread_id, None)

threads

threads() -> list[str]

Return the known thread ids.

Source code in src\windlass\agent\checkpoint.py

def threads(self) -> list[str]:
    """Return the known thread ids."""
    with self._lock:
        return sorted(self._data)

SQLiteCheckpointer

SQLiteCheckpointer(
    path: str | Path = ".windlass/checkpoints.db", max_history: int = 50
)

Bases: Checkpointer

Persists checkpoints to a SQLite database.

Survives restarts and is shared between processes on one machine — enough for a single-node deployment, and it needs nothing beyond the standard library.

Parameters:

Name	Type	Description	Default
path	str | Path	Database file. Parent directories are created. ":memory:" gives a transient database.	'.windlass/checkpoints.db'
max_history	int	Snapshots retained per thread.	50

Raises:

Type	Description
SerializationError	When the database cannot be opened.

Note

Each call opens its own connection. That costs a little per write and buys thread safety without a connection pool — the right trade at the rate an agent checkpoints.

Example

import tempfile, pathlib db = pathlib.Path(tempfile.mkdtemp()) / "state.db" saver = SQLiteCheckpointer(db) saver.put("t", {"step": 7}) SQLiteCheckpointer(db).get("t")["step"] 7

Source code in src\windlass\agent\checkpoint.py

def __init__(
    self, path: str | Path = ".windlass/checkpoints.db", max_history: int = 50
) -> None:
    self.path = Path(path) if str(path) != ":memory:" else Path(":memory:")
    self.max_history = max_history
    self._lock = threading.RLock()
    if str(self.path) != ":memory:":
        self.path.parent.mkdir(parents=True, exist_ok=True)
    self._shared: sqlite3.Connection | None = (
        sqlite3.connect(":memory:", check_same_thread=False)
        if str(self.path) == ":memory:"
        else None
    )
    self._init()

put

put(thread_id: str, state: dict[str, Any]) -> None

Insert a snapshot and prune old ones.

Raises:

Type	Description
SerializationError	When the state is not JSON-serialisable or the write fails.

Source code in src\windlass\agent\checkpoint.py

def put(self, thread_id: str, state: dict[str, Any]) -> None:
    """Insert a snapshot and prune old ones.

    Raises:
        SerializationError: When the state is not JSON-serialisable or the
            write fails.
    """
    try:
        payload = json.dumps(state, default=str)
    except (TypeError, ValueError) as exc:
        raise SerializationError(
            f"Agent state for thread {thread_id!r} is not JSON-serialisable: {exc}",
            hint="Keep custom objects out of agent state, or store an id instead.",
        ) from exc

    with self._lock:
        connection = self._connect()
        try:
            connection.execute(
                "INSERT INTO checkpoints (thread_id, state, saved_at) VALUES (?, ?, ?)",
                (thread_id, payload, time.time()),
            )
            connection.execute(
                """
                DELETE FROM checkpoints
                WHERE thread_id = ?
                  AND id NOT IN (
                      SELECT id FROM checkpoints
                      WHERE thread_id = ? ORDER BY id DESC LIMIT ?
                  )
                """,
                (thread_id, thread_id, self.max_history),
            )
            connection.commit()
        except sqlite3.Error as exc:
            raise SerializationError(f"Could not save the checkpoint: {exc}") from exc
        finally:
            if self._shared is None:
                connection.close()

get

get(thread_id: str) -> dict[str, Any] | None

Return the newest snapshot, or None.

Source code in src\windlass\agent\checkpoint.py

def get(self, thread_id: str) -> dict[str, Any] | None:
    """Return the newest snapshot, or ``None``."""
    rows = self.history(thread_id, limit=1)
    return rows[0] if rows else None

history

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent snapshots, newest first.

Source code in src\windlass\agent\checkpoint.py

def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent snapshots, newest first."""
    with self._lock:
        connection = self._connect()
        try:
            cursor = connection.execute(
                "SELECT state FROM checkpoints WHERE thread_id = ? ORDER BY id DESC LIMIT ?",
                (thread_id, limit),
            )
            rows = cursor.fetchall()
        except sqlite3.Error as exc:
            _log.warning("Could not read checkpoints for %s: %s", thread_id, exc)
            return []
        finally:
            if self._shared is None:
                connection.close()

    snapshots: list[dict[str, Any]] = []
    for (payload,) in rows:
        try:
            snapshots.append(json.loads(payload))
        except json.JSONDecodeError:  # pragma: no cover - corrupt row
            _log.warning("Skipping a corrupt checkpoint for thread %s.", thread_id)
    return snapshots

delete

delete(thread_id: str) -> None

Discard a thread's snapshots.

Source code in src\windlass\agent\checkpoint.py

def delete(self, thread_id: str) -> None:
    """Discard a thread's snapshots."""
    with self._lock:
        connection = self._connect()
        try:
            connection.execute("DELETE FROM checkpoints WHERE thread_id = ?", (thread_id,))
            connection.commit()
        except sqlite3.Error as exc:  # pragma: no cover
            _log.warning("Could not delete checkpoints for %s: %s", thread_id, exc)
        finally:
            if self._shared is None:
                connection.close()

threads

threads() -> list[str]

Return the known thread ids.

Source code in src\windlass\agent\checkpoint.py

def threads(self) -> list[str]:
    """Return the known thread ids."""
    with self._lock:
        connection = self._connect()
        try:
            cursor = connection.execute("SELECT DISTINCT thread_id FROM checkpoints")
            return sorted(row[0] for row in cursor.fetchall())
        except sqlite3.Error:  # pragma: no cover
            return []
        finally:
            if self._shared is None:
                connection.close()

close

close() -> None

Close the shared in-memory connection, if there is one.

Source code in src\windlass\agent\checkpoint.py

def close(self) -> None:
    """Close the shared in-memory connection, if there is one."""
    if self._shared is not None:
        self._shared.close()
        self._shared = None

AgentRuntime

AgentRuntime(
    *,
    llm: LLM,
    tools: ToolRegistry | Sequence[Any] | None = None,
    system_prompt: str | None = None,
    max_iterations: int = 10,
    memory: Memory | None = None,
    guardrail: Guardrail | None = None,
    checkpointer: Checkpointer | None = None,
    tracer: Tracer | None = None,
    parallel_tools: bool = True,
    require_approval: bool = False,
    max_tool_call_retries: int = 2,
    name: str = "agent"
)

A tool-calling agent.

Parameters:

Name	Type	Description	Default
llm	LLM	The reasoning model. Must support tool calling if tools are bound.	required
tools	ToolRegistry | Sequence[Any] | None	Tools the agent may call.	None
system_prompt	str | None	Instructions prepended to every run.	None
max_iterations	int	Ceiling on reason/act cycles. Prevents a confused agent from looping until your budget runs out.	10
memory	Memory | None	Optional conversation memory, keyed by thread_id.	None
guardrail	Guardrail | None	Optional input/output policy.	None
checkpointer	Checkpointer | None	Optional durable state, enabling resume and time travel.	None
tracer	Tracer | None	Observability backend.	None
parallel_tools	bool	Execute simultaneous tool calls concurrently.	True
require_approval	bool	Pause before every tool call, not just those whose tools declare requires_approval.	False
max_tool_call_retries	int	How many times a run may recover from the model emitting an unparseable tool call before giving up. Each recovery costs one iteration from max_iterations, so a model that cannot get it right still terminates. 0 disables recovery and restores the previous behaviour of failing immediately.	2
name	str	Agent name, used in traces and multi-agent routing.	'agent'

Attributes:

Name	Type	Description
tools		The bound :class:~windlass.tools.ToolRegistry.
name		The agent's name.

Raises:

Type	Description
AgentError	When tools are bound to a model that cannot call them.

Source code in src\windlass\agent\runtime.py

def __init__(
    self,
    *,
    llm: LLM,
    tools: ToolRegistry | Sequence[Any] | None = None,
    system_prompt: str | None = None,
    max_iterations: int = 10,
    memory: Memory | None = None,
    guardrail: Guardrail | None = None,
    checkpointer: Checkpointer | None = None,
    tracer: Tracer | None = None,
    parallel_tools: bool = True,
    require_approval: bool = False,
    max_tool_call_retries: int = 2,
    name: str = "agent",
) -> None:
    self.llm = llm
    self.tools = tools if isinstance(tools, ToolRegistry) else ToolRegistry(list(tools or []))
    self.system_prompt = system_prompt if system_prompt is not None else DEFAULT_SYSTEM_PROMPT
    self.max_iterations = max(1, max_iterations)
    self.memory = memory
    self.guardrail = guardrail
    self.checkpointer = checkpointer
    self.tracer = tracer or NullTracer()
    self.parallel_tools = parallel_tools
    self.require_approval = require_approval
    self.max_tool_call_retries = max(0, max_tool_call_retries)
    self.name = name

    if len(self.tools) and not llm.supports_tools:
        raise AgentError(
            f"The {llm.name!r} provider does not support tool calling, but "
            f"{len(self.tools)} tool(s) are bound.",
            hint="Use a tool-capable model, or drop the tools.",
            context={"provider": llm.name, "tools": self.tools.names()},
        )

arun async

arun(
    prompt: Any,
    *,
    thread_id: str | None = None,
    max_iterations: int | None = None,
    **llm_kwargs: Any
) -> AgentResponse

Run the agent to completion.

Parameters:

Name	Type	Description	Default
prompt	Any	A string, a message, or a full transcript.	required
thread_id	str | None	Conversation thread. Required to use memory or checkpointing across calls; generated when omitted.	None
max_iterations	int | None	Override for the step budget.	None
**llm_kwargs	Any	Per-call model overrides.	{}

Returns:

Type	Description
AgentResponse	The answer, transcript, per-step trace and aggregate usage.

Raises:

Type	Description
MaxIterationsExceeded	When the budget runs out before an answer.
InterruptedError_	When a tool needs human approval. Resume with :meth:aresume.
GuardrailViolation	When a guardrail blocks the input or output.
AgentError	When the run cannot proceed.

Performance

One model call per iteration, plus tool time. Independent tool calls in the same turn run concurrently, which is usually where the wall clock is won.

Example

import asyncio from windlass import Windlass agent = Windlass.agent().llm("fake", responses=["Done."]) asyncio.run(agent.arun("Say done.")).output 'Done.'

Source code in src\windlass\agent\runtime.py

async def arun(
    self,
    prompt: Any,
    *,
    thread_id: str | None = None,
    max_iterations: int | None = None,
    **llm_kwargs: Any,
) -> AgentResponse:
    """Run the agent to completion.

    Args:
        prompt: A string, a message, or a full transcript.
        thread_id: Conversation thread. Required to use memory or
            checkpointing across calls; generated when omitted.
        max_iterations: Override for the step budget.
        **llm_kwargs: Per-call model overrides.

    Returns:
        The answer, transcript, per-step trace and aggregate usage.

    Raises:
        MaxIterationsExceeded: When the budget runs out before an answer.
        AgentInterrupt: When a tool needs human approval. Resume with
            :meth:`aresume`.
        GuardrailViolation: When a guardrail blocks the input or output.
        AgentError: When the run cannot proceed.

    Performance:
        One model call per iteration, plus tool time. Independent tool calls
        in the same turn run concurrently, which is usually where the wall
        clock is won.

    Example:
        >>> import asyncio
        >>> from windlass import Windlass
        >>> agent = Windlass.agent().llm("fake", responses=["Done."])
        >>> asyncio.run(agent.arun("Say done.")).output
        'Done.'
    """
    thread = thread_id or new_id("thread")
    budget = max_iterations or self.max_iterations
    started = time.perf_counter()

    with self.tracer.span(f"agent.{self.name}", kind="agent", inputs=str(prompt)) as span:
        messages = await self._prepare(prompt, thread)
        steps: list[AgentStep] = []
        usage = Usage(calls=0)

        corrections = 0
        for index in range(budget):
            try:
                completion = await self._think(messages, index, **llm_kwargs)
            except MalformedToolCallError as exc:
                # The model emitted a tool call the provider could not
                # parse — usually by nesting one call inside another's
                # arguments, which the protocol cannot express. That is the
                # same class of mistake as inventing a tool name, so it gets
                # the same treatment: show the model what went wrong and let
                # it try again, rather than ending the run.
                corrections += 1
                if corrections > self.max_tool_call_retries:
                    raise
                _log.warning(
                    "Model produced an unparseable tool call (correction %d/%d): %s",
                    corrections,
                    self.max_tool_call_retries,
                    exc,
                )
                messages.append(Message.user(exc.feedback))
                steps.append(AgentStep(index=index, thought=f"[recovered] {exc}"))
                continue

            usage = usage + completion.usage
            assistant = completion.to_message()
            messages.append(assistant)

            if not completion.tool_calls:
                output = await self._finish(completion.content, thread, prompt, messages)
                steps.append(
                    AgentStep(index=index, thought=completion.content, usage=completion.usage)
                )
                span.set_output(output)
                span.set_usage(usage)
                self._checkpoint(thread, messages, steps, done=True)
                return AgentResponse(
                    output=output,
                    messages=messages,
                    steps=steps,
                    usage=usage,
                    latency_ms=(time.perf_counter() - started) * 1000,
                    thread_id=thread,
                )

            pending = self._approvals(completion.tool_calls)
            if pending:
                self._checkpoint(thread, messages, steps, pending=pending)
                raise AgentInterrupt(
                    f"{len(pending)} tool call(s) need approval: "
                    + ", ".join(c.name for c in pending),
                    payload=[c.model_dump() for c in pending],
                    thread_id=thread,
                )

            results = await self._act(completion.tool_calls)
            messages.extend(Message.tool(r) for r in results)
            steps.append(
                AgentStep(
                    index=index,
                    thought=completion.content,
                    tool_calls=completion.tool_calls,
                    tool_results=results,
                    usage=completion.usage,
                )
            )
            self._checkpoint(thread, messages, steps)

        raise MaxIterationsExceeded(budget)

run

run(
    prompt: Any,
    *,
    thread_id: str | None = None,
    max_iterations: int | None = None,
    **llm_kwargs: Any
) -> AgentResponse

Blocking :meth:arun.

Source code in src\windlass\agent\runtime.py

def run(
    self,
    prompt: Any,
    *,
    thread_id: str | None = None,
    max_iterations: int | None = None,
    **llm_kwargs: Any,
) -> AgentResponse:
    """Blocking :meth:`arun`."""
    return run_sync(
        self.arun(prompt, thread_id=thread_id, max_iterations=max_iterations, **llm_kwargs)
    )

astream async

astream(
    prompt: Any, *, thread_id: str | None = None, **llm_kwargs: Any
) -> AsyncIterator[StreamEvent]

Stream the agent's work as it happens.

You get text deltas as the model produces them, tool_call events when it decides to act, and a final done. That is what a live agent UI needs: users tolerate latency far better when they can see progress.

Parameters:

Name	Type	Description	Default
prompt	Any	A string, a message, or a full transcript.	required
thread_id	str | None	Conversation thread.	None
**llm_kwargs	Any	Per-call model overrides.	{}

Yields:

Type	Description
AsyncIterator[StreamEvent]	class:~windlass.core.types.StreamEvent values.

Raises:

Type	Description
MaxIterationsExceeded	When the budget runs out.

Source code in src\windlass\agent\runtime.py

async def astream(
    self,
    prompt: Any,
    *,
    thread_id: str | None = None,
    **llm_kwargs: Any,
) -> AsyncIterator[StreamEvent]:
    """Stream the agent's work as it happens.

    You get text deltas as the model produces them, ``tool_call`` events
    when it decides to act, and a final ``done``. That is what a live agent
    UI needs: users tolerate latency far better when they can see progress.

    Args:
        prompt: A string, a message, or a full transcript.
        thread_id: Conversation thread.
        **llm_kwargs: Per-call model overrides.

    Yields:
        :class:`~windlass.core.types.StreamEvent` values.

    Raises:
        MaxIterationsExceeded: When the budget runs out.
    """
    thread = thread_id or new_id("thread")
    messages = await self._prepare(prompt, thread)
    schemas = self._schemas()

    for _ in range(self.max_iterations):
        collected: list[str] = []
        calls: list[ToolCall] = []

        async for event in self.llm.astream(messages, tools=schemas, **llm_kwargs):
            if event.type == "text" and event.delta:
                collected.append(event.delta)
                yield event
            elif event.type == "tool_call" and event.tool_call:
                calls.append(event.tool_call)
                yield event

        text = "".join(collected)
        messages.append(Message(role=Role.ASSISTANT, content=text, tool_calls=calls))

        if not calls:
            await self._finish(text, thread, prompt, messages)
            yield StreamEvent(type="done")
            return

        results = await self._act(calls)
        messages.extend(Message.tool(r) for r in results)

    raise MaxIterationsExceeded(self.max_iterations)

stream

stream(prompt: Any, **kwargs: Any) -> Iterator[StreamEvent]

Blocking :meth:astream.

Source code in src\windlass\agent\runtime.py

def stream(self, prompt: Any, **kwargs: Any) -> Iterator[StreamEvent]:
    """Blocking :meth:`astream`."""
    return iter_sync(self.astream(prompt, **kwargs))

astream_text async

astream_text(prompt: Any, **kwargs: Any) -> AsyncIterator[str]

Stream only the assistant's text — the common case for a chat UI.

Source code in src\windlass\agent\runtime.py

async def astream_text(self, prompt: Any, **kwargs: Any) -> AsyncIterator[str]:
    """Stream only the assistant's text — the common case for a chat UI."""
    async for event in self.astream(prompt, **kwargs):
        if event.type == "text" and event.delta:
            yield event.delta

aresume async

aresume(
    thread_id: str,
    *,
    approved: bool = True,
    feedback: str = "",
    edited_arguments: dict[str, dict[str, Any]] | None = None,
    **llm_kwargs: Any
) -> AgentResponse

Resume a run that paused for human approval.

Parameters:

Name	Type	Description	Default
thread_id	str	The thread reported on the :class:AgentInterrupt.	required
approved	bool	True to execute the pending calls, False to reject them.	True
feedback	str	Message shown to the model when rejecting, so it can try a different approach instead of simply repeating itself.	''
edited_arguments	dict[str, dict[str, Any]] | None	Replacement arguments keyed by tool-call id. This is the "human edits the action" case — approve the intent, fix the parameters.	None
**llm_kwargs	Any	Per-call model overrides.	{}

Returns:

Type	Description
AgentResponse	The completed run.

Raises:

Type	Description
AgentError	When there is no checkpoint, or nothing was pending.

Example

from windlass import Windlass, tool, AgentInterrupt @tool(requires_approval=True) ... def deploy(env: str) -> str: ... '''Deploy to an environment.''' ... return f"deployed to {env}" agent = (Windlass.agent() ... .llm("fake", responses=["", "Deployed."], ... tool_calls=[[ToolCall(name="deploy", ... arguments={"env": "prod"})], []]) ... .tool(deploy).checkpoint()) try: ... agent.run("Deploy to prod", thread_id="t1") ... except AgentInterrupt as pause: ... resumed = agent.resume("t1", approved=True) resumed.output 'Deployed.'

Source code in src\windlass\agent\runtime.py

async def aresume(
    self,
    thread_id: str,
    *,
    approved: bool = True,
    feedback: str = "",
    edited_arguments: dict[str, dict[str, Any]] | None = None,
    **llm_kwargs: Any,
) -> AgentResponse:
    """Resume a run that paused for human approval.

    Args:
        thread_id: The thread reported on the :class:`AgentInterrupt`.
        approved: True to execute the pending calls, False to reject them.
        feedback: Message shown to the model when rejecting, so it can try a
            different approach instead of simply repeating itself.
        edited_arguments: Replacement arguments keyed by tool-call id. This
            is the "human edits the action" case — approve the intent, fix
            the parameters.
        **llm_kwargs: Per-call model overrides.

    Returns:
        The completed run.

    Raises:
        AgentError: When there is no checkpoint, or nothing was pending.

    Example:
        >>> from windlass import Windlass, tool, AgentInterrupt
        >>> @tool(requires_approval=True)
        ... def deploy(env: str) -> str:
        ...     '''Deploy to an environment.'''
        ...     return f"deployed to {env}"
        >>> agent = (Windlass.agent()
        ...          .llm("fake", responses=["", "Deployed."],
        ...               tool_calls=[[ToolCall(name="deploy",
        ...                                     arguments={"env": "prod"})], []])
        ...          .tool(deploy).checkpoint())
        >>> try:
        ...     agent.run("Deploy to prod", thread_id="t1")
        ... except AgentInterrupt as pause:
        ...     resumed = agent.resume("t1", approved=True)
        >>> resumed.output
        'Deployed.'
    """
    if self.checkpointer is None:
        raise AgentError(
            "Resuming needs a checkpointer.",
            hint="Enable one with .checkpoint() on the agent builder.",
        )
    snapshot = self.checkpointer.get(thread_id)
    if snapshot is None:
        raise AgentError(
            f"No checkpoint found for thread {thread_id!r}.",
            context={"threads": self.checkpointer.threads()[:20]},
        )
    pending_raw = snapshot.get("pending") or []
    if not pending_raw:
        raise AgentError(
            f"Thread {thread_id!r} is not waiting for approval.",
            hint="Only interrupted runs can be resumed.",
        )

    messages = [Message.model_validate(m) for m in snapshot.get("messages", [])]
    steps = [AgentStep.model_validate(s) for s in snapshot.get("steps", [])]
    pending = [ToolCall.model_validate(c) for c in pending_raw]

    if approved:
        if edited_arguments:
            pending = [
                call.model_copy(
                    update={"arguments": edited_arguments.get(call.id, call.arguments)}
                )
                for call in pending
            ]
        results = await self._act(pending)
    else:
        from windlass.core.types import ToolResult

        reason = feedback or "A human rejected this action."
        results = [
            ToolResult(call_id=c.id, name=c.name, content=reason, is_error=True)
            for c in pending
        ]

    messages.extend(Message.tool(r) for r in results)
    steps.append(AgentStep(index=len(steps), tool_calls=pending, tool_results=results))
    self._checkpoint(thread_id, messages, steps)

    response = await self.arun(
        messages,
        thread_id=thread_id,
        max_iterations=max(1, self.max_iterations - len(steps)),
        **llm_kwargs,
    )

    # ``arun`` starts a fresh step list, so without this the approved call —
    # the one a human explicitly authorised — is absent from ``steps`` while
    # still present in ``messages``. Anything reading ``steps`` as the audit
    # trail would show a payment that no step accounts for.
    if steps:
        combined = [*steps, *response.steps]
        for index, step in enumerate(combined):
            step.index = index
        response.steps = combined
    return response

resume

resume(thread_id: str, **kwargs: Any) -> AgentResponse

Blocking :meth:aresume.

Source code in src\windlass\agent\runtime.py

def resume(self, thread_id: str, **kwargs: Any) -> AgentResponse:
    """Blocking :meth:`aresume`."""
    return run_sync(self.aresume(thread_id, **kwargs))

pending_approvals

pending_approvals(thread_id: str) -> list[ToolCall]

Return the tool calls awaiting approval on a thread.

Parameters:

Name	Type	Description	Default
thread_id	str	The thread to inspect.	required

Returns:

Type	Description
list[ToolCall]	The pending calls, or [].

Source code in src\windlass\agent\runtime.py

def pending_approvals(self, thread_id: str) -> list[ToolCall]:
    """Return the tool calls awaiting approval on a thread.

    Args:
        thread_id: The thread to inspect.

    Returns:
        The pending calls, or ``[]``.
    """
    if self.checkpointer is None:
        return []
    snapshot = self.checkpointer.get(thread_id) or {}
    return [ToolCall.model_validate(c) for c in snapshot.get("pending") or []]

state

state(thread_id: str) -> dict[str, Any] | None

Return the latest checkpoint for a thread, or None.

Source code in src\windlass\agent\runtime.py

def state(self, thread_id: str) -> dict[str, Any] | None:
    """Return the latest checkpoint for a thread, or ``None``."""
    return self.checkpointer.get(thread_id) if self.checkpointer else None

history

history(thread_id: str, limit: int = 20) -> list[dict[str, Any]]

Return recent checkpoints for a thread, newest first.

Source code in src\windlass\agent\runtime.py

def history(self, thread_id: str, limit: int = 20) -> list[dict[str, Any]]:
    """Return recent checkpoints for a thread, newest first."""
    return self.checkpointer.history(thread_id, limit) if self.checkpointer else []

native_llm

native_llm() -> Any

Return the model provider's own client.

Source code in src\windlass\agent\runtime.py

def native_llm(self) -> Any:
    """Return the model provider's own client."""
    return self.llm.native()

native_graph

native_graph() -> Any

Return the underlying execution graph.

The built-in runtime has no graph — it is a loop. Use .graph() on the agent builder to get a LangGraph-backed runtime whose native_graph() returns a real StateGraph you can add nodes and edges to.

Raises:

Type	Description
AgentError	Always, explaining how to get a real graph.

Source code in src\windlass\agent\runtime.py

def native_graph(self) -> Any:
    """Return the underlying execution graph.

    The built-in runtime has no graph — it is a loop. Use ``.graph()`` on the
    agent builder to get a LangGraph-backed runtime whose ``native_graph()``
    returns a real ``StateGraph`` you can add nodes and edges to.

    Raises:
        AgentError: Always, explaining how to get a real graph.
    """
    raise AgentError(
        "The built-in runtime is a loop, not a graph.",
        hint="Build the agent with .graph() to use LangGraph, then call "
        "native_graph() to add nodes, edges and conditional edges.",
    )

describe

describe() -> dict[str, Any]

Return a JSON-safe description of the agent's configuration.

Source code in src\windlass\agent\runtime.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe description of the agent's configuration."""
    parts: dict[str, Any] = {
        "name": self.name,
        "runtime": "builtin",
        "llm": self.llm.describe(),
        "tools": self.tools.describe(),
        "max_iterations": self.max_iterations,
        "parallel_tools": self.parallel_tools,
    }
    for label, component in (
        ("memory", self.memory),
        ("guardrail", self.guardrail),
        ("tracer", self.tracer),
    ):
        if component is not None:
            parts[label] = component.describe()
    if self.checkpointer is not None:
        parts["checkpointer"] = type(self.checkpointer).__name__
    return parts

aclose async

aclose() -> None

Release resources held by the model and any MCP connections.

Source code in src\windlass\agent\runtime.py

async def aclose(self) -> None:
    """Release resources held by the model and any MCP connections."""
    for component in (self.llm, self.tracer):
        close = getattr(component, "aclose", None)
        if close is not None:
            try:
                await close()
            except Exception as exc:
                _log.debug("Closing %s failed: %s", component, exc)

close

close() -> None

Blocking :meth:aclose.

Source code in src\windlass\agent\runtime.py

def close(self) -> None:
    """Blocking :meth:`aclose`."""
    run_sync(self.aclose())

AgentTool

AgentTool(agent: Any, *, name: str, description: str = '', **config: Any)

Bases: Tool

Exposes an agent as a tool the supervisor can call.

This is what makes delegation work with ordinary tool calling: the supervisor does not need a special routing mechanism, because a worker is a tool from its point of view.

Parameters:

Name	Type	Description	Default
agent	Any	The worker agent, or anything with an arun/run method.	required
name	str	Tool name the supervisor sees. Should say what the worker is for.	required
description	str	What this specialist is good at. The supervisor routes on this text, so it deserves the same care as a prompt.	''
**config	Any	Forwarded to :class:~windlass.interfaces.tool.Tool.	{}

Example

from windlass import Windlass worker = Windlass.agent().llm("fake", responses=["done"]) t = AgentTool(worker, name="helper", description="Does small jobs.") t.run(task="tidy up").content 'done'

Source code in src\windlass\agent\supervisor.py

def __init__(
    self,
    agent: Any,
    *,
    name: str,
    description: str = "",
    **config: Any,
) -> None:
    super().__init__(
        name=name,
        description=description or f"Delegate a task to the {name!r} specialist.",
        parameters={
            "type": "object",
            "properties": {
                "task": {
                    "type": "string",
                    "description": "A complete, self-contained description of "
                    "the work. The specialist cannot see the wider conversation.",
                }
            },
            "required": ["task"],
        },
        **config,
    )
    self.agent = agent

acall async

acall(**kwargs: Any) -> Any

Run the worker agent on the delegated task.

Parameters:

Name	Type	Description	Default
**kwargs	Any	Must contain task.	{}

Returns:

Type	Description
Any	The worker's output text.

Raises:

Type	Description
AgentError	When the worker fails.

Source code in src\windlass\agent\supervisor.py

async def acall(self, **kwargs: Any) -> Any:
    """Run the worker agent on the delegated task.

    Args:
        **kwargs: Must contain ``task``.

    Returns:
        The worker's output text.

    Raises:
        AgentError: When the worker fails.
    """
    task = str(kwargs.get("task", "")).strip()
    if not task:
        return "No task was provided."
    runner = getattr(self.agent, "arun", None)
    if runner is None:
        raise AgentError(f"{self.name!r} is not runnable — it has no arun method.")
    response = await runner(task)
    return getattr(response, "output", str(response))

native

native() -> Any

Return the wrapped agent.

Source code in src\windlass\agent\supervisor.py

def native(self) -> Any:
    """Return the wrapped agent."""
    return self.agent

Supervisor

Supervisor(
    *,
    llm: LLM,
    agents: dict[str, Any] | None = None,
    descriptions: dict[str, str] | None = None,
    system_prompt: str | None = None,
    max_iterations: int = 10,
    tools: list[Any] | None = None,
    tracer: Tracer | None = None,
    **kwargs: Any
)

Bases: AgentRuntime

An agent that delegates to specialist agents.

Parameters:

Name	Type	Description	Default
llm	LLM	The coordinating model. It only routes and summarises, so a mid-sized model is usually the right call.	required
agents	dict[str, Any] | None	Workers, keyed by the name the supervisor will use. Values may be :class:~windlass.agent.runtime.AgentRuntime instances, builders, or anything with arun.	None
descriptions	dict[str, str] | None	What each worker is for, keyed by the same names. Falls back to the worker's own description.	None
system_prompt	str | None	Override for the routing instructions.	None
max_iterations	int	Ceiling on routing rounds.	10
tools	list[Any] | None	Tools the supervisor may call itself, alongside its workers.	None
tracer	Tracer | None	Observability backend.	None
**kwargs	Any	Forwarded to :class:~windlass.agent.runtime.AgentRuntime.	{}

Raises:

Type	Description
AgentError	When no workers are supplied.

Performance

Workers requested in the same turn run concurrently, so a supervisor that fans out to three specialists waits for the slowest, not the sum.

Source code in src\windlass\agent\supervisor.py

def __init__(
    self,
    *,
    llm: LLM,
    agents: dict[str, Any] | None = None,
    descriptions: dict[str, str] | None = None,
    system_prompt: str | None = None,
    max_iterations: int = 10,
    tools: list[Any] | None = None,
    tracer: Tracer | None = None,
    **kwargs: Any,
) -> None:
    workers = dict(agents or {})
    if not workers:
        raise AgentError(
            "A supervisor needs at least one specialist agent.",
            hint="Supervisor(llm=..., agents={'researcher': agent, ...})",
        )

    notes = dict(descriptions or {})
    agent_tools: list[Tool] = [
        AgentTool(worker, name=name, description=notes.get(name, _describe(worker, name)))
        for name, worker in workers.items()
    ]

    roster = "\n".join(f"  - {t.name}: {t.description}" for t in agent_tools)
    super().__init__(
        llm=llm,
        tools=[*agent_tools, *(tools or [])],
        system_prompt=system_prompt or SUPERVISOR_PROMPT.format(roster=roster),
        max_iterations=max_iterations,
        tracer=tracer or NullTracer(),
        name=kwargs.pop("name", "supervisor"),
        **kwargs,
    )
    self.agents = workers

add_agent

add_agent(name: str, agent: Any, description: str = '') -> Supervisor

Add a specialist after construction.

Parameters:

Name	Type	Description	Default
name	str	The name the supervisor will use.	required
agent	Any	The worker.	required
description	str	What it is good at.	''

Returns:

Type	Description
Supervisor	self.

Source code in src\windlass\agent\supervisor.py

def add_agent(self, name: str, agent: Any, description: str = "") -> Supervisor:
    """Add a specialist after construction.

    Args:
        name: The name the supervisor will use.
        agent: The worker.
        description: What it is good at.

    Returns:
        ``self``.
    """
    self.agents[name] = agent
    self.tools.add(
        AgentTool(agent, name=name, description=description or _describe(agent, name))
    )
    return self

abroadcast async

abroadcast(task: str) -> dict[str, AgentResponse]

Run every specialist on the same task, concurrently.

Useful for ensembling — ask three specialists the same question and compare — and for fan-out work where every worker has something to contribute.

Parameters:

Name	Type	Description	Default
task	str	The task to send to every worker.	required

Returns:

Type	Description
dict[str, AgentResponse]	A {worker_name: response} mapping. Workers that fail are
dict[str, AgentResponse]	reported as an error response rather than raising, so one failure
dict[str, AgentResponse]	does not lose the others' work.

Example

import asyncio from windlass import Windlass boss = Supervisor( ... llm=Windlass.llm("fake"), ... agents={"a": Windlass.agent().llm("fake", responses=["A"])}, ... ) asyncio.run(boss.abroadcast("go"))["a"].output 'A'

Source code in src\windlass\agent\supervisor.py

async def abroadcast(self, task: str) -> dict[str, AgentResponse]:
    """Run every specialist on the same task, concurrently.

    Useful for ensembling — ask three specialists the same question and
    compare — and for fan-out work where every worker has something to
    contribute.

    Args:
        task: The task to send to every worker.

    Returns:
        A ``{worker_name: response}`` mapping. Workers that fail are
        reported as an error response rather than raising, so one failure
        does not lose the others' work.

    Example:
        >>> import asyncio
        >>> from windlass import Windlass
        >>> boss = Supervisor(
        ...     llm=Windlass.llm("fake"),
        ...     agents={"a": Windlass.agent().llm("fake", responses=["A"])},
        ... )
        >>> asyncio.run(boss.abroadcast("go"))["a"].output
        'A'
    """
    names = list(self.agents)

    async def _run(name: str) -> AgentResponse:
        worker = self.agents[name]
        runner = getattr(worker, "arun", None)
        if runner is None:
            raise AgentError(f"{name!r} is not runnable.")
        return await runner(task)

    started = time.perf_counter()
    outcomes = await gather_bounded(
        [_run(n) for n in names], limit=len(names), return_exceptions=True
    )

    results: dict[str, AgentResponse] = {}
    for name, outcome in zip(names, outcomes, strict=True):
        if isinstance(outcome, BaseException):
            _log.warning("Specialist %s failed: %s", name, outcome)
            results[name] = AgentResponse(
                output=f"{name} failed: {outcome}",
                metadata={"error": True, "agent": name},
            )
        else:
            results[name] = outcome
    _log.debug(
        "Broadcast to %d specialists in %.2fs", len(names), time.perf_counter() - started
    )
    return results

broadcast

broadcast(task: str) -> dict[str, AgentResponse]

Blocking :meth:abroadcast.

Source code in src\windlass\agent\supervisor.py

def broadcast(self, task: str) -> dict[str, AgentResponse]:
    """Blocking :meth:`abroadcast`."""
    return run_sync(self.abroadcast(task))

apipeline async

apipeline(task: str, order: list[str]) -> AgentResponse

Chain specialists in sequence, each seeing the previous output.

The other multi-agent shape: research → draft → edit, where each stage genuinely depends on the last.

Parameters:

Name	Type	Description	Default
task	str	The initial task.	required
order	list[str]	Worker names, in execution order.	required

Returns:

Type	Description
AgentResponse	The final worker's response, with every stage recorded in
AgentResponse	attr:~windlass.core.types.AgentResponse.steps.

Raises:

Type	Description
AgentError	When a named worker does not exist.

Example

import asyncio from windlass import Windlass boss = Supervisor( ... llm=Windlass.llm("fake"), ... agents={ ... "draft": Windlass.agent().llm("fake", responses=["a draft"]), ... "edit": Windlass.agent().llm("fake", responses=["an edit"]), ... }, ... ) asyncio.run(boss.apipeline("write", ["draft", "edit"])).output 'an edit'

Source code in src\windlass\agent\supervisor.py

async def apipeline(self, task: str, order: list[str]) -> AgentResponse:
    """Chain specialists in sequence, each seeing the previous output.

    The other multi-agent shape: research → draft → edit, where each stage
    genuinely depends on the last.

    Args:
        task: The initial task.
        order: Worker names, in execution order.

    Returns:
        The final worker's response, with every stage recorded in
        :attr:`~windlass.core.types.AgentResponse.steps`.

    Raises:
        AgentError: When a named worker does not exist.

    Example:
        >>> import asyncio
        >>> from windlass import Windlass
        >>> boss = Supervisor(
        ...     llm=Windlass.llm("fake"),
        ...     agents={
        ...         "draft": Windlass.agent().llm("fake", responses=["a draft"]),
        ...         "edit": Windlass.agent().llm("fake", responses=["an edit"]),
        ...     },
        ... )
        >>> asyncio.run(boss.apipeline("write", ["draft", "edit"])).output
        'an edit'
    """
    missing = [n for n in order if n not in self.agents]
    if missing:
        raise AgentError(
            f"Unknown specialist(s): {', '.join(missing)}.",
            context={"available": sorted(self.agents)},
        )

    started = time.perf_counter()
    thread = new_id("pipeline")
    current = task
    steps: list[AgentStep] = []
    transcript: list[Message] = [Message.user(task)]
    usage = Usage(calls=0)

    for index, name in enumerate(order):
        worker = self.agents[name]
        response = await worker.arun(current)
        current = response.output
        usage = usage + response.usage
        transcript.append(Message.assistant(current, name=name))
        steps.append(AgentStep(index=index, thought=current, node=name, usage=response.usage))

    return AgentResponse(
        output=current,
        messages=transcript,
        steps=steps,
        usage=usage,
        latency_ms=(time.perf_counter() - started) * 1000,
        thread_id=thread,
        metadata={"pattern": "pipeline", "order": order},
    )

pipeline

pipeline(task: str, order: list[str]) -> AgentResponse

Blocking :meth:apipeline.

Source code in src\windlass\agent\supervisor.py

def pipeline(self, task: str, order: list[str]) -> AgentResponse:
    """Blocking :meth:`apipeline`."""
    return run_sync(self.apipeline(task, order))

describe

describe() -> dict[str, Any]

Return a JSON-safe description including the roster.

Source code in src\windlass\agent\supervisor.py

def describe(self) -> dict[str, Any]:
    """Return a JSON-safe description including the roster."""
    return {
        **super().describe(),
        "pattern": "supervisor",
        "specialists": sorted(self.agents),
    }