windlass.agent.runtime

runtime

The built-in agent runtime.

A reason/act loop: call the model, execute whatever tools it asks for, feed the results back, repeat until it answers or runs out of budget. That is what an agent is, and Windlass implements it directly rather than requiring a graph library — so agents work on a bare pip install windlass.

The runtime supports everything the LangGraph path does except graph topology: streaming, memory, guardrails, checkpointing, human-in-the-loop approval, parallel tool execution and full tracing. Reach for :class:~windlass.agent.graph.LangGraphRuntime (.graph() on the builder) when you need conditional routing, subgraphs or explicit state machines.

Example

from windlass import Windlass, tool @tool ... def add(a: int, b: int) -> int: ... '''Add two numbers.''' ... return a + b agent = Windlass.agent().llm("fake", responses=["4"]).tool(add) agent.run("What is 2 + 2?").output '4'

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())