Agent

The high-level entry point — configure an Agent, run it, and read back a typed AgentResult.

For the concepts, start with Agent and the agent loop.

Agent class

Agent(
    model: str | ModelProtocol | None = None,
    tools: list[Tool] | None = None,
    system_prompt: str | None = None,
    reflexion: ReflexionConfig | bool | None = None,
    grounding: GroundingConfig | bool | None = None,
    max_iterations: int = 20,
    conversation_manager: ConversationManager | None = None,
    checkpointer: Checkpointer | None = None,
    hooks: list[HookProvider] | None = None,
    config: AgentConfig | None = None,
    **kwargs,
)

Agent

Agent

Bases: AgentRuntimeMixin, BaseModel

Primary entry point for Tulip agents.

Manages the ReAct loop with optional Reflexion and Grounding.

Usage

agent = Agent( model="openai:gpt-4o", # or anthropic:claude-sonnet-4-6 tools=[search, calculate], system_prompt="You are a helpful assistant.", )

Async streaming

async for event in agent.run("What is 2+2?"): print(event)

Sync execution

result = agent.run_sync("What is 2+2?") print(result.message)

Initialize an Agent.

Parameters:

Name	Type	Description	Default
model	str | Any | None	Model string or ModelProtocol instance	None
tools	list[Tool] | None	List of tools available to the agent	None
system_prompt	str | None	System prompt for the agent	None
reflexion	ReflexionConfig | bool | None	Reflexion config (True for defaults, False/None to disable)	None
grounding	GroundingConfig | bool | None	Grounding config (True for defaults, False/None to disable)	None
max_iterations	int	Maximum iterations before stopping	20
conversation_manager	Any | None	Conversation manager for message pruning	None
checkpointer	Any | None	Checkpointer for state persistence	None
hooks	list[Any] | None	Lifecycle hooks	None
config	AgentConfig | None	Full AgentConfig (overrides other params)	None
**kwargs	Any	Any other :class:~tulip.agent.config.AgentConfig field, forwarded verbatim. 36 options are reachable only this way and are therefore invisible to help(Agent), to editor autocomplete, and to inspect.signature(Agent) -- Pydantic's metaclass builds __signature__ from the explicit parameters above and drops the **kwargs. They are real, supported, and documented; they just cannot be discovered by introspection. The ones most often reached for: termination, output_schema (+ output_schema_strict, output_schema_retries), memory_manager, skills, playbook, planning, auxiliary_model, gsar, require_verification, plugins, temperature, max_tokens, token_budget, time_budget_seconds, max_concurrency, tool_execution, tool_result_store, web_search, web_fetch, image_generator, speech_provider, name, agent_id, metadata. See :class:~tulip.agent.config.AgentConfig for the full set and the type of each.	{}

Source code in .sdk/src/tulip/agent/agent.py

def __init__(
    self,
    model: str | Any | None = None,
    tools: list[Tool] | None = None,
    system_prompt: str | None = None,
    reflexion: ReflexionConfig | bool | None = None,
    grounding: GroundingConfig | bool | None = None,
    max_iterations: int = 20,
    conversation_manager: Any | None = None,
    checkpointer: Any | None = None,
    hooks: list[Any] | None = None,
    config: AgentConfig | None = None,
    **kwargs: Any,
):
    """
    Initialize an Agent.

    Args:
        model: Model string or ModelProtocol instance
        tools: List of tools available to the agent
        system_prompt: System prompt for the agent
        reflexion: Reflexion config (True for defaults, False/None to disable)
        grounding: Grounding config (True for defaults, False/None to disable)
        max_iterations: Maximum iterations before stopping
        conversation_manager: Conversation manager for message pruning
        checkpointer: Checkpointer for state persistence
        hooks: Lifecycle hooks
        config: Full AgentConfig (overrides other params)
        **kwargs: Any other :class:`~tulip.agent.config.AgentConfig`
            field, forwarded verbatim. 36 options are reachable only
            this way and are therefore invisible to ``help(Agent)``,
            to editor autocomplete, and to ``inspect.signature(Agent)``
            -- Pydantic's metaclass builds ``__signature__`` from the
            explicit parameters above and drops the ``**kwargs``. They
            are real, supported, and documented; they just cannot be
            discovered by introspection. The ones most often reached
            for:

            ``termination``, ``output_schema`` (+ ``output_schema_strict``,
            ``output_schema_retries``), ``memory_manager``, ``skills``,
            ``playbook``, ``planning``, ``auxiliary_model``, ``gsar``,
            ``require_verification``, ``plugins``, ``temperature``,
            ``max_tokens``, ``token_budget``, ``time_budget_seconds``,
            ``max_concurrency``, ``tool_execution``, ``tool_result_store``,
            ``web_search``, ``web_fetch``, ``image_generator``,
            ``speech_provider``, ``name``, ``agent_id``, ``metadata``.

            See :class:`~tulip.agent.config.AgentConfig` for the full
            set and the type of each.
    """
    # Build config from params or use provided
    if config is not None:
        agent_config = config
    else:
        # Handle reflexion
        reflexion_config = None
        if reflexion is True:
            reflexion_config = ReflexionConfig()
        elif isinstance(reflexion, ReflexionConfig):
            reflexion_config = reflexion

        # Handle grounding
        grounding_config = None
        if grounding is True:
            grounding_config = GroundingConfig()
        elif isinstance(grounding, GroundingConfig):
            grounding_config = grounding

        agent_config = AgentConfig(
            model=model or "openai:gpt-4o",
            tools=tools or [],
            system_prompt=system_prompt or "You are a helpful AI assistant.",
            reflexion=reflexion_config,
            grounding=grounding_config,
            max_iterations=max_iterations,
            conversation_manager=conversation_manager,
            checkpointer=checkpointer,
            hooks=hooks or [],
            **kwargs,
        )

    super().__init__(config=agent_config)
    self._initialize()

observation_pack property

observation_pack: Any

The agent's :class:~tulip.memory.observation_pack.ObservationPack, or None.

Its stats(session) says what it did for a thread (or a thread-less run, by run id).

name property

name: str | None

Display name from the config (multi-agent label, logs, traces).

is_cancelled property

is_cancelled: bool

Whether a cancel-all (cancel() with no thread id) is pending.

model property

model: Any

Get the model instance.

tools property

tools: ToolRegistry

Get the tool registry.

system_prompt property

system_prompt: str

Get the configured system prompt as a string.

If the config value is a callable (dynamic prompt), it is coerced to its repr so this property never returns non-str. Use self.config.system_prompt directly to access the raw value (string or callable) when you need to invoke the dynamic form.

arun async

arun

Run the agent asynchronously and return a single AgentResult.

This is the async, browser-safe equivalent of :meth:run_sync: it drives the same event loop (self.run(...)), builds the same AgentResult, but without spawning a thread or calling asyncio.run. That makes it usable in environments where threads are unavailable — notably Pyodide / WebAssembly, where run_sync cannot work.

Unlike run_sync it does NOT close the underlying model client or checkpointer pool afterward — the caller owns the loop, so those resources stay bound to it and can be reused across arun calls.

Each call starts from an empty conversation unless you give it somewhere to remember: prior turns are reloaded only when the agent has a checkpointer configured and you pass thread_id. Calling run_sync(prompt) in a loop therefore produces independent turns, not a conversation — which is the right default for a shared agent, and a surprise if you expected it to follow along.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt to process	required
thread_id	str | None	Conversation to continue. Required (with a configured checkpointer) for this run to see earlier turns.	None
metadata	dict[str, Any] | None	Additional metadata for tools	None

Returns:

Type	Description
AgentResult	AgentResult with final message and state

Source code in .sdk/src/tulip/agent/agent.py

async def arun(
    self,
    prompt: str,
    *,
    thread_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    model_kwargs: dict[str, Any] | None = None,
) -> AgentResult:
    """
    Run the agent asynchronously and return a single ``AgentResult``.

    This is the async, browser-safe equivalent of :meth:`run_sync`: it
    drives the same event loop (``self.run(...)``), builds the same
    ``AgentResult``, but without spawning a thread or calling
    ``asyncio.run``. That makes it usable in environments where threads
    are unavailable — notably Pyodide / WebAssembly, where ``run_sync`` cannot work.

    Unlike ``run_sync`` it does NOT close the underlying model client or
    checkpointer pool afterward — the caller owns the loop, so those
    resources stay bound to it and can be reused across ``arun`` calls.

    **Each call starts from an empty conversation** unless you give it
    somewhere to remember: prior turns are reloaded only when the agent has
    a ``checkpointer`` configured *and* you pass ``thread_id``. Calling
    ``run_sync(prompt)`` in a loop therefore produces independent turns,
    not a conversation — which is the right default for a shared agent, and
    a surprise if you expected it to follow along.

    Args:
        prompt: User prompt to process
        thread_id: Conversation to continue. Required (with a configured
            checkpointer) for this run to see earlier turns.
        metadata: Additional metadata for tools

    Returns:
        AgentResult with final message and state
    """
    started_at = datetime.now(UTC)
    stop_reason: StopReason = "complete"
    final_message: str = ""
    tool_errors = 0

    callback = self.config.callback_handler

    # Only forward ``model_kwargs`` when the caller supplied it, so an
    # existing subclass or test double that overrides ``run`` with the
    # previous signature keeps working.
    run_kwargs: dict[str, Any] = {"thread_id": thread_id, "metadata": metadata}
    if model_kwargs is not None:
        run_kwargs["model_kwargs"] = model_kwargs

    grounding_score: float | None = None
    ungrounded_claims: list[str] = []
    # The runtime already counts these -- ``_reflexion_evals`` and
    # ``_grounding_evals`` -- but they are locals in a generator, so the
    # numbers never left. Counting the events it yields gets the same
    # figures without threading state through the loop.
    reflexion_evaluations = 0
    grounding_evaluations = 0

    # The run this call drives leaves its final state in ``slot`` — never
    # on the shared agent, where a concurrent run on another thread would
    # overwrite it between the last event and this read.
    slot = ResultSlot(owner=self)
    slot_token = ARUN_RESULT_SLOT.set(slot)
    try:
        async with _closing(self.run(prompt, **run_kwargs)) as events:
            async for event in events:
                # Fire callback if set
                if callback is not None:
                    callback(event)

                if isinstance(event, TerminateEvent):
                    stop_reason = _normalize_stop_reason(event.reason)
                    final_message = event.final_message or ""
                elif isinstance(event, ToolCompleteEvent):
                    if event.error:
                        tool_errors += 1
                elif isinstance(event, ReflectEvent):
                    reflexion_evaluations += 1
                elif isinstance(event, GroundingEvent):
                    grounding_evaluations += 1
                    # The grounding loop already ran and emitted its verdict;
                    # it simply never reached the result. Keep the last one:
                    # with `max_replans` the answer is re-grounded after each
                    # replan, and the score that describes the answer being
                    # returned is the final one.
                    grounding_score = event.score
                    ungrounded_claims = list(event.ungrounded_claims)
    finally:
        ARUN_RESULT_SLOT.reset(slot_token)

    # The actual final state of THIS run. ``None`` only when ``run`` is
    # overridden by something that never goes through the runtime loop.
    state = slot.state
    if state is None:
        state = await self._create_initial_state(prompt, thread_id, metadata)
        if final_message:
            state = state.with_message(Message.assistant(final_message))

    # Structured-output coercion (no-op when output_schema is unset).
    parsed_obj = None
    parse_error_msg = None
    structured_message = final_message
    if self.config.output_schema is not None:
        parsed_obj, parse_error_msg, state = await self._structure_output(
            state, final_message or ""
        )
        if parsed_obj is not None:
            # Replace ``message`` with the canonical JSON form so callers
            # using ``result.message`` still see a schema-valid string.
            structured_message = parsed_obj.model_dump_json()

    # Run GSAR judgment when configured. Single-pass v1: judge
    # the final answer, surface the result on AgentResult.
    # Full Algorithm-1 outer loop (regenerate / replan) lives in
    # tulip.reasoning.gsar_evaluator and can be wired
    # explicitly when the caller wants the loop dynamics.
    gsar_judgment, gsar_score_value, gsar_decision = await self._run_gsar_judgment(
        state, structured_message or final_message
    )

    # ``fail_on_low_score`` was documented as raising GSARValidationError
    # and never read, so an agent configured to refuse un-grounded output
    # shipped it silently -- the one outcome the setting exists to prevent.
    if (
        self.config.gsar is not None
        and self.config.gsar.fail_on_low_score
        and gsar_decision is not None
        and gsar_decision != "proceed"
    ):
        raise GSARValidationError(
            f"GSAR judgment returned {gsar_decision!r}"
            + (f" (score {gsar_score_value:.2f})" if gsar_score_value is not None else "")
            + "; fail_on_low_score is set.",
            decision=gsar_decision,
            score=gsar_score_value,
        )

    elapsed_ms = (datetime.now(UTC) - started_at).total_seconds() * 1000
    metrics = ExecutionMetrics(
        iterations=state.iteration,
        tool_calls=len(state.tool_executions),
        tool_errors=tool_errors,
        total_tokens=state.total_tokens_used,
        prompt_tokens=state.prompt_tokens_used,
        completion_tokens=state.completion_tokens_used,
        cache_creation_input_tokens=state.cache_creation_tokens_used,
        cache_read_input_tokens=state.cache_read_tokens_used,
        cost_usd=state.cost_usd_used if state.priced else None,
        duration_ms=elapsed_ms,
        reflexion_evaluations=reflexion_evaluations,
        grounding_evaluations=grounding_evaluations,
    )

    return AgentResult.from_state(
        state=state,
        stop_reason=stop_reason,
        metrics=metrics,
        started_at=started_at,
        parsed=parsed_obj,
        parse_error=parse_error_msg,
        message=structured_message,
        grounding_score=grounding_score,
        ungrounded_claims=ungrounded_claims,
        gsar_judgment=gsar_judgment,
        gsar_score=gsar_score_value,
        gsar_decision=gsar_decision,
    )

drain_memory async

drain_memory

Wait for the memory manager's background extractions to finish.

With LLMMemoryManager(extract_mode="background") extraction runs after a turn's final event; await this at graceful shutdown (and in tests) so none is lost — bound it with asyncio.timeout(...) if needed. A no-op without a memory manager or in "inline" mode. :meth:run_sync calls it before returning.

Source code in .sdk/src/tulip/agent/agent.py

async def drain_memory(self) -> None:
    """Wait for the memory manager's background extractions to finish.

    With ``LLMMemoryManager(extract_mode="background")`` extraction runs
    after a turn's final event; await this at graceful shutdown (and in
    tests) so none is lost — bound it with ``asyncio.timeout(...)`` if
    needed. A no-op without a memory manager or in ``"inline"`` mode.
    :meth:`run_sync` calls it before returning.
    """
    manager = self._memory_manager or self.config.memory_manager
    drain = getattr(manager, "drain", None)
    if drain is not None:
        await drain()

run_sync

run_sync

Run the agent synchronously.

Blocking wrapper around :meth:arun: it drives self.arun(...) on a fresh event loop (or a worker thread when a loop is already running) and closes the model / checkpointer clients bound to that loop before returning. In thread-less environments (Pyodide / WASM) this cannot work — call await self.arun(...) directly there.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt to process	required
thread_id	str | None	Optional thread ID for checkpointing	None
metadata	dict[str, Any] | None	Additional metadata for tools	None

Returns:

Type	Description
AgentResult	AgentResult with final message and state

Source code in .sdk/src/tulip/agent/agent.py

def run_sync(
    self,
    prompt: str,
    *,
    thread_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    model_kwargs: dict[str, Any] | None = None,
) -> AgentResult:
    """
    Run the agent synchronously.

    Blocking wrapper around :meth:`arun`: it drives ``self.arun(...)`` on
    a fresh event loop (or a worker thread when a loop is already
    running) and closes the model / checkpointer clients bound to that
    loop before returning. In thread-less environments (Pyodide / WASM)
    this cannot work — call ``await self.arun(...)`` directly there.

    Args:
        prompt: User prompt to process
        thread_id: Optional thread ID for checkpointing
        metadata: Additional metadata for tools

    Returns:
        AgentResult with final message and state
    """

    async def _run_and_close_clients() -> AgentResult:
        # Wrap ``self.arun`` so any model-level httpx client is shut down
        # *inside* this asyncio.run loop. Otherwise the client's
        # connections remain bound to the loop we're about to close;
        # when ``run_sync`` is called again, the next ``asyncio.run``
        # opens a fresh loop and the old client's ``__del__`` tries
        # to ``aclose`` against the now-closed loop, raising
        # ``RuntimeError: Event loop is closed``.
        try:
            return await self.arun(
                prompt, thread_id=thread_id, metadata=metadata, model_kwargs=model_kwargs
            )
        finally:
            # Background memory extraction is bound to this loop, which
            # closes on return: finish it here rather than lose it.
            try:
                await self.drain_memory()
            except Exception:  # noqa: BLE001 — cleanup must never mask a real error from arun()
                pass

            close = getattr(self.model, "close", None)
            if close is not None:
                try:
                    await close()
                except Exception:  # noqa: BLE001 — cleanup must never mask a real error from arun()
                    pass

            # Same reasoning for the checkpointer's connection pool.
            # A driver connection pool is bound to the asyncio loop
            # that created it. Closing it here drains the connections
            # *inside* this loop. Skipping this step means the next
            # ``run_sync`` opens a fresh loop with the old pool still
            # holding TCP handles from the dead loop — which surfaces
            # as connection-reset errors on the next save.
            ckpt = getattr(getattr(self, "config", None), "checkpointer", None)
            ckpt_close = getattr(ckpt, "close", None) if ckpt is not None else None
            if ckpt_close is not None:
                try:
                    await ckpt_close()
                except Exception:  # noqa: BLE001 — cleanup must never mask arun() errors
                    pass

            # Drain any background tasks the SDK spawned (httpx's TLS
            # teardown schedules ``loop.call_soon`` callbacks via
            # anyio that fire after ``client.close()`` returns). If
            # we don't await them, the loop closes mid-flight and the
            # callbacks raise "Event loop is closed" on the asyncio
            # default exception handler — visible in stderr as
            # "Task exception was never retrieved".
            try:
                pending = [
                    t
                    for t in asyncio.all_tasks()
                    if t is not asyncio.current_task() and not t.done()
                ]
                if pending:
                    await asyncio.wait(pending, timeout=2.0)
            except Exception:  # noqa: BLE001 — best-effort drain; never block teardown
                pass

    try:
        asyncio.get_running_loop()
    except RuntimeError:
        # No running loop, create a new one
        return asyncio.run(_run_and_close_clients())
    else:
        # There's a running loop, run in a thread to avoid nesting
        import concurrent.futures

        with concurrent.futures.ThreadPoolExecutor(max_workers=1) as executor:
            future = executor.submit(asyncio.run, _run_and_close_clients())
            return future.result()

invoke

invoke

Invoke the agent (alias for run_sync).

Parameters:

Name	Type	Description	Default
prompt	str	User prompt to process	required
thread_id	str | None	Optional thread ID for checkpointing	None
metadata	dict[str, Any] | None	Additional metadata for tools	None

Returns:

Type	Description
AgentResult	AgentResult with final message and state

Source code in .sdk/src/tulip/agent/agent.py

def invoke(
    self,
    prompt: str,
    *,
    thread_id: str | None = None,
    metadata: dict[str, Any] | None = None,
) -> AgentResult:
    """
    Invoke the agent (alias for run_sync).

    Args:
        prompt: User prompt to process
        thread_id: Optional thread ID for checkpointing
        metadata: Additional metadata for tools

    Returns:
        AgentResult with final message and state
    """
    return self.run_sync(prompt, thread_id=thread_id, metadata=metadata)

cancel

cancel

Cancel running agent work from any thread or task.

Sets a signal the agent loop checks at each iteration; the run stops gracefully with stop_reason="cancelled".

• cancel(thread_id="t-42") stops only the in-flight run(s) on that thread — the form a multi-user server wants, where one Agent serves every conversation.
• cancel() (no argument) keeps the pre-2.17 meaning: stop every in-flight run of this agent, and if none is running yet, the next run to start.

Thread-safe — can be called from any thread while the agent is running.

Parameters:

Name	Type	Description	Default
thread_id	str | None	Only cancel runs on this conversation.	None

Returns:

Type	Description
int	How many in-flight runs were signalled (a no-argument call with
int	nothing running still arms the signal for the next run and
int	returns 0).

Example

import threading

def run_agent(): result = agent.run_sync("Long task...") print(result.stop_reason) # "cancelled"

t = threading.Thread(target=run_agent) t.start() time.sleep(5) agent.cancel() # Stop from main thread t.join()

Source code in .sdk/src/tulip/agent/agent.py

def cancel(self, thread_id: str | None = None) -> int:
    """Cancel running agent work from any thread or task.

    Sets a signal the agent loop checks at each iteration; the run stops
    gracefully with ``stop_reason="cancelled"``.

    * ``cancel(thread_id="t-42")`` stops only the in-flight run(s) on that
      thread — the form a multi-user server wants, where one Agent serves
      every conversation.
    * ``cancel()`` (no argument) keeps the pre-2.17 meaning: stop every
      in-flight run of this agent, and if none is running yet, the next run
      to start.

    Thread-safe — can be called from any thread while the agent is running.

    Args:
        thread_id: Only cancel runs on this conversation.

    Returns:
        How many in-flight runs were signalled (a no-argument call with
        nothing running still arms the signal for the next run and
        returns 0).

    Example:
        import threading

        def run_agent():
            result = agent.run_sync("Long task...")
            print(result.stop_reason)  # "cancelled"

        t = threading.Thread(target=run_agent)
        t.start()
        time.sleep(5)
        agent.cancel()  # Stop from main thread
        t.join()
    """
    with self._runs_lock:
        runs = list(self._active_runs.values())
    if thread_id is not None:
        targets = [rc for rc in runs if rc.thread_id == thread_id]
    else:
        if self._cancel_signal is None:
            self._cancel_signal = threading.Event()
        self._cancel_signal.set()
        targets = runs
    for rc in targets:
        rc.cancel.set()
    return len(targets)

pending_interrupts

pending_interrupts

Thread ids that have a paused run held in this process's memory.

A thread paused in another process (or before a restart) is not listed; resume(thread_id=...) still rehydrates it from the checkpointer.

Source code in .sdk/src/tulip/agent/agent.py

def pending_interrupts(self) -> list[str | None]:
    """Thread ids that have a paused run held in this process's memory.

    A thread paused in another process (or before a restart) is not
    listed; ``resume(thread_id=...)`` still rehydrates it from the
    checkpointer.
    """
    return list(self._interrupts)

run_subagent async

run_subagent

Run an isolated child agent and return its :class:SubagentResult.

Thin binding of :func:tulip.agent.subagent.run_subagent to this agent: the child defaults to this agent's model, and this agent's :meth:cancel stops the child — whether the call comes from one of this agent's tool bodies mid-run or from a harness between runs.

Isolation is the point: the child gets a fresh conversation, its own system prompt, and ONLY the tools in tools — never this agent's toolset by inheritance. When called from a tool body during a run, the child's token usage folds into this run's counters, so token_budget and TerminateEvent.usage keep counting delegated spend. See :mod:tulip.agent.subagent for the full contract and a harness-side task-tool example.

Parameters:

Name	Type	Description	Default
prompt	str	The delegated task.	required
tools	list[Tool] | None	Explicit tool allowlist for the child (default: none).	None
system_prompt	str	The child's system prompt.	'You are a focused subagent. Complete the delegated task.'
model	str | Any | None	Child model override; defaults to this agent's model.	None
name	str	Attribution label stamped on the child's events.	'subagent'
max_iterations	int	The child's iteration cap.	10
hooks	list[Any] | None	Lifecycle hooks for the child (policy attaches here).	None
on_event	Callable[[TulipEvent], Awaitable[None] | None] | None	Callback receiving every child event (sync or async).	None
**agent_kwargs	Any	Further AgentConfig fields for the child.	{}

Returns:

Type	Description
SubagentResult	SubagentResult with the child's text, usage, iterations, and
SubagentResult	stop reason.

Source code in .sdk/src/tulip/agent/agent.py

async def run_subagent(
    self,
    prompt: str,
    *,
    tools: list[Tool] | None = None,
    system_prompt: str = "You are a focused subagent. Complete the delegated task.",
    model: str | Any | None = None,
    name: str = "subagent",
    max_iterations: int = 10,
    hooks: list[Any] | None = None,
    on_event: Callable[[TulipEvent], Awaitable[None] | None] | None = None,
    **agent_kwargs: Any,
) -> SubagentResult:
    """Run an isolated child agent and return its :class:`SubagentResult`.

    Thin binding of :func:`tulip.agent.subagent.run_subagent` to this
    agent: the child defaults to this agent's model, and this agent's
    :meth:`cancel` stops the child — whether the call comes from one of
    this agent's tool bodies mid-run or from a harness between runs.

    Isolation is the point: the child gets a fresh conversation, its own
    system prompt, and ONLY the tools in ``tools`` — never this agent's
    toolset by inheritance. When called from a tool body during a run,
    the child's token usage folds into this run's counters, so
    ``token_budget`` and ``TerminateEvent.usage`` keep counting
    delegated spend. See :mod:`tulip.agent.subagent` for the full
    contract and a harness-side ``task``-tool example.

    Args:
        prompt: The delegated task.
        tools: Explicit tool allowlist for the child (default: none).
        system_prompt: The child's system prompt.
        model: Child model override; defaults to this agent's model.
        name: Attribution label stamped on the child's events.
        max_iterations: The child's iteration cap.
        hooks: Lifecycle hooks for the child (policy attaches here).
        on_event: Callback receiving every child event (sync or async).
        **agent_kwargs: Further ``AgentConfig`` fields for the child.

    Returns:
        SubagentResult with the child's text, usage, iterations, and
        stop reason.
    """
    from tulip.agent.subagent import run_subagent

    if self._cancel_signal is None:
        self._cancel_signal = threading.Event()
    return await run_subagent(
        prompt,
        model=model if model is not None else self.model,
        tools=tools,
        system_prompt=system_prompt,
        name=name,
        max_iterations=max_iterations,
        hooks=hooks,
        on_event=on_event,
        cancel_signal=self._cancel_signal,
        **agent_kwargs,
    )

as_tool

as_tool

Wrap this agent as a Tool for use by another agent.

The returned tool accepts a prompt string and returns the agent's final response. This enables agent delegation — a parent agent can call a sub-agent as if it were any other tool.

Parameters:

Name	Type	Description	Default
name	str | None	Tool name (defaults to agent_id or "sub_agent")	None
description	str | None	Tool description (defaults to system prompt excerpt)	None

Returns:

Type	Description
Tool	A Tool that runs this agent when called

Example

researcher = Agent( ... model=model, tools=[search], system_prompt="You research topics." ... ) writer = Agent(model=model, tools=[researcher.as_tool("research")]) result = writer.run_sync("Write about quantum computing")

Source code in .sdk/src/tulip/agent/agent.py

def as_tool(
    self,
    name: str | None = None,
    description: str | None = None,
) -> Tool:
    """
    Wrap this agent as a Tool for use by another agent.

    The returned tool accepts a prompt string and returns the agent's
    final response. This enables agent delegation — a parent agent
    can call a sub-agent as if it were any other tool.

    Args:
        name: Tool name (defaults to agent_id or "sub_agent")
        description: Tool description (defaults to system prompt excerpt)

    Returns:
        A Tool that runs this agent when called

    Example:
        >>> researcher = Agent(
        ...     model=model, tools=[search], system_prompt="You research topics."
        ... )
        >>> writer = Agent(model=model, tools=[researcher.as_tool("research")])
        >>> result = writer.run_sync("Write about quantum computing")
    """
    from tulip.tools.decorator import tool as tool_decorator

    agent = self
    tool_name = name or self.config.agent_id or "sub_agent"
    tool_desc = description or (
        "Delegate a task to a sub-agent. "
        "The sub-agent has its own tools and will work independently "
        "to answer your request. Send a clear, specific prompt."
    )

    @tool_decorator(name=tool_name, description=tool_desc)
    def agent_tool(prompt: str) -> str:
        """Run the sub-agent with the given prompt and return its response.

        Args:
            prompt: The task or question to delegate to the sub-agent

        Returns:
            The sub-agent's final response
        """
        result = agent.run_sync(prompt)
        if result.success:
            return result.message
        return f"Sub-agent finished with status '{result.stop_reason}': {result.message}"

    return agent_tool

resume async

resume

Resume agent execution after an interrupt.

When a tool calls ask_user() and the agent yields an InterruptEvent, call this method with the user's response to continue execution.

Resume is keyed strictly by thread. Paused runs are held in memory per thread_id; resume(thread_id="A") only ever continues thread A — from memory when this process paused it, otherwise rehydrated from the configured checkpointer (a fresh process: the pod that paused is gone). It never picks up another thread's pause. Without thread_id the call is accepted only when exactly one run is paused in memory (the single-conversation back-compat form); with several paused it raises rather than guess.

A resumed run continues the SAME turn: its iteration count and budgets carry on from the pause.

perform_dangling closes the gap the 2.11.1 fold left open for approval holds. The fold answers the dangling call with the human's verdict TEXT — the model reads "approve" as the call's result and, reasonably, never re-issues a call that already returned, so nothing ever performs the action (measured live; see FINDING-an-approved-action-can-silently-not-happen.md for the first half of this story). With perform_dangling=True and a dangling call that is not ask_user, the runtime RE-INVOKES that call — same tool, same arguments, through the normal executor — and folds the REAL result instead. A gated tool decides for itself under the caller's primed decision: approve executes, deny refuses, and either way the model sees what actually happened. ask_user and the no-dangling-call case are untouched, so answers behave exactly as before regardless of the flag.

Parameters:

Name	Type	Description	Default
response	str	The user's response to the interrupt question	required
thread_id	str | None	The paused thread to continue. Resumes that thread's in-memory interrupt, or rehydrates it from the checkpointer.	None
perform_dangling	bool	Re-invoke a dangling non-ask_user call and fold its real result, instead of folding response as text	False
metadata	dict[str, Any] | None	Invocation metadata for the resumed segment (what tools see as ctx.invocation_metadata and hooks as event.run.metadata). Defaults to the paused run's metadata (in memory) or the thread's checkpointed metadata. Ephemeral keys (mcp_headers) are never checkpointed: an in-process resume reuses the paused run's, a cross-process resume that needs them passes them here again.	None

Raises:

Type	Description
ApprovalPendingError	perform_dangling=True and the held call interrupted again — its approval has not been decided yet. The thread stays paused and nothing is folded or checkpointed.
RuntimeError	Nothing to resume (no paused run for the thread and no checkpoint), or thread_id omitted while several runs are paused.

Yields:

Type	Description
AsyncGenerator[TulipEvent, None]	TulipEvent instances for the remaining execution

Example

async for event in agent.run("Build an app"): ... if isinstance(event, InterruptEvent): ... answer = input(event.question) ... async for event in agent.resume(answer): ... handle(event)

Source code in .sdk/src/tulip/agent/agent.py

async def resume(
    self,
    response: str,
    *,
    thread_id: str | None = None,
    perform_dangling: bool = False,
    metadata: dict[str, Any] | None = None,
) -> AsyncGenerator[TulipEvent, None]:
    """
    Resume agent execution after an interrupt.

    When a tool calls ask_user() and the agent yields an InterruptEvent,
    call this method with the user's response to continue execution.

    **Resume is keyed strictly by thread.** Paused runs are held in memory
    per ``thread_id``; ``resume(thread_id="A")`` only ever continues
    thread A — from memory when this process paused it, otherwise
    rehydrated from the configured checkpointer (a fresh process: the pod
    that paused is gone). It never picks up another thread's pause.
    Without ``thread_id`` the call is accepted only when exactly one run
    is paused in memory (the single-conversation back-compat form); with
    several paused it raises rather than guess.

    A resumed run continues the SAME turn: its iteration count and budgets
    carry on from the pause.

    ``perform_dangling`` closes the gap the 2.11.1 fold left open for
    approval holds. The fold answers the dangling call with the human's
    verdict TEXT — the model reads "approve" as the call's result and,
    reasonably, never re-issues a call that already returned, so nothing
    ever performs the action (measured live; see
    FINDING-an-approved-action-can-silently-not-happen.md for the first
    half of this story). With ``perform_dangling=True`` and a dangling
    call that is not ``ask_user``, the runtime RE-INVOKES that call —
    same tool, same arguments, through the normal executor — and folds
    the REAL result instead. A gated tool decides for itself under the
    caller's primed decision: approve executes, deny refuses, and either
    way the model sees what actually happened. ``ask_user`` and the
    no-dangling-call case are untouched, so answers behave exactly as
    before regardless of the flag.

    Args:
        response: The user's response to the interrupt question
        thread_id: The paused thread to continue. Resumes that thread's
            in-memory interrupt, or rehydrates it from the checkpointer.
        perform_dangling: Re-invoke a dangling non-``ask_user`` call and
            fold its real result, instead of folding ``response`` as text
        metadata: Invocation metadata for the resumed segment (what tools
            see as ``ctx.invocation_metadata`` and hooks as
            ``event.run.metadata``). Defaults to the paused run's metadata
            (in memory) or the thread's checkpointed metadata. Ephemeral
            keys (``mcp_headers``) are never checkpointed: an in-process
            resume reuses the paused run's, a cross-process resume that
            needs them passes them here again.

    Raises:
        ApprovalPendingError: ``perform_dangling=True`` and the held call
            interrupted again — its approval has not been decided yet. The
            thread stays paused and nothing is folded or checkpointed.
        RuntimeError: Nothing to resume (no paused run for the thread and
            no checkpoint), or ``thread_id`` omitted while several runs
            are paused.

    Yields:
        TulipEvent instances for the remaining execution

    Example:
        >>> async for event in agent.run("Build an app"):
        ...     if isinstance(event, InterruptEvent):
        ...         answer = input(event.question)
        ...         async for event in agent.resume(answer):
        ...             handle(event)
    """
    pending = self._take_interrupt(thread_id)
    if pending is not None:
        state = pending.state
        # The prompt of the paused run; the state already has the history.
        prompt = pending.prompt
        thread_id = pending.thread_id
        if metadata is not None:
            run_metadata, ephemeral = self._split_run_metadata(metadata)
        else:
            run_metadata, ephemeral = pending.metadata, dict(pending.ephemeral)
    else:
        # Rehydrate: no in-memory interrupt for this thread, so reload the
        # paused state from the checkpointer (the cross-process path).
        if thread_id is None or self.config.checkpointer is None:
            raise RuntimeError(
                "No interrupt to resume from. Call run() first, or pass "
                "thread_id with a configured checkpointer to rehydrate."
            )
        loaded = await self.config.checkpointer.load(thread_id)
        if loaded is None:
            raise RuntimeError(f"No checkpoint found for thread {thread_id!r} to resume from.")
        state = loaded
        prompt = ""
        # A checkpoint never holds ephemeral metadata (``mcp_headers``):
        # a cross-process resume that needs it passes ``metadata=``.
        run_metadata, ephemeral = self._split_run_metadata(
            metadata if metadata is not None else dict(loaded.metadata)
        )
        # Checkpoints never carry the (ephemeral) memory block, so the
        # rest of this turn gets it re-injected, as the in-memory path has.
        # (A fresh process has not initialised the agent yet.)
        memory_manager = self._memory_manager or self.config.memory_manager
        if memory_manager is not None:
            state = await memory_manager.on_session_start(state)

    # This resumed segment is a run of its own: cancellable by thread,
    # visible to hooks as ``event.run``, isolated from concurrent runs.
    rc = self._begin_run(state, prompt, thread_id, run_metadata, ephemeral=ephemeral)
    try:
        folded_events: list[TulipEvent] = []
        state = await self._fold_resume_response(
            state, response, rc, perform_dangling, pending, folded_events
        )
    except BaseException:
        self._end_run(rc, state)
        raise
    for folded_event in folded_events:
        yield folded_event

    # Continue execution from the interrupted state
    async with contextlib.aclosing(
        self._run_from_state(state, prompt, thread_id, run_metadata, _run=rc)
    ) as segment:
        async for event in segment:
            yield event

continue_turn async

continue_turn

Continue a turn that stopped before it finished, from its checkpoint.

A process killed mid-turn (a deploy, an OOM, a lost pod) leaves the thread's latest checkpoint inside the turn — with checkpoint_every_n_iterations set, at the last iteration whose tool results were saved. run() on that thread would start a NEW turn with a new user message; this continues the SAME turn instead: no user message is added, iteration count and budgets carry on, and every call whose result is in the checkpoint stays done — the model is called next, with those results in front of it.

A call the checkpoint holds without a result (the process stopped while it ran) is answered with an error saying its outcome is unknown, never re-run: re-running a side effect the first attempt may already have performed is the model's decision to make, with that error in front of it.

A thread paused on an interrupt in this process is answered with :meth:resume, which folds the human's response; continue_turn refuses it.

Parameters:

Name	Type	Description	Default
thread_id	str	The thread whose turn to continue.	required
metadata	dict[str, Any] | None	Invocation metadata for the continued segment. Defaults to the thread's checkpointed metadata; ephemeral keys (mcp_headers) are never checkpointed, so a segment that needs them passes them here again.	None
model_kwargs	dict[str, Any] | None	Per-call model parameters, as for :meth:run.	None

Raises:

Type	Description
RuntimeError	No checkpointer, no checkpoint for the thread, the thread is paused on an in-memory interrupt, or its last turn already finished with an answer (start a new one with run()).

Yields:

Type	Description
AsyncGenerator[TulipEvent, None]	TulipEvent instances for the rest of the turn.

Source code in .sdk/src/tulip/agent/agent.py

async def continue_turn(
    self,
    thread_id: str,
    *,
    metadata: dict[str, Any] | None = None,
    model_kwargs: dict[str, Any] | None = None,
) -> AsyncGenerator[TulipEvent, None]:
    """Continue a turn that stopped before it finished, from its checkpoint.

    A process killed mid-turn (a deploy, an OOM, a lost pod) leaves the
    thread's latest checkpoint inside the turn — with
    ``checkpoint_every_n_iterations`` set, at the last iteration whose
    tool results were saved. ``run()`` on that thread would start a NEW
    turn with a new user message; this continues the SAME turn instead:
    no user message is added, iteration count and budgets carry on, and
    every call whose result is in the checkpoint stays done — the model
    is called next, with those results in front of it.

    A call the checkpoint holds without a result (the process stopped
    while it ran) is answered with an error saying its outcome is
    unknown, never re-run: re-running a side effect the first attempt
    may already have performed is the model's decision to make, with
    that error in front of it.

    A thread paused on an interrupt in this process is answered with
    :meth:`resume`, which folds the human's response; ``continue_turn``
    refuses it.

    Args:
        thread_id: The thread whose turn to continue.
        metadata: Invocation metadata for the continued segment. Defaults
            to the thread's checkpointed metadata; ephemeral keys
            (``mcp_headers``) are never checkpointed, so a segment that
            needs them passes them here again.
        model_kwargs: Per-call model parameters, as for :meth:`run`.

    Raises:
        RuntimeError: No checkpointer, no checkpoint for the thread, the
            thread is paused on an in-memory interrupt, or its last turn
            already finished with an answer (start a new one with
            ``run()``).

    Yields:
        TulipEvent instances for the rest of the turn.
    """
    checkpointer = self.config.checkpointer
    if checkpointer is None:
        raise RuntimeError("continue_turn needs a checkpointer on the agent")
    if thread_id in self._interrupts:
        raise RuntimeError(
            f"Thread {thread_id!r} is paused on an interrupt; answer it with resume()."
        )
    loaded = await checkpointer.load(thread_id)
    if loaded is None:
        raise RuntimeError(f"No checkpoint found for thread {thread_id!r} to continue.")
    if _turn_finished(loaded):
        raise RuntimeError(
            f"The last turn on thread {thread_id!r} already finished; "
            "call run() to start a new one."
        )
    self._initialize()
    state = close_unfinished_calls(loaded)
    run_metadata, ephemeral = self._split_run_metadata(
        metadata if metadata is not None else dict(loaded.metadata)
    )
    # Checkpoints never carry the (ephemeral) memory block; the rest of
    # the turn gets it re-injected, as a cross-process resume does.
    if self._memory_manager is not None:
        state = await self._memory_manager.on_session_start(state)

    rc = self._begin_run(state, "", thread_id, run_metadata, ephemeral=ephemeral)
    async with contextlib.aclosing(
        self._run_from_state(state, "", thread_id, run_metadata, model_kwargs, _run=rc)
    ) as segment:
        async for event in segment:
            yield event

get_state async

get_state

The thread's state at checkpoint_id, or its latest.

Raises:

Type	Description
RuntimeError	The agent has no checkpointer.
LookupError	No such thread or checkpoint.

Source code in .sdk/src/tulip/agent/agent.py

async def get_state(self, thread_id: str, checkpoint_id: str | None = None) -> AgentState:
    """The thread's state at ``checkpoint_id``, or its latest.

    Raises:
        RuntimeError: The agent has no checkpointer.
        LookupError: No such thread or checkpoint.
    """
    state = await self._require_checkpointer().load(thread_id, checkpoint_id)
    if state is None:
        where = f"checkpoint {checkpoint_id!r} of " if checkpoint_id else ""
        raise LookupError(f"no {where}thread {thread_id!r}")
    loaded: AgentState = state
    return loaded

get_state_history async

get_state_history

(checkpoint_id, state) pairs for the thread, newest first.

Source code in .sdk/src/tulip/agent/agent.py

async def get_state_history(
    self, thread_id: str, limit: int = 10
) -> list[tuple[str, AgentState]]:
    """``(checkpoint_id, state)`` pairs for the thread, newest first."""
    checkpointer = self._require_checkpointer()
    history: list[tuple[str, AgentState]] = []
    for checkpoint_id in await checkpointer.list_checkpoints(thread_id, limit=limit):
        state = await checkpointer.load(thread_id, checkpoint_id)
        if state is not None:
            history.append((checkpoint_id, state))
    return history

update_state async

update_state

Write a new checkpoint: the given one (or the latest) plus these changes.

The source checkpoint is never modified, so history stays intact and the edit can itself be inspected or forked. The next run or resume on the thread continues from the new checkpoint.

Returns:

Type	Description
str	The new checkpoint's id.

Source code in .sdk/src/tulip/agent/agent.py

async def update_state(
    self,
    thread_id: str,
    *,
    messages: list[Message] | None = None,
    metadata: dict[str, Any] | None = None,
    checkpoint_id: str | None = None,
) -> str:
    """Write a new checkpoint: the given one (or the latest) plus these changes.

    The source checkpoint is never modified, so history stays intact and
    the edit can itself be inspected or forked. The next ``run`` or
    ``resume`` on the thread continues from the new checkpoint.

    Returns:
        The new checkpoint's id.
    """
    state = await self.get_state(thread_id, checkpoint_id)
    for message in messages or []:
        state = state.with_message(message)
    for key, value in (metadata or {}).items():
        state = state.with_metadata(key, value)
    new_id: str = await self._require_checkpointer().save(
        state,
        thread_id,
        metadata={"source": "update_state", "parent_checkpoint_id": checkpoint_id},
    )
    return new_id

fork async

fork

Start a new thread from the given checkpoint (or the latest).

The new thread carries the conversation up to that point and nothing after it; running on it never changes the parent thread.

Returns:

Type	Description
str	The new thread's id.

Source code in .sdk/src/tulip/agent/agent.py

async def fork(
    self,
    thread_id: str,
    checkpoint_id: str | None = None,
    *,
    new_thread_id: str | None = None,
) -> str:
    """Start a new thread from the given checkpoint (or the latest).

    The new thread carries the conversation up to that point and nothing
    after it; running on it never changes the parent thread.

    Returns:
        The new thread's id.
    """
    from uuid import uuid4  # noqa: PLC0415

    state = await self.get_state(thread_id, checkpoint_id)
    forked = new_thread_id or f"{thread_id}-fork-{uuid4().hex[:8]}"
    if forked == thread_id:
        raise ValueError("a fork needs a thread id different from its parent")
    await self._require_checkpointer().save(
        state,
        forked,
        metadata={"forked_from": thread_id, "parent_checkpoint_id": checkpoint_id},
    )
    return forked

add_tool

add_tool

Register a tool on this agent after construction.

Tulip compiles config.tools into the runtime ToolRegistry once, inside __init__ (via :func:tulip.agent.initializer. initialize_agent). Mutating self.config.tools directly after that point is a silent no-op — the model never sees the added tool because the registry has already been built.

Use this method (or :meth:add_tools) when you want to compose a specialist fleet at runtime: build each specialist, wrap it via Agent.as_tool(...), and attach the wrappers to the orchestrator.

The tool is also appended to self.config.tools so that a subsequent re-initialisation (e.g. after a config-driven clone) sees the same shape.

Raises:

Type	Description
TypeError	if tool is not a :class:tulip.tools.Tool instance. Callable functions must be wrapped with the :func:@tool decorator first.
ValueError	if a tool with the same name is already registered (propagated from :meth:ToolRegistry.register).

Source code in .sdk/src/tulip/agent/agent.py

def add_tool(self, tool: Tool) -> None:
    """Register a tool on this agent after construction.

    Tulip compiles ``config.tools`` into the runtime ``ToolRegistry``
    once, inside ``__init__`` (via :func:`tulip.agent.initializer.
    initialize_agent`). Mutating ``self.config.tools`` directly after
    that point is a silent no-op — the model never sees the added
    tool because the registry has already been built.

    Use this method (or :meth:`add_tools`) when you want to compose a
    specialist fleet at runtime: build each specialist, wrap it via
    ``Agent.as_tool(...)``, and attach the wrappers to the
    orchestrator.

    The tool is also appended to ``self.config.tools`` so that a
    subsequent re-initialisation (e.g. after a config-driven
    clone) sees the same shape.

    Raises:
        TypeError: if ``tool`` is not a :class:`tulip.tools.Tool`
            instance. Callable functions must be wrapped with the
            :func:`@tool` decorator first.
        ValueError: if a tool with the same ``name`` is already
            registered (propagated from
            :meth:`ToolRegistry.register`).
    """
    if not isinstance(tool, Tool):
        raise TypeError(
            f"Expected Tool instance (use @tool to wrap a function), got {type(tool)}"
        )
    self._initialize()
    self._tool_registry.register(tool)
    # Mirror into config so a re-initialisation reconstructs the
    # same surface. ``config.tools`` is a list[Any] by Pydantic
    # declaration, so we mutate in place rather than reassigning.
    self.config.tools.append(tool)

add_tools

add_tools

Register multiple tools at once.

Equivalent to calling :meth:add_tool for each entry. If any single registration fails (wrong type, duplicate name), the whole call fails: tools registered before the failing one remain in the registry. Validate inputs ahead of time when atomic behaviour is required.

Source code in .sdk/src/tulip/agent/agent.py

def add_tools(self, tools: list[Tool]) -> None:
    """Register multiple tools at once.

    Equivalent to calling :meth:`add_tool` for each entry. If any
    single registration fails (wrong type, duplicate name), the
    whole call fails: tools registered before the failing one
    remain in the registry. Validate inputs ahead of time when
    atomic behaviour is required.
    """
    for t in tools:
        self.add_tool(t)

run async

run

Run the agent with streaming events.

Parameters:

Name	Type	Description	Default
prompt	str	User prompt to process	required
thread_id	str | None	Optional thread ID for checkpointing	None
metadata	dict[str, Any] | None	Additional metadata for tools	None
model_kwargs	dict[str, Any] | None	Per-call parameters forwarded to the model for this run — tool_choice, logprobs, extra_body and the rest of the provider's request surface. Model configuration is fixed for a model's lifetime, which is the wrong shape for anything that has to vary per run.	None
stream_tokens	bool	Also yield :class:ModelChunkEvent as the model produces them, so a UI can render text and chain-of-thought as they arrive instead of waiting for the turn to finish. Tool and termination events are unchanged. Off by default: it changes which event types a consumer sees.	False

Yields:

Type	Description
AsyncGenerator[TulipEvent, None]	TulipEvent instances for each step

Source code in .sdk/src/tulip/agent/runtime_loop.py

@_bus_bridge
async def run(
    self,
    prompt: str,
    *,
    thread_id: str | None = None,
    metadata: dict[str, Any] | None = None,
    model_kwargs: dict[str, Any] | None = None,
    stream_tokens: bool = False,
) -> AsyncGenerator[TulipEvent, None]:
    """
    Run the agent with streaming events.

    Args:
        prompt: User prompt to process
        thread_id: Optional thread ID for checkpointing
        metadata: Additional metadata for tools
        model_kwargs: Per-call parameters forwarded to the model for this
            run — ``tool_choice``, ``logprobs``, ``extra_body`` and the
            rest of the provider's request surface. Model configuration is
            fixed for a model's lifetime, which is the wrong shape for
            anything that has to vary per run.
        stream_tokens: Also yield :class:`ModelChunkEvent` as the model
            produces them, so a UI can render text and chain-of-thought as
            they arrive instead of waiting for the turn to finish. Tool and
            termination events are unchanged. Off by default: it changes
            which event types a consumer sees.

    Yields:
        TulipEvent instances for each step
    """
    self._initialize()
    # Attaching may list tools with the per-run MCP headers, so it sees
    # the full metadata; nothing it is given is persisted.
    await self._attach_mcp_tools(metadata)

    # From here on the run's metadata is split: per-run secrets
    # (``mcp_headers``) ride on the in-memory run context only, never on
    # the state — and so never in a checkpoint, an event or a result.
    metadata, ephemeral = self._split_run_metadata(metadata)

    # Claim the enclosing ``arun``'s result slot before any await that
    # could let a nested run of this agent start in the same context.
    result_slot = claim_result_slot(self)

    # A new user message on a thread supersedes whatever interrupt was
    # parked for it in memory: the checkpoint this run writes is the
    # thread's truth from now on, and resume() must not revive the stale
    # in-memory state over it. (Thread-less runs share one slot and keep
    # the pre-2.17 behaviour: only a new pause replaces it.)
    if thread_id is not None:
        self._interrupts.pop(thread_id, None)

    # Create initial state
    state = await self._create_initial_state(prompt, thread_id, metadata)

    # Everything that belongs to THIS run lives on ``rc``, never on the
    # agent: one Agent instance serves concurrent runs.
    rc = self._begin_run(state, prompt, thread_id, metadata, ephemeral=ephemeral)
    rc.result_slot = result_slot

    # Track metrics
    started_at = datetime.now(UTC)
    # Whether the run has had its one note to converge (``budget_nudge_at``).
    _budget_nudged = False
    _total_tokens = 0
    _tool_calls_count = 0
    _tool_errors_count = 0
    _reflexion_evals = 0
    _grounding_evals = 0
    _last_assistant_content: str | None = None
    _last_no_tool_calls = False
    _verifier_attempts = 0
    # Whether an empty reply has already been sent back this turn.
    _empty_sent_back = False
    # Unfinished text tool calls sent back since the last call that ran.
    _unfinished_sent_back = 0
    # Hold a call's content chunks until we know they are not a draft the
    # verifier will send back (``hold_final_answer_tokens``).
    _hold_tokens = (
        stream_tokens
        and self.config.final_answer_verifier is not None
        and self.config.hold_final_answer_tokens
    )
    _held_chunks: list[Any] = []

    # The composable termination condition was copied and reset for this
    # run by ``_begin_run``: TimeLimit's clock starts now, and no other
    # in-flight run's clock is touched.
    termination = rc.termination

    try:
        # Run hooks: before_invocation
        state = await self._run_before_invocation_hooks(prompt, state)

        # Inject long-term memories, after the turn's prompt.
        if self._memory_manager is not None:
            state = await self._memory_manager.on_session_start(state)
    except BaseException:
        self._end_run(rc, state)
        raise

    # Publish this run as the parent context for subagents spawned from
    # tool bodies: links their cancellation to THIS run's signal (which
    # both cancel(thread_id=...) and a cancel-all set) and gives their
    # usage reports a place to land (folded each iteration). Entered right
    # before the try whose finally exits it.
    _subagent_ctx = enter_parent_run(
        rc.cancel,
        time_budget_seconds=self.config.time_budget_seconds,
        hooks=self.config.hooks,
        observation_pack=self._child_observation_pack(state, rc),
    )

    try:
        # Main ReAct loop
        _open_iteration: int | None = None
        while True:
            # Fold any child-agent usage reported since the last pass, so
            # the budget checks below and every TerminateEvent count
            # delegated spend as spend.
            state = fold_subagent_usage(state)

            # Iteration boundary. The previous iteration is closed here
            # rather than at each exit: this loop has seven ways out, and a
            # hook contract that depends on remembering all of them is one
            # that breaks the next time an eighth is added.
            if _open_iteration is not None:
                await self._run_iteration_end_hooks(_open_iteration, state)
            await self._run_iteration_start_hooks(state.iteration, state)
            _open_iteration = state.iteration

            # Check time budget
            if self.config.time_budget_seconds is not None:
                elapsed = (datetime.now(UTC) - started_at).total_seconds()
                if elapsed >= self.config.time_budget_seconds:
                    yield TerminateEvent(
                        reason="time_budget",
                        iterations_used=state.iteration,
                        final_confidence=state.confidence,
                        usage=_usage_of(state),
                        cost_usd=_cost_of(state),
                        reported_cost_usd=_reported_cost_of(state),
                        total_tool_calls=len(state.tool_executions),
                        final_message=_last_assistant_content,
                    )
                    break

            # Check external cancellation (this run, or cancel-all)
            if self._run_cancelled(rc):
                yield TerminateEvent(
                    reason="cancelled",
                    iterations_used=state.iteration,
                    final_confidence=state.confidence,
                    usage=_usage_of(state),
                    cost_usd=_cost_of(state),
                    reported_cost_usd=_reported_cost_of(state),
                    total_tool_calls=_tool_calls_count,
                    final_message="Agent cancelled by external signal.",
                )
                break

            # User-supplied composable termination condition runs first
            # so MaxIterations(...) | TextMention("DONE") and friends
            # actually fire before the hard-coded fallbacks.
            if termination is not None:
                user_stop, user_reason = termination.check(
                    state,
                    last_message=_last_assistant_content or "",
                    no_tool_calls=_last_no_tool_calls,
                )
                if user_stop:
                    yield TerminateEvent(
                        reason=user_reason or "complete",
                        iterations_used=state.iteration,
                        final_confidence=state.confidence,
                        usage=_usage_of(state),
                        cost_usd=_cost_of(state),
                        reported_cost_usd=_reported_cost_of(state),
                        total_tool_calls=len(state.tool_executions),
                        final_message=_last_assistant_content,
                    )
                    break

            # A loop gets a warning before it can stop the run: the model
            # is told what it is repeating and asked to change approach.
            state, loop_warning = self._warn_tool_loop(state)
            if loop_warning is not None:
                yield loop_warning

            # Check termination conditions
            should_stop, stop_reason = state.should_terminate
            if should_stop and stop_reason:
                if stop_reason == "max_iterations" and state.iteration > 0:
                    # Inject summary request and do one final call WITHOUT tools
                    state = state.with_message(
                        Message.system(
                            "[Iteration Limit Reached]\n"
                            "You have used all available iterations. "
                            "Provide a final summary of your findings and conclusions "
                            "based on the work done so far. Do NOT call any more tools."
                        )
                    )
                    # Call model without tool schemas to force text response.
                    # Use the auxiliary (cheap) model when configured —
                    # this is just a final summary, no need to spend
                    # primary-model budget.
                    messages = list(state.messages)
                    if self._conversation_manager:
                        if hasattr(self._conversation_manager, "async_apply"):
                            messages = await self._conversation_manager.async_apply(messages)
                        else:
                            messages = self._conversation_manager.apply(messages)
                    messages = self._validate_messages(messages)

                    summary_model = self._auxiliary_model or self._model
                    response = await summary_model.complete(
                        messages=messages,
                        tools=None,  # No tools — force text summary
                        temperature=self.config.temperature,
                        max_tokens=self.config.max_tokens,
                    )
                    prompt_toks = response.usage.get("prompt_tokens", 0)
                    completion_toks = response.usage.get("completion_tokens", 0)
                    _total_tokens += prompt_toks + completion_toks
                    state = state.with_response_usage(
                        response.usage, getattr(response, "cost_usd", None)
                    )

                    # A model that is not done answers with its next call in
                    # its own markup, the request having no tools; that is
                    # no summary.
                    written = response.message.content
                    if written and self._is_text_tool_call(written):
                        written = None
                    summary = (
                        written
                        or _last_assistant_content
                        or self._build_fallback_summary(state)
                    )
                    yield TerminateEvent(
                        reason="max_iterations",
                        iterations_used=state.iteration,
                        final_confidence=state.confidence,
                        usage=_usage_of(state),
                        cost_usd=_cost_of(state),
                        reported_cost_usd=_reported_cost_of(state),
                        total_tool_calls=len(state.tool_executions),
                        final_message=summary,
                    )
                    break

                # All other stop reasons: hard stop
                yield TerminateEvent(
                    reason=stop_reason,
                    iterations_used=state.iteration,
                    final_confidence=state.confidence,
                    usage=_usage_of(state),
                    cost_usd=_cost_of(state),
                    reported_cost_usd=_reported_cost_of(state),
                    total_tool_calls=len(state.tool_executions),
                    final_message=_last_assistant_content,
                )
                break

            # Increment iteration
            state = state.next_iteration()

            # Planning: inject plan prompt on first iteration
            if self.config.planning and state.iteration == 1:
                state = state.with_message(
                    Message.system(
                        "[Planning Phase]\n"
                        "Before taking any action, create a step-by-step plan.\n"
                        "Format your plan as a numbered list:\n"
                        "1. First step\n"
                        "2. Second step\n"
                        "...\n\n"
                        "After stating your plan, begin executing step 1.\n"
                        "Do NOT call tools without a plan."
                    )
                )

            # Budget warning in explicit mode — nudge model to complete
            if self.config.completion_mode == "explicit":
                remaining = self.config.max_iterations - state.iteration
                if remaining == 2:
                    state = state.with_message(
                        Message.system(
                            f"[Budget Warning] You have {remaining} iterations left. "
                            "Start wrapping up. Call task_complete(summary='your findings') "
                            "to finish, or you'll hit the iteration limit."
                        )
                    )
                elif remaining == 0:
                    state = state.with_message(
                        Message.system(
                            "[Final Iteration] This is your LAST iteration. "
                            "You MUST call task_complete now with a summary of everything "
                            "you've found. Do NOT call any other tools."
                        )
                    )

            if not _budget_nudged:
                state, nudge = self._budget_nudge(state, started_at)
                if nudge is not None:
                    _budget_nudged = True
                    yield nudge

            # Get model response. When the caller asked for tokens, the
            # model call runs as a task and its chunks are drained here, so
            # they surface while the model is still producing rather than
            # after it finishes.
            if state.would_exceed_cost_budget(self.config.max_tokens or 4096):
                yield TerminateEvent(
                    reason="cost_budget",
                    iterations_used=state.iteration,
                    final_confidence=state.confidence,
                    usage=_usage_of(state),
                    cost_usd=_cost_of(state),
                    reported_cost_usd=_reported_cost_of(state),
                    total_tool_calls=len(state.tool_executions),
                    final_message=_last_assistant_content,
                )
                break

            # Keep the request inside the window before it is sent. The
            # compacted history replaces the state's, so the run continues
            # from the summary on its own, with no prompt to resume.
            state, compaction = await self._compact_context(state, rc)
            if compaction is not None:
                yield compaction
                if compaction.exhausted:
                    yield TerminateEvent(
                        reason="context_exhausted",
                        iterations_used=state.iteration,
                        final_confidence=state.confidence,
                        usage=_usage_of(state),
                        cost_usd=_cost_of(state),
                        reported_cost_usd=_reported_cost_of(state),
                        total_tool_calls=len(state.tool_executions),
                        final_message=f"[context exhausted] {compaction.detail}",
                    )
                    break

            if stream_tokens:
                chunk_queue: asyncio.Queue[Any] = asyncio.Queue()
                model_task = asyncio.create_task(
                    self._get_model_response(state, model_kwargs, chunk_queue, run=rc)
                )
                while not model_task.done() or not chunk_queue.empty():
                    try:
                        chunk = await asyncio.wait_for(chunk_queue.get(), timeout=0.05)
                    except TimeoutError:
                        continue
                    if chunk is _DISCARDED_CALL:
                        _held_chunks = []
                        continue
                    if isinstance(chunk, ModelRetryEvent):
                        # A notice, not model text: the hold never applies.
                        yield chunk
                        continue
                    if _hold_tokens and not _is_reasoning_only(chunk):
                        _held_chunks.append(chunk)
                        continue
                    yield chunk
                response, state = await model_task
            else:
                try:
                    response, state = await self._get_model_response(
                        state, model_kwargs, run=rc
                    )
                except Exception:
                    for retry in rc.drain_retries():
                        yield retry
                    raise
                for retry in rc.drain_retries():
                    yield retry
            for custom in rc.drain():
                yield custom
            prompt_toks = response.usage.get("prompt_tokens", 0)
            completion_toks = response.usage.get("completion_tokens", 0)
            _total_tokens += prompt_toks + completion_toks
            state = state.with_response_usage(
                response.usage, getattr(response, "cost_usd", None)
            )
            _last_assistant_content = response.message.content
            # Track for the user-supplied termination condition. Updated again
            # below if a text tool call is parsed out of the body.
            _last_no_tool_calls = not response.message.tool_calls

            # Store plan from first iteration if planning enabled
            if self.config.planning and state.iteration == 1 and response.message.content:
                state = state.with_metadata("plan", response.message.content)

            # Emit think event. Reasoning models surface their CoT in
            # ``response.reasoning`` (separate channel) — prefer it over
            # the answer text in ``message.content``.
            yield ThinkEvent(
                iteration=state.iteration,
                reasoning=self._reasoning_for(response),
                tool_calls=list(response.message.tool_calls),
            )

            # No structured tool calls: the model may have written its call
            # in the body — its own markup leaked as text, or, for a model
            # without native tool calling, a plain text call.
            recovered = self._recover_text_tool_calls(response, state)
            if recovered is not None:
                response, state = recovered
                _last_assistant_content = response.message.content
                _last_no_tool_calls = False

            # A call block cut off before its end — the output-token limit
            # in the middle of a large write — is neither a call nor an
            # answer: the model is asked for the call again.
            if response.message.tool_calls:
                _unfinished_sent_back = 0
            elif _unfinished_sent_back < _MAX_UNFINISHED_SENDS:
                unfinished = self._unfinished_text_tool_call(response.message.content)
                if unfinished is not None:
                    _unfinished_sent_back += 1
                    state = self._send_back_unfinished_call(state, unfinished, replace=True)
                    _last_assistant_content = unfinished.prose
                    _last_no_tool_calls = False
                    _held_chunks = []  # the broken call is never shown
                    continue

            # Empty-content safety net (fixes #280).
            #
            # The model can return zero tool_calls AND zero
            # ``content`` for several reasons that all look the
            # same to the runtime:
            #
            #   - Reasoning-only iteration (gpt-5.x / o-series /
            #     Gemini 2.5 thinking mode generate
            #     ``completion_tokens`` that land in a separate
            #     reasoning channel — ``message.content`` is
            #     None even though tokens were consumed).
            #   - The model decided "I'm done" without writing
            #     anything after a long context (observed on
            #     Gemini with > 50K-token system prompts).
            #   - The model returned an empty assistant message
            #     between tool calls and the runtime collapses
            #     to this branch on the trailing iteration.
            #
            # Without the safety net, ``TerminateEvent.final_message``
            # becomes None → ``AgentResult.message`` becomes ""
            # → callers see empty output despite real work done.
            # Mirror the MaxIterations summary-injection path:
            # force one no-tools completion to extract the
            # model's actual answer based on the conversation
            # so far. Costs ~one extra call only when the bug
            # shape would otherwise produce empty output.
            #
            # Not on the first empty reply of a turn that is calling
            # tools, though: there an empty reply is far more often a
            # call the provider failed to deliver (a reasoning-only
            # turn, a call left in the reasoning channel) than an
            # answer. Asking for a final answer with the tools taken
            # away then ends a run mid-task — "the system requested my
            # final answer before I could make the edits" — and
            # reports it as complete. The model is sent back once,
            # with its tools, to carry on or to answer.
            #
            # A model that answers the no-tools request with a call is not
            # done either, so the call is made (``_summary_as_tool_step``).
            summary_answer: str | None = None
            if (
                not response.message.tool_calls
                and self.config.completion_mode != "explicit"
                and not response.message.content
            ):
                if state.tool_executions and not _empty_sent_back:
                    _empty_sent_back = True
                    state = state.with_message(Message.system(_EMPTY_REPLY_NOTE))
                    _last_no_tool_calls = False
                    continue
                before_note = state
                state = state.with_message(Message.system(_FINAL_ANSWER_REQUEST))
                messages = list(state.messages)
                if self._conversation_manager:
                    if hasattr(self._conversation_manager, "async_apply"):
                        messages = await self._conversation_manager.async_apply(messages)
                    else:
                        messages = self._conversation_manager.apply(messages)
                messages = self._validate_messages(messages)

                summary_model = self._auxiliary_model or self._model
                summary_resp: ModelResponse | None = None
                try:
                    summary_resp = await summary_model.complete(
                        messages=messages,
                        tools=None,  # No tools — force text
                        temperature=self.config.temperature,
                        max_tokens=self.config.max_tokens,
                    )
                    s_prompt_toks = summary_resp.usage.get("prompt_tokens", 0)
                    s_completion_toks = summary_resp.usage.get("completion_tokens", 0)
                    _total_tokens += s_prompt_toks + s_completion_toks
                    state = state.with_response_usage(
                        summary_resp.usage, getattr(summary_resp, "cost_usd", None)
                    )
                except Exception:  # noqa: BLE001
                    # Summary call failed — fall back to the last
                    # assistant content we saw or a deterministic stub.
                    # Better than empty.
                    summary_resp = None
                # The state without the note asking for an answer, which
                # the model is not shown again if it is sent back on.
                without_note = state.model_copy(update={"messages": before_note.messages})
                as_call = (
                    self._summary_as_tool_step(summary_resp, without_note)
                    if summary_resp is not None
                    else None
                )
                if isinstance(as_call, tuple):
                    # The model's own markup, written because the request
                    # had no tools to call: DeepSeek answers "give your
                    # final answer" with its next read.
                    response, state = as_call
                    _last_assistant_content = response.message.content
                    _last_no_tool_calls = False
                    _unfinished_sent_back = 0
                elif as_call is not None and _unfinished_sent_back < _MAX_UNFINISHED_SENDS:
                    _unfinished_sent_back += 1
                    state = self._send_back_unfinished_call(
                        without_note, as_call, replace=False
                    )
                    _last_no_tool_calls = False
                    continue
                else:
                    summary_answer = (
                        (summary_resp.message.content if summary_resp is not None else None)
                        or _last_assistant_content
                        or self._build_fallback_summary(state)
                    )

            # A held call that turned out to be a tool step is not a final
            # answer: release its chunks now, in order.
            is_final = (
                not response.message.tool_calls and self.config.completion_mode != "explicit"
            )
            if _held_chunks and not is_final:
                for held in _held_chunks:
                    out = held
                    if self.config.hold_tool_step_text and held.content:
                        # Words the verifier never saw stay off the
                        # stream; tool-call and usage parts still go out.
                        if not held.tool_calls and not getattr(held, "usage", None):
                            continue
                        out = held.model_copy(update={"content": None})
                    yield out
                _held_chunks = []

            # If still no tool calls — in auto mode we're done, in explicit mode we continue
            if is_final:
                # Apply grounding before final response if enabled
                if (
                    self.config.grounding
                    and self.config.grounding.enabled
                    and self.config.grounding.check_before_final
                    and self._grounding_evaluator
                    and response.message.content
                    and len(state.tool_executions) > 0
                ):
                    grounding_event, state = await self._apply_grounding(
                        state, response.message.content
                    )
                    _grounding_evals += 1
                    yield grounding_event

                    # If grounding fails, inject guidance and continue loop
                    if grounding_event.requires_replan and _grounding_evals <= (
                        self.config.grounding.max_replans
                    ):
                        from tulip.reasoning.grounding import GroundingResult

                        replan_guidance = self._grounding_evaluator.get_replan_guidance(
                            GroundingResult(
                                score=grounding_event.score,
                                ungrounded_claims=grounding_event.ungrounded_claims,
                                requires_replan=True,
                            )
                        )
                        state = state.with_message(
                            Message.system(f"[Grounding Check Failed]\n{replan_guidance}")
                        )
                        continue  # Re-enter loop for replanning

                # An empty reply was answered by the no-tools request
                # above; its answer is the run's.
                final_content = (
                    summary_answer if summary_answer is not None else response.message.content
                )

                # Pluggable final-answer verification: runs on every final
                # answer, tool call or not.
                if self.config.final_answer_verifier is not None and final_content:
                    verdict = await self._verify_final_answer(
                        final_content, state, rc, _verifier_attempts
                    )
                    if verdict.replanning:
                        yield verdict
                        _verifier_attempts += 1
                        _held_chunks = []  # the rejected draft is never shown
                        state = self._queue_verifier_replan(
                            state, final_content, verdict.feedback or "", verdict
                        )
                        # The model is sent back: a termination condition
                        # on "no tool calls" must not end the run on the
                        # reply it is being sent back from.
                        _last_no_tool_calls = False
                        continue
                    replacement = await self._final_answer_fallback(
                        verdict, final_content, state, rc
                    )
                    if replacement is not None:
                        verdict = verdict.model_copy(update={"replaced": True})
                    yield verdict
                    if replacement is not None:
                        state = self._replace_final_answer(state, final_content, replacement)
                        final_content = replacement
                        _held_chunks = []  # the failed draft is never shown
                        if stream_tokens and _hold_tokens:
                            yield ModelChunkEvent(content=replacement)
                for held in _held_chunks:
                    yield held
                _held_chunks = []

                yield TerminateEvent(
                    reason="complete",
                    iterations_used=state.iteration,
                    final_confidence=state.confidence,
                    usage=_usage_of(state),
                    cost_usd=_cost_of(state),
                    reported_cost_usd=_reported_cost_of(state),
                    total_tool_calls=len(state.tool_executions),
                    final_message=final_content,
                )
                break

            # Execute tool calls.
            #
            # Three phases so ``tool_execution="concurrent"`` is real
            # (#210): per-call serial hook/cache work is hoisted out
            # of the executor call so the survivors can run in a
            # single ``asyncio.gather``. Without this split, the
            # earlier per-call loop fed the executor singletons and
            # ``ConcurrentExecutor`` collapsed to ``SequentialExecutor``.
            #
            #   Phase 1 — per call, serial: emit ToolStartEvent, run
            #     before-hooks, resolve cancel/idempotent-cache short-
            #     circuits (recorded on state immediately so a later
            #     same-args call in this batch still cache-hits).
            #   Phase 2 — one batched ``executor.execute(...)`` over
            #     the surviving calls.
            #   Phase 3 — per result, in tool_call order: interrupt
            #     detection, result truncation/offload, state update,
            #     ToolCompleteEvent, after-hook (with retry), write/
            #     verification tracking.
            tool_results: list[ToolResult] = []
            reasoning_step_tools: list[ToolExecution] = []

            # Phase 1 — pre-execute.
            slots: list[dict[str, Any]] = []
            to_execute_indices: list[int] = []
            to_execute_calls: list[ToolCall] = []
            # Within-batch idempotent dedup: when an ``@tool(idempotent=True)``
            # appears twice in the same batch with the same arguments, the
            # second slot becomes a ``batch_cache_ref`` pointing at the first
            # slot. Phase 3 copies the first slot's result back. Without
            # this, the README contract ("idempotent body fires once per
            # run") is silently violated whenever a model fans out
            # duplicates in one response — see #210 follow-up.
            batch_idempotent_seen: dict[tuple[str, str], int] = {}

            for tool_call in response.message.tool_calls:
                _tool_calls_count += 1

                yield ToolStartEvent(
                    tool_name=tool_call.name,
                    tool_call_id=tool_call.id,
                    arguments=tool_call.arguments,
                )

                tool_event = await self._run_before_tool_hooks(
                    tool_call.name, tool_call.id, tool_call.arguments, run=rc
                )
                for custom in rc.drain():
                    yield custom

                if tool_event.cancel:
                    cancel_msg = (
                        tool_event.cancel
                        if isinstance(tool_event.cancel, str)
                        else "Cancelled by hook"
                    )
                    cancel_result = ToolResult(
                        tool_call_id=tool_call.id,
                        name=tool_call.name,
                        content=cancel_msg,
                        error=None,
                        duration_ms=0.0,
                    )
                    cancel_execution = ToolExecution(
                        tool_name=cancel_result.name,
                        tool_call_id=cancel_result.tool_call_id,
                        arguments=tool_call.arguments,
                        result=cancel_result.content,
                    )
                    # NOTE: state is updated in Phase 3 (tool_call order),
                    # not here, so concurrent batches keep deterministic
                    # ordering on ``state.tool_executions``.
                    slots.append(
                        {
                            "tool_call": tool_call,
                            "arguments": tool_call.arguments,
                            "kind": "cancel",
                            "result": cancel_result,
                            "execution": cancel_execution,
                        }
                    )
                    continue

                modified_args = tool_event.arguments
                # What the tool is called with. ``modified_args`` is what
                # is recorded (state, checkpoint, after-hooks); hook
                # ``secret_arguments`` ride only on the invocation.
                invoke_args = _invocation_arguments(tool_event)

                # Idempotent dedup: if the tool declared idempotent=True
                # and a prior call in this run used the same arguments,
                # reuse the prior result instead of invoking the body.
                # Without this, ``@tool(idempotent=True)`` is silently a
                # no-op for the main Agent.run() path (despite being
                # advertised on the README hero example).
                cached = self._maybe_cached_idempotent_result(
                    state, tool_call.name, modified_args, tool_call.id
                )
                if cached is not None:
                    cache_execution = ToolExecution(
                        tool_name=cached.name,
                        tool_call_id=cached.tool_call_id,
                        arguments=modified_args,
                        result=cached.content if cached.success else None,
                        error=cached.error,
                        duration_ms=cached.duration_ms,
                        idempotent_cache_hit=True,
                    )
                    slots.append(
                        {
                            "tool_call": tool_call,
                            "arguments": modified_args,
                            "kind": "cache",
                            "result": cached,
                            "execution": cache_execution,
                        }
                    )
                    continue

                # Within-batch dedup for idempotent tools whose duplicate
                # appears alongside the original in this same response.
                # Without this, a model emitting ``[submit(X), submit(X)]``
                # in one assistant message would fire the body twice — a
                # regression vs the pre-batching per-call loop, which
                # sneaked the dedup in via serial state updates.
                idempotent_key = self._idempotent_batch_key(tool_call.name, modified_args)
                if idempotent_key is not None:
                    prior_slot = batch_idempotent_seen.get(idempotent_key)
                    if prior_slot is not None:
                        slots.append(
                            {
                                "tool_call": tool_call,
                                "arguments": modified_args,
                                "kind": "batch_cache_ref",
                                "ref_slot": prior_slot,
                                "result": None,
                                "execution": None,
                            }
                        )
                        continue
                    batch_idempotent_seen[idempotent_key] = len(slots)

                slots.append(
                    {
                        "tool_call": tool_call,
                        "arguments": modified_args,
                        "invoke_arguments": invoke_args,
                        "kind": "execute",
                        "result": None,
                        "execution": None,
                    }
                )
                to_execute_indices.append(len(slots) - 1)
                to_execute_calls.append(tool_call.model_copy(update={"arguments": invoke_args}))

            # Phase 2 — stream the survivors through the executor.
            # Constructed after Phase 1 so ``state`` already reflects
            # cancel / cache short-circuits, matching the pre-extraction
            # per-call-loop semantics.
            #
            # Using ``execute_streaming`` (yields ``(input_idx, result)``
            # in completion order for ConcurrentExecutor, input order for
            # SequentialExecutor) gives us two things in one pass:
            #   * Opt-in completion-order ``ToolCompleteEvent`` for UI
            #     streaming (``AgentConfig.tool_event_order='completion'``).
            #   * Active sibling cancellation on interrupt: a ``break``
            #     out of the ``async for`` triggers the executor's
            #     ``finally`` clause, which cancels in-flight sibling
            #     tasks. The pre-streaming ``gather`` impl had no such
            #     hook — every sibling completed and only the fold was
            #     halted.
            if to_execute_calls:
                ctx_factory = ToolContextFactory(
                    run_id=state.run_id,
                    agent_id=state.agent_id,
                    iteration=state.iteration,
                    state=state,
                    invocation_metadata=metadata or {},
                    ephemeral_metadata=rc.ephemeral,
                )
                batch_start = time.perf_counter()
                interrupted_slot_idx: int | None = None
                results_stream = self._executor.execute_streaming(
                    to_execute_calls,
                    self._tool_registry,
                    ctx_factory,
                )
                # Tools that report progress (every MCP tool) need the
                # stream merged with their progress, or it would arrive
                # only once they had finished.
                merged: AsyncIterator[tuple[int, ToolResult] | TulipEvent] = (
                    _interleave_progress(results_stream, asyncio.Queue())
                    if self._emits_progress(to_execute_calls)
                    else results_stream
                )
                try:
                    async for item in merged:
                        # Progress, or a subagent's event forwarded live.
                        if isinstance(item, TulipEvent):
                            yield item
                            continue
                        input_idx, batched_result = item
                        slot_idx = to_execute_indices[input_idx]
                        slots[slot_idx]["result"] = batched_result

                        if self.config.tool_event_order == "completion":
                            # Stream the complete event the moment the
                            # tool finishes. Phase 3 will skip its own
                            # ToolCompleteEvent emission for execute-kind
                            # slots when in completion mode.
                            yield _complete_event(batched_result)

                        # Interrupt detection — break to trigger the
                        # executor's cancellation of in-flight siblings.
                        # Phase 3 will run the InterruptEvent emission
                        # path on the interrupting slot exactly like the
                        # pre-streaming code did.
                        if (
                            batched_result.content
                            and '"__interrupt__": true' in batched_result.content
                        ):
                            interrupted_slot_idx = slot_idx
                            break
                except Exception as e:  # noqa: BLE001 — ``_execute_one`` already traps user-tool exceptions; this catches a failure of the executor itself (registry blow-up, ctx_factory error) so a single bad call doesn't abort the whole turn.
                    batch_duration = (time.perf_counter() - batch_start) * 1000
                    for slot_idx, tc in zip(to_execute_indices, to_execute_calls, strict=True):
                        if slots[slot_idx]["result"] is None:
                            slots[slot_idx]["result"] = ToolResult(
                                tool_call_id=tc.id,
                                name=tc.name,
                                content="",
                                error=str(e),
                                duration_ms=batch_duration,
                            )
                finally:
                    if merged is not results_stream:
                        # Cancels a pull still in flight after an
                        # interrupt ``break`` — the in-flight siblings
                        # with it — and closes the executor stream.
                        with contextlib.suppress(Exception):
                            await merged.aclose()  # type: ignore[attr-defined]

                # Synthesize "cancelled by sibling interrupt" results
                # for slots whose tasks were cancelled (the interrupt
                # break happened before they completed). Folded as
                # ``kind="cancel"`` so Phase 3 treats them identically
                # to hook-cancelled calls — Start/Complete pair emits,
                # state records the cancellation, no after-hook fires.
                if interrupted_slot_idx is not None:
                    for slot_idx in to_execute_indices:
                        slot = slots[slot_idx]
                        if slot["result"] is not None:
                            continue
                        tc = slot["tool_call"]
                        cancelled_result = ToolResult(
                            tool_call_id=tc.id,
                            name=tc.name,
                            content="Cancelled: sibling tool requested interrupt",
                            error=None,
                            duration_ms=0.0,
                        )
                        cancelled_execution = ToolExecution(
                            tool_name=tc.name,
                            tool_call_id=tc.id,
                            arguments=slot["arguments"],
                            result=cancelled_result.content,
                        )
                        slot["result"] = cancelled_result
                        slot["execution"] = cancelled_execution
                        slot["kind"] = "cancel"

            # Phase 3 — per-result fold, in tool_call order.
            for slot in slots:
                tool_call = slot["tool_call"]
                modified_args = slot["arguments"]
                kind = slot["kind"]
                result = slot["result"]

                if kind == "cancel":
                    tool_results.append(result)
                    state = state.with_tool_execution(slot["execution"])
                    reasoning_step_tools.append(slot["execution"])
                    yield _complete_event(result, result=result.content, duration_ms=0.0)
                    continue
                if kind == "cache":
                    tool_results.append(result)
                    state = state.with_tool_execution(slot["execution"])
                    reasoning_step_tools.append(slot["execution"])
                    yield _complete_event(result, result=result.content)
                    continue
                if kind == "batch_cache_ref":
                    # Same-args duplicate of an earlier slot in this same
                    # batch. The referenced slot's result is now resolved
                    # (Phase 2 ran ``asyncio.gather`` to completion before
                    # this loop). Reuse it and rebrand the result/execution
                    # for this call's id.
                    ref_slot = slots[slot["ref_slot"]]
                    ref_result: ToolResult = ref_slot["result"]
                    result = ref_result.model_copy(
                        update={
                            "tool_call_id": tool_call.id,
                            "name": tool_call.name,
                            "duration_ms": 0.0,
                        }
                    )
                    batch_cache_execution = ToolExecution(
                        tool_name=result.name,
                        tool_call_id=result.tool_call_id,
                        arguments=modified_args,
                        result=result.content if result.success else None,
                        error=result.error,
                        duration_ms=0.0,
                        idempotent_cache_hit=True,
                    )
                    tool_results.append(result)
                    state = state.with_tool_execution(batch_cache_execution)
                    reasoning_step_tools.append(batch_cache_execution)
                    yield _complete_event(result, result=result.content, duration_ms=0.0)
                    continue

                # Interrupt marker from ``ask_user``. Sibling calls in
                # this batch have already executed in parallel — we
                # still pause here, leaving them un-folded into state.
                # An interrupt is rare and ``ask_user`` is the only
                # tool that emits it; we trust users not to fan it
                # out alongside high-cost siblings.
                if result.content and '"__interrupt__": true' in result.content:
                    import json as _json

                    try:
                        interrupt_data = _json.loads(result.content)
                        if interrupt_data.get("__interrupt__"):
                            self._park_interrupt(rc, state)
                            # Checkpoint BEFORE yielding: a consumer that
                            # stops iterating on the interrupt (an HTTP
                            # layer parking the run) never resumes this
                            # generator, so the finally-side save would
                            # only run at GC — too late if the process
                            # dies while paused.
                            if self.config.checkpointer and thread_id:
                                await self.config.checkpointer.save(_durable(state), thread_id)
                            yield InterruptEvent(
                                question=interrupt_data.get("question", ""),
                                options=interrupt_data.get("options"),
                                fields=interrupt_data.get("fields"),
                                interrupt_id=result.tool_call_id,
                                metadata=interrupt_data.get("metadata") or {},
                            )
                            return  # Pause the generator
                    except (ValueError, KeyError):
                        pass

                # Cap oversized tool results so they don't blow the
                # model's context window. When ``tool_result_store`` is
                # configured the full payload is offloaded through it and
                # a recoverable reference key inlined; otherwise the
                # middle is cut and both ends kept.
                # With ObservationPack on, nothing is lost to the cap: an
                # output up to the pack's inline limit goes whole (the pack
                # sends it as a placeholder once it is old), and a larger
                # one is archived whole and cut around a pointer to it.
                if (
                    self.config.max_tool_result_length > 0
                    and result.content
                    and text_length(result.content) > self.config.max_tool_result_length
                ):
                    packed = (
                        self._intake_large_output(result, state, rc)
                        if self.config.tool_result_store is None
                        else None
                    )
                    if packed is not None:
                        result = packed
                    elif self.config.tool_result_store is not None:
                        result = self.config.tool_result_store.maybe_offload(
                            result,
                            run_id=state.run_id,
                            iteration=state.iteration,
                        )
                    else:
                        # Cutting through an embedded image would leave
                        # corrupt base64, so images go before the cut.
                        result = result.model_copy(
                            update={
                                "content": truncate_tool_output(
                                    strip_images(result.content),
                                    self.config.max_tool_result_length,
                                    self.config.tool_result_head_fraction,
                                )
                            }
                        )

                # After-hooks run BEFORE the result is folded into state
                # or emitted, so ``event.result`` replacement actually
                # lands where the docs promise it does: in the message
                # the model reads, in ``state.tool_executions``, and in
                # the ToolCompleteEvent a consumer sees. (Previously the
                # hook ran last: only ``retry`` had any effect, a
                # replaced result was silently discarded, and a retried
                # result never reached the model because ``tool_results``
                # already held the original.) In completion mode Phase 2
                # has already streamed the event, so there the event
                # carries the pre-hook result; state and messages still
                # get the hook's version.
                after_tool_event = await self._run_after_tool_hooks(
                    result.name,
                    result.content if result.success else None,
                    result.error,
                    tool_call_id=result.tool_call_id,
                    arguments=modified_args,
                    run=rc,
                )
                hook_events = rc.drain()

                if after_tool_event.retry:
                    try:
                        retry_ctx_factory = ToolContextFactory(
                            run_id=state.run_id,
                            agent_id=state.agent_id,
                            iteration=state.iteration,
                            state=state,
                            invocation_metadata=metadata or {},
                            ephemeral_metadata=rc.ephemeral,
                        )
                        [result] = await self._executor.execute(
                            [
                                tool_call.model_copy(
                                    update={
                                        "arguments": slot.get("invoke_arguments", modified_args)
                                    }
                                )
                            ],
                            self._tool_registry,
                            retry_ctx_factory,
                        )
                    except Exception as e:  # noqa: BLE001 — user tool bodies can raise anything; surface as ToolResult.error
                        result = ToolResult(
                            tool_call_id=tool_call.id,
                            name=tool_call.name,
                            content="",
                            error=str(e),
                            duration_ms=0.0,
                        )
                else:
                    result = _apply_hook_result(result, after_tool_event)

                tool_results.append(result)

                execution = ToolExecution(
                    tool_name=result.name,
                    tool_call_id=result.tool_call_id,
                    arguments=modified_args,
                    result=result.content if result.success else None,
                    error=result.error,
                    duration_ms=result.duration_ms,
                )
                state = state.with_tool_execution(execution)
                reasoning_step_tools.append(execution)

                if result.error:
                    _tool_errors_count += 1

                # Skip the per-slot ToolCompleteEvent yield in
                # completion mode — Phase 2 already streamed it the
                # moment this tool finished. Sequential mode (default)
                # still emits here so consumers see events in
                # tool_call order, carrying the post-hook result.
                if self.config.tool_event_order == "sequential":
                    yield _complete_event(result)
                # UI-only events the after-hook emitted, right behind the
                # call they describe. Never folded into state/messages.
                for custom in hook_events:
                    yield custom

                if result.name in self.config.verify_tools:
                    rc.has_unverified_writes = True
                if result.name in self.config.verification_tools:
                    rc.has_unverified_writes = False

            # Add tool results to messages
            for result in tool_results:
                state = state.with_message(Message.tool(result))

            # Inject verification reminder if write-like tools were used
            if self.config.verify_tools:
                tools_used = {e.tool_name for e in reasoning_step_tools}
                wrote = tools_used & self.config.verify_tools
                if wrote:
                    state = state.with_message(
                        Message.system(
                            "[Verification Reminder] You modified files/data. "
                            "Before completing, verify your changes:\n"
                            "- Run tests or checks if available\n"
                            "- Read back modified files to confirm correctness\n"
                            "- Fix any issues found\n"
                            "Do NOT call task_complete until verified."
                        )
                    )

            # Apply Reflexion if enabled
            if (
                self.config.reflexion
                and self.config.reflexion.enabled
                and self._reflector
                and state.iteration % self.config.reflexion.evaluate_every_n_iterations == 0
            ):
                reflect_event, state = await self._apply_reflexion(state, reasoning_step_tools)
                _reflexion_evals += 1
                yield reflect_event

                # Inject guidance when agent is stuck or looping
                if self.config.reflexion.include_guidance and reflect_event.guidance:
                    guidance = f"[Agent Self-Reflection]\n{reflect_event.guidance}"
                    # Add replan suggestion if planning is enabled and agent is stuck
                    if self.config.planning and reflect_event.assessment in (
                        "stuck",
                        "loop_detected",
                    ):
                        guidance += (
                            "\n\n[Replan] Your current approach isn't working. "
                            "Create a NEW plan with a different strategy, then execute it."
                        )
                    state = state.with_message(Message.system(guidance))

            # Record reasoning step
            reasoning_step = ReasoningStep(
                iteration=state.iteration,
                thought=response.message.content,
                tool_calls=list(response.message.tool_calls),
                tool_results=reasoning_step_tools,
                reflection=None,  # Will be updated if reflexion was applied
                confidence_delta=0.0,
            )
            state = state.with_reasoning_step(reasoning_step)

            # The iteration's tool results are folded in, so this is a
            # boundary ``continue_turn`` can pick the turn up from.
            state = await self._checkpoint_iteration(state, thread_id)

        # The loop is done; close whichever iteration it left open.
        if _open_iteration is not None:
            await self._run_iteration_end_hooks(_open_iteration, state)

    except Exception as e:
        # Emit error termination
        state = state.with_error(str(e))
        yield TerminateEvent(
            reason="error",
            iterations_used=state.iteration,
            final_confidence=state.confidence,
            usage=_usage_of(state),
            cost_usd=_cost_of(state),
            reported_cost_usd=_reported_cost_of(state),
            total_tool_calls=len(state.tool_executions),
            error=_error_text(e),
        )
        raise

    finally:
        # Fold what the last tool batch's children reported (the loop may
        # have exited before its next top-of-iteration fold), THEN stop
        # being anyone's parent, so the state handed to ``_end_run``
        # below counts delegated spend on every exit path.
        state = fold_subagent_usage(state)
        exit_parent_run(_subagent_ctx)

        # Consume a pending cancel-all so it does not leak into the next run
        if self._cancel_signal is not None:
            self._cancel_signal.clear()

        # Save output to state if output_key configured
        if self.config.output_key:
            final_msg = ""
            for msg in reversed(state.messages):
                if msg.role.value == "assistant" and msg.content:
                    final_msg = msg.content
                    break
            if final_msg:
                state = state.with_metadata(self.config.output_key, final_msg)

        # Hand the final state to THIS run's arun (never read back off
        # the shared agent), then release the run's bookkeeping. The
        # list of iteration saves is bookkeeping too, not a result.
        superseded, state = _take_iteration_checkpoints(state)
        self._end_run(rc, state)

        # Run hooks: after_invocation
        _duration_ms = (datetime.now(UTC) - started_at).total_seconds() * 1000  # noqa: F841
        await self._run_after_invocation_hooks(state, len(state.errors) == 0)

        # Extract and persist long-term memories from this session.
        if self._memory_manager is not None:
            # Without turn-only messages: a rejected draft and the
            # verifier's note are not facts about the user.
            await self._memory_manager.on_session_end(_without_ephemeral_messages(state))

        # Final checkpoint
        await self._checkpoint_final(state, thread_id, superseded)

AgentConfig

AgentConfig

Bases: BaseModel

Configuration for an Agent instance.

All parameters can be validated before agent creation.

tool_step_text_needs_the_answer_hold

tool_step_text_needs_the_answer_hold() -> AgentConfig

hold_tool_step_text drops what hold_final_answer_tokens holds; without that hold nothing is held, and the flag would silently do nothing.

Source code in .sdk/src/tulip/agent/config.py

@model_validator(mode="after")
def tool_step_text_needs_the_answer_hold(self) -> AgentConfig:
    """``hold_tool_step_text`` drops what ``hold_final_answer_tokens`` holds;
    without that hold nothing is held, and the flag would silently do nothing."""
    if self.hold_tool_step_text and not self.hold_final_answer_tokens:
        raise ValueError(
            "hold_tool_step_text needs hold_final_answer_tokens=True (and a "
            "final_answer_verifier): only held text can be dropped."
        )
    return self

point_unknown_fields_at_model_kwargs classmethod

point_unknown_fields_at_model_kwargs(data: Any) -> Any

Say where provider configuration goes, instead of only that it cannot go here.

extra="forbid" catches typos, which is worth keeping, but on its own it answers the wrong question. A reader who writes the documented one-string form and adds the base_url that provider requires gets Extra inputs are not permitted and no hint that the setting is supported at all, one field over. That is how a supported path reads as a missing feature.

Source code in .sdk/src/tulip/agent/config.py

@model_validator(mode="before")
@classmethod
def point_unknown_fields_at_model_kwargs(cls, data: Any) -> Any:
    """Say where provider configuration goes, instead of only that it cannot go here.

    ``extra="forbid"`` catches typos, which is worth keeping, but on its
    own it answers the wrong question. A reader who writes the documented
    one-string form and adds the ``base_url`` that provider *requires*
    gets ``Extra inputs are not permitted`` and no hint that the setting
    is supported at all, one field over. That is how a supported path
    reads as a missing feature.
    """
    if not isinstance(data, dict):
        return data
    unknown = sorted(set(data) - set(cls.model_fields))
    if unknown:
        raise ValueError(
            f"{unknown} {'is' if len(unknown) == 1 else 'are'} not "
            f"{'a ' if len(unknown) == 1 else ''}field"
            f"{'' if len(unknown) == 1 else 's'} of AgentConfig. If this is "
            "provider configuration for a string model, pass it as "
            "model_kwargs={...} — e.g. "
            'Agent(model="openai-compatible:my-model", '
            'model_kwargs={"base_url": "https://host/v1"}).'
        )
    return data

validate_model classmethod

validate_model(v: Any) -> Any

Validate model is a string or ModelProtocol.

Source code in .sdk/src/tulip/agent/config.py

@field_validator("model", mode="before")
@classmethod
def validate_model(cls, v: Any) -> Any:
    """Validate model is a string or ModelProtocol."""
    if isinstance(v, str):
        if ":" not in v:
            raise ValueError(
                f"Model string must be 'provider:model', got: {v}. Example: 'openai:gpt-4o'"
            )
        return v
    # Assume it's a ModelProtocol instance
    return v

validate_tools classmethod

validate_tools(v: Any) -> list[Any]

Ensure tools is a list.

Source code in .sdk/src/tulip/agent/config.py

@field_validator("tools", mode="before")
@classmethod
def validate_tools(cls, v: Any) -> list[Any]:
    """Ensure tools is a list."""
    if v is None:
        return []
    if not isinstance(v, list):
        return [v]
    return v

with_reflexion

with_reflexion(enabled: bool = True, confidence_threshold: float = 0.85, **kwargs: Any) -> AgentConfig

Return a copy with Reflexion configured.

Source code in .sdk/src/tulip/agent/config.py

def with_reflexion(
    self,
    enabled: bool = True,
    confidence_threshold: float = 0.85,
    **kwargs: Any,
) -> AgentConfig:
    """Return a copy with Reflexion configured."""
    return self.model_copy(
        update={
            "reflexion": ReflexionConfig(
                enabled=enabled,
                confidence_threshold=confidence_threshold,
                **kwargs,
            )
        }
    )

with_grounding

with_grounding(enabled: bool = True, threshold: float = 0.65, **kwargs: Any) -> AgentConfig

Return a copy with Grounding configured.

Source code in .sdk/src/tulip/agent/config.py

def with_grounding(
    self,
    enabled: bool = True,
    threshold: float = 0.65,
    **kwargs: Any,
) -> AgentConfig:
    """Return a copy with Grounding configured."""
    return self.model_copy(
        update={
            "grounding": GroundingConfig(
                enabled=enabled,
                threshold=threshold,
                **kwargs,
            )
        }
    )

with_hooks

with_hooks(*hooks: Any) -> AgentConfig

Return a copy with additional hooks.

Source code in .sdk/src/tulip/agent/config.py

def with_hooks(self, *hooks: Any) -> AgentConfig:
    """Return a copy with additional hooks."""
    return self.model_copy(update={"hooks": [*self.hooks, *hooks]})

Reasoning configs

AgentConfig.reflexion, AgentConfig.grounding, and AgentConfig.gsar accept typed configuration objects defined alongside the config itself. The boolean shorthand (reflexion=True) coerces into a default instance — the explicit form is what to reach for when you want to tune thresholds, swap models, or override defaults.

ReflexionConfig

Bases: BaseModel

Configuration for Reflexion reasoning pattern.

GroundingConfig

Bases: BaseModel

Configuration for Grounding evaluation.

GSARConfig

Bases: BaseModel

Configuration for the GSAR typed-grounding layer.

Wires the framework from arXiv:2604.23366 onto an Agent. When set on :class:AgentConfig, the agent runs the configured judge over its final assistant message + tool-execution history after the loop completes; the resulting :class:~tulip.reasoning.gsar_judge.JudgeOutput, scalar score S, and decision δ are surfaced on :class:~tulip.agent.result.AgentResult.

This is a single-pass v1 — the agent produces an answer, the judge scores it, and the result is exposed for the caller to act on. The full Algorithm-1 outer loop with regenerate / replan callbacks lives separately in :mod:tulip.reasoning.gsar_evaluator; wire it explicitly when you want the loop dynamics.

AgentResult

AgentResult

Bases: BaseModel

Result from an agent execution.

Contains the final message, state, and execution metrics.

success property

success: bool

Whether execution completed successfully.

confidence property

confidence: float

Final confidence score.

iterations property

iterations: int

Number of iterations used.

text property

text: str

Alias for message.

Many AI SDKs surface the final assistant text as .text; Tulip's primary field is .message. Both names now work.

messages property

messages: tuple[Message, ...]

All messages from the conversation.

tool_executions property

tool_executions: tuple[ToolExecution, ...]

All tool executions.

reasoning_steps property

reasoning_steps: tuple[ReasoningStep, ...]

All reasoning steps.

last_assistant_message property

last_assistant_message: str | None

Get the last assistant message content.

parsed_as

parsed_as(schema: type[T]) -> T

Return parsed cast to schema, with a runtime check.

Use this when you want a typed handle on the structured output without casting yourself::

picks = result.parsed_as(VendorList)
for v in picks.vendors:
    ...

Raises ValueError if parsed is None (parse failed or no schema configured) and TypeError if parsed is the wrong concrete type.

Source code in .sdk/src/tulip/agent/result.py

def parsed_as(self, schema: type[T]) -> T:
    """Return ``parsed`` cast to ``schema``, with a runtime check.

    Use this when you want a typed handle on the structured output without
    casting yourself::

        picks = result.parsed_as(VendorList)
        for v in picks.vendors:
            ...

    Raises ``ValueError`` if ``parsed`` is None (parse failed or no schema
    configured) and ``TypeError`` if ``parsed`` is the wrong concrete type.
    """
    if self.parsed is None:
        if self.parse_error:
            raise ValueError(f"AgentResult has no parsed output: {self.parse_error}")
        raise ValueError("AgentResult has no parsed output (no output_schema was configured)")
    if not isinstance(self.parsed, schema):
        raise TypeError(f"Expected {schema.__name__}, got {type(self.parsed).__name__}")
    return self.parsed

to_dict

to_dict() -> dict[str, Any]

Export result to dictionary.

Source code in .sdk/src/tulip/agent/result.py

def to_dict(self) -> dict[str, Any]:
    """Export result to dictionary."""
    return self.model_dump(mode="json")

from_state classmethod

from_state(state: AgentState, stop_reason: StopReason, metrics: ExecutionMetrics | None = None, started_at: datetime | None = None, error: str | None = None, grounding_score: float | None = None, ungrounded_claims: list[str] | None = None, parsed: BaseModel | None = None, parse_error: str | None = None, message: str | None = None, gsar_judgment: Any = None, gsar_score: float | None = None, gsar_decision: str | None = None) -> AgentResult

Create a result from final state.

Extracts the final message from the last assistant response unless an explicit message is supplied (used after a structuring re-prompt).

Source code in .sdk/src/tulip/agent/result.py

@classmethod
def from_state(
    cls,
    state: AgentState,
    stop_reason: StopReason,
    metrics: ExecutionMetrics | None = None,
    started_at: datetime | None = None,
    error: str | None = None,
    grounding_score: float | None = None,
    ungrounded_claims: list[str] | None = None,
    parsed: BaseModel | None = None,
    parse_error: str | None = None,
    message: str | None = None,
    gsar_judgment: Any = None,
    gsar_score: float | None = None,
    gsar_decision: str | None = None,
) -> AgentResult:
    """
    Create a result from final state.

    Extracts the final message from the last assistant response unless an
    explicit ``message`` is supplied (used after a structuring re-prompt).
    """
    # Find the last assistant message if not provided
    final_message = message
    if final_message is None:
        final_message = ""
        for msg in reversed(state.messages):
            if msg.role.value == "assistant":
                final_message = msg.content or ""
                break

    return cls(
        message=final_message,
        state=state,
        stop_reason=stop_reason,
        metrics=metrics or ExecutionMetrics(),
        started_at=started_at or state.started_at,
        completed_at=datetime.now(UTC),
        error=error,
        grounding_score=grounding_score,
        ungrounded_claims=ungrounded_claims or [],
        parsed=parsed,
        parse_error=parse_error,
        gsar_judgment=gsar_judgment,
        gsar_score=gsar_score,
        gsar_decision=gsar_decision,
    )

Result sub-types

AgentResult.metrics is an ExecutionMetrics, AgentResult.stop_reason is a StopReason literal, and the streaming entry point yields StreamingResult between events.

ExecutionMetrics

Bases: BaseModel

Metrics from agent execution.

tools_success_rate property

tools_success_rate: float

Percentage of successful tool calls.

tokens_per_iteration property

tokens_per_iteration: float

Average tokens per iteration.

StopReason module-attribute

StopReason = Literal['complete', 'terminal_tool', 'confidence_met', 'max_iterations', 'tool_loop', 'no_tools', 'grounding_failed', 'token_budget', 'cost_budget', 'time_budget', 'context_exhausted', 'interrupted', 'error', 'cancelled']

StreamingResult

Bases: BaseModel

Partial result during streaming.

Used to provide intermediate state during agent execution.

AgentState

AgentState

Bases: BaseModel

Immutable state for an agent execution.

All updates return a new state instance (functional updates).

priced property

priced: bool

Whether both prices are known, so spend can be measured.

tool_loop property

tool_loop: ToolLoop | None

The loop the run's last steps are in, or None (see :mod:tulip.core.loops).

A loop is the same step — calls by name and arguments, and their results — repeated back to back, or the same short cycle of steps repeated whole. Re-reading a file between other work, the same tool with other arguments, and a repeat whose result changed are progress.

has_tool_loop property

has_tool_loop: bool

Whether the run's last steps are a loop (warned about or not).

tool_loop_persists property

tool_loop_persists: bool

Whether the run's loop went on past the point where it is warned about.

This is what stops a run. The runtime warns the model when a loop reaches its threshold (:meth:with_tool_loop_warning); the warning is the model's chance to change approach, and one more repeat is the model ignoring it.

last_tool_calls property

last_tool_calls: list[ToolCall]

Get tool calls from the last assistant message of the current turn.

The scan stops at the most recent user message: calls made before it belong to an earlier turn, and treating them as "last" made a new turn on a thread whose previous turn ended with a terminal tool stop immediately with terminal_tool before the model saw the message.

called_terminal_tool property

called_terminal_tool: bool

Check if a terminal tool was called.

should_terminate property

should_terminate: tuple[bool, str | None]

Check if the agent should terminate.

In "auto" mode: stops on confidence, no_tools, tool_loop, or terminal_tool. In "explicit" mode: only stops on terminal_tool, max_iterations, or budgets. Use "explicit" for multi-step tasks that require verification before completion.

Returns:

Type	Description
tuple[bool, str | None]	Tuple of (should_stop, reason)

total_tokens property

total_tokens: int

Total tokens used. Returns real count if tracked, else char/4 estimate.

with_message

with_message(message: Message) -> AgentState

Add a message to the conversation.

Source code in .sdk/src/tulip/core/state.py

def with_message(self, message: Message) -> AgentState:
    """Add a message to the conversation."""
    return self.model_copy(
        update={
            "messages": (*self.messages, message),
            "updated_at": datetime.now(UTC),
        }
    )

with_messages

with_messages(messages: list[Message]) -> AgentState

Add multiple messages to the conversation.

Source code in .sdk/src/tulip/core/state.py

def with_messages(self, messages: list[Message]) -> AgentState:
    """Add multiple messages to the conversation."""
    return self.model_copy(
        update={
            "messages": (*self.messages, *messages),
            "updated_at": datetime.now(UTC),
        }
    )

with_iteration

with_iteration(iteration: int) -> AgentState

Update the current iteration.

Source code in .sdk/src/tulip/core/state.py

def with_iteration(self, iteration: int) -> AgentState:
    """Update the current iteration."""
    return self.model_copy(
        update={
            "iteration": iteration,
            "updated_at": datetime.now(UTC),
        }
    )

next_iteration

next_iteration() -> AgentState

Increment iteration counter.

Source code in .sdk/src/tulip/core/state.py

def next_iteration(self) -> AgentState:
    """Increment iteration counter."""
    return self.with_iteration(self.iteration + 1)

with_provider_state

with_provider_state(provider_state: dict[str, Any] | None) -> AgentState

Replace the provider continuation state.

Server-stateful transports (e.g. OCIResponsesModel) return a continuation token in ModelResponse.provider_state; the agent calls this to thread the token into the next turn.

Source code in .sdk/src/tulip/core/state.py

def with_provider_state(self, provider_state: dict[str, Any] | None) -> AgentState:
    """Replace the provider continuation state.

    Server-stateful transports (e.g. ``OCIResponsesModel``) return
    a continuation token in ``ModelResponse.provider_state``; the
    agent calls this to thread the token into the next turn.
    """
    return self.model_copy(
        update={
            "provider_state": provider_state,
            "updated_at": datetime.now(UTC),
        }
    )

with_tool_execution

with_tool_execution(execution: ToolExecution) -> AgentState

Record a tool execution.

Source code in .sdk/src/tulip/core/state.py

def with_tool_execution(self, execution: ToolExecution) -> AgentState:
    """Record a tool execution."""
    return self.model_copy(
        update={
            "tool_executions": (*self.tool_executions, execution),
            "tool_history": (*self.tool_history, execution.tool_name),
            "updated_at": datetime.now(UTC),
        }
    )

with_reasoning_step

with_reasoning_step(step: ReasoningStep) -> AgentState

Add a reasoning step to the trace.

Source code in .sdk/src/tulip/core/state.py

def with_reasoning_step(self, step: ReasoningStep) -> AgentState:
    """Add a reasoning step to the trace."""
    return self.model_copy(
        update={
            "reasoning_steps": (*self.reasoning_steps, step),
            "updated_at": datetime.now(UTC),
        }
    )

with_confidence

with_confidence(confidence: float) -> AgentState

Update confidence score.

Source code in .sdk/src/tulip/core/state.py

def with_confidence(self, confidence: float) -> AgentState:
    """Update confidence score."""
    clamped = max(0.0, min(1.0, confidence))
    return self.model_copy(
        update={
            "confidence": clamped,
            "confidence_history": (*self.confidence_history, clamped),
            "updated_at": datetime.now(UTC),
        }
    )

adjust_confidence

adjust_confidence(delta: float, diminishing: bool = True) -> AgentState

Adjust confidence with optional diminishing returns.

Parameters:

Name	Type	Description	Default
delta	float	Raw confidence adjustment (-1.0 to 1.0)	required
diminishing	bool	If True, positive deltas are scaled by (1 - current_confidence)	True

Source code in .sdk/src/tulip/core/state.py

def adjust_confidence(self, delta: float, diminishing: bool = True) -> AgentState:
    """
    Adjust confidence with optional diminishing returns.

    Args:
        delta: Raw confidence adjustment (-1.0 to 1.0)
        diminishing: If True, positive deltas are scaled by (1 - current_confidence)
    """
    if diminishing and delta > 0:
        # Diminishing returns: harder to increase confidence as it gets higher
        effective_delta = delta * (1.0 - self.confidence)
    else:
        effective_delta = delta

    return self.with_confidence(self.confidence + effective_delta)

with_error

with_error(error: str) -> AgentState

Record an error.

Source code in .sdk/src/tulip/core/state.py

def with_error(self, error: str) -> AgentState:
    """Record an error."""
    return self.model_copy(
        update={
            "errors": (*self.errors, error),
            "updated_at": datetime.now(UTC),
        }
    )

with_metadata

with_metadata(key: str, value: Any) -> AgentState

Set a metadata value.

Source code in .sdk/src/tulip/core/state.py

def with_metadata(self, key: str, value: Any) -> AgentState:
    """Set a metadata value."""
    return self.model_copy(
        update={
            "metadata": {**self.metadata, key: value},
            "updated_at": datetime.now(UTC),
        }
    )

with_token_usage

with_token_usage(prompt_tokens: int, completion_tokens: int, cache_creation_tokens: int = 0, cache_read_tokens: int = 0, *, cached_tokens: int = 0, cache_write_tokens: int = 0, reported_cost_usd: float | None = None) -> AgentState

Record token usage from a model response.

cache_creation_tokens and cache_read_tokens are populated only when Anthropic returns prompt-cache stats on the response usage (i.e., the AnthropicModel was configured with prompt_cache=True). Default 0 for other providers. cached_tokens and cache_write_tokens are the OpenAI-style counters, already inside prompt_tokens. reported_cost_usd is the provider's own figure for the call, when it gives one.

Source code in .sdk/src/tulip/core/state.py

def with_token_usage(  # noqa: PLR0913 — one keyword per counter a provider reports
    self,
    prompt_tokens: int,
    completion_tokens: int,
    cache_creation_tokens: int = 0,
    cache_read_tokens: int = 0,
    *,
    cached_tokens: int = 0,
    cache_write_tokens: int = 0,
    reported_cost_usd: float | None = None,
) -> AgentState:
    """Record token usage from a model response.

    ``cache_creation_tokens`` and ``cache_read_tokens`` are populated
    only when Anthropic returns prompt-cache stats on the response
    usage (i.e., the AnthropicModel was configured with
    ``prompt_cache=True``). Default 0 for other providers.
    ``cached_tokens`` and ``cache_write_tokens`` are the OpenAI-style
    counters, already inside ``prompt_tokens``. ``reported_cost_usd`` is
    the provider's own figure for the call, when it gives one.
    """
    update: dict[str, Any] = {
        "total_tokens_used": self.total_tokens_used + prompt_tokens + completion_tokens,
        "prompt_tokens_used": self.prompt_tokens_used + prompt_tokens,
        "completion_tokens_used": self.completion_tokens_used + completion_tokens,
        "cache_creation_tokens_used": (self.cache_creation_tokens_used + cache_creation_tokens),
        "cache_read_tokens_used": self.cache_read_tokens_used + cache_read_tokens,
        "cached_tokens_used": self.cached_tokens_used + cached_tokens,
        "cache_write_tokens_used": self.cache_write_tokens_used + cache_write_tokens,
        "cost_usd_used": self.cost_usd_used
        + (
            self.cost_of(
                prompt_tokens,
                completion_tokens,
                cache_creation_tokens=cache_creation_tokens,
                cache_read_tokens=cache_read_tokens,
            )
            or 0.0
        ),
        "updated_at": datetime.now(UTC),
    }
    if reported_cost_usd is not None:
        update["reported_cost_usd"] = self.reported_cost_usd + reported_cost_usd
        update["reported_cost_calls"] = self.reported_cost_calls + 1
    return self.model_copy(update=update)

with_response_usage

with_response_usage(usage: Mapping[str, Any] | None, reported_cost_usd: float | None = None) -> AgentState

Record a model response's usage dict, every counter it carries.

Source code in .sdk/src/tulip/core/state.py

def with_response_usage(
    self, usage: Mapping[str, Any] | None, reported_cost_usd: float | None = None
) -> AgentState:
    """Record a model response's ``usage`` dict, every counter it carries."""
    usage = usage or {}

    def count(key: str) -> int:
        value = usage.get(key, 0)
        return value if isinstance(value, int) and not isinstance(value, bool) else 0

    # Only a real number counts: a test double's attribute is not a price.
    cost = (
        float(reported_cost_usd)
        if isinstance(reported_cost_usd, int | float)
        and not isinstance(reported_cost_usd, bool)
        else None
    )
    return self.with_token_usage(
        count("prompt_tokens"),
        count("completion_tokens"),
        cache_creation_tokens=count("cache_creation_input_tokens"),
        cache_read_tokens=count("cache_read_input_tokens"),
        cached_tokens=count("cached_tokens"),
        cache_write_tokens=count("cache_write_tokens"),
        reported_cost_usd=cost,
    )

cost_of

cost_of(prompt_tokens: int, completion_tokens: int, *, cache_creation_tokens: int = 0, cache_read_tokens: int = 0) -> float | None

USD for a call of this size, or None when the prices are unknown.

cache_creation_tokens and cache_read_tokens are the counts Anthropic (and Bedrock) report beside prompt_tokens, priced at their published multiples of the input price: a write at 1.25x, a read at 0.1x. Left out, as they were, a cached Claude run looked almost free and a max_cost_usd never stopped it. OpenAI-style cached_tokens are inside prompt_tokens and stay at the full input price, since the discount differs by provider: that overstates a cached call rather than understating it.

Source code in .sdk/src/tulip/core/state.py

def cost_of(
    self,
    prompt_tokens: int,
    completion_tokens: int,
    *,
    cache_creation_tokens: int = 0,
    cache_read_tokens: int = 0,
) -> float | None:
    """USD for a call of this size, or ``None`` when the prices are unknown.

    ``cache_creation_tokens`` and ``cache_read_tokens`` are the counts
    Anthropic (and Bedrock) report beside ``prompt_tokens``, priced at
    their published multiples of the input price: a write at 1.25x, a
    read at 0.1x. Left out, as they were, a cached Claude run looked
    almost free and a ``max_cost_usd`` never stopped it. OpenAI-style
    ``cached_tokens`` are inside ``prompt_tokens`` and stay at the full
    input price, since the discount differs by provider: that overstates
    a cached call rather than understating it.
    """
    if self.input_price_per_mtok is None or self.output_price_per_mtok is None:
        return None
    input_tokens = (
        prompt_tokens
        + cache_creation_tokens * _CACHE_WRITE_MULTIPLIER
        + cache_read_tokens * _CACHE_READ_MULTIPLIER
    )
    return (
        input_tokens * self.input_price_per_mtok
        + completion_tokens * self.output_price_per_mtok
    ) / 1_000_000

would_exceed_cost_budget

would_exceed_cost_budget(max_output_tokens: int) -> bool

Whether the next call could cross the spend budget, at its worst case.

The worst case is the current conversation as input (a char/4 estimate) plus max_output_tokens of output. Checked before the call, which is what stops one large turn: a check after the call only stops the next one.

Source code in .sdk/src/tulip/core/state.py

def would_exceed_cost_budget(self, max_output_tokens: int) -> bool:
    """Whether the next call could cross the spend budget, at its worst case.

    The worst case is the current conversation as input (a char/4 estimate)
    plus ``max_output_tokens`` of output. Checked *before* the call, which is
    what stops one large turn: a check after the call only stops the next one.
    """
    if self.cost_budget_usd is None:
        return False
    prompt_estimate = sum(
        estimate_tokens(m.content)
        + sum(len(str(tc.arguments or "")) for tc in m.tool_calls) // 4
        for m in self.messages
    )
    worst = self.cost_of(prompt_estimate, max_output_tokens)
    return worst is not None and self.cost_usd_used + worst > self.cost_budget_usd

tool_loop_warned

tool_loop_warned(loop: ToolLoop) -> bool

Whether the model has already been warned about loop.

Source code in .sdk/src/tulip/core/state.py

def tool_loop_warned(self, loop: ToolLoop) -> bool:
    """Whether the model has already been warned about ``loop``."""
    return any(sig == loop.signature for sig, _ in self.tool_loop_warnings)

with_tool_loop_warning

with_tool_loop_warning(loop: ToolLoop) -> AgentState

Record that the model was warned about loop at this step.

Source code in .sdk/src/tulip/core/state.py

def with_tool_loop_warning(self, loop: ToolLoop) -> AgentState:
    """Record that the model was warned about ``loop`` at this step."""
    warnings = (*self.tool_loop_warnings, (loop.signature, len(self.reasoning_steps)))
    return self.model_copy(update={"tool_loop_warnings": warnings})

to_checkpoint

to_checkpoint() -> dict[str, Any]

Serialize state for checkpointing.

Source code in .sdk/src/tulip/core/state.py

def to_checkpoint(self) -> dict[str, Any]:
    """Serialize state for checkpointing."""
    return self.model_dump(mode="json")

from_checkpoint classmethod

from_checkpoint(data: dict[str, Any]) -> AgentState

Restore state from checkpoint.

Source code in .sdk/src/tulip/core/state.py

@classmethod
def from_checkpoint(cls, data: dict[str, Any]) -> AgentState:
    """Restore state from checkpoint."""
    return cls.model_validate(data)

State sub-types

AgentState.tool_executions is a tuple of ToolExecution, AgentState.reasoning_steps is a tuple of ReasoningStep. Both are surfaced on AgentResult via the matching properties.

ToolExecution

Bases: BaseModel

Record of a single tool execution.

success property

success: bool

Whether the execution succeeded.

ReasoningStep

Bases: BaseModel

A single step in the agent's reasoning trace.

Composition

The composition helpers chain or fan-out multiple agents while keeping the same AgentResult shape at the boundary. See the Multi-agent composition page for the full pipeline classes; the functional builders below are the ergonomic entry points.

PipelineResult

Bases: BaseModel

Result from a pipeline execution.

sequential

sequential(*agents: Any, prompt_template: str | None = None) -> SequentialPipeline

Create a sequential pipeline from agents.

Parameters:

Name	Type	Description	Default
*agents	Any	Agents to run in order	()
prompt_template	str | None	Optional template for passing output between agents	None

Source code in .sdk/src/tulip/agent/composition.py

def sequential(*agents: Any, prompt_template: str | None = None) -> SequentialPipeline:
    """Create a sequential pipeline from agents.

    Args:
        *agents: Agents to run in order
        prompt_template: Optional template for passing output between agents
    """
    kwargs: dict[str, Any] = {"agents": list(agents)}
    if prompt_template:
        kwargs["prompt_template"] = prompt_template
    return SequentialPipeline(**kwargs)

parallel

parallel(*agents: Any, merge_strategy: str = 'concatenate') -> ParallelPipeline

Create a parallel pipeline from agents.

Parameters:

Name	Type	Description	Default
*agents	Any	Agents to run concurrently	()
merge_strategy	str	How to merge results ('concatenate' or 'last')	'concatenate'

Source code in .sdk/src/tulip/agent/composition.py

def parallel(*agents: Any, merge_strategy: str = "concatenate") -> ParallelPipeline:
    """Create a parallel pipeline from agents.

    Args:
        *agents: Agents to run concurrently
        merge_strategy: How to merge results ('concatenate' or 'last')
    """
    return ParallelPipeline(agents=list(agents), merge_strategy=merge_strategy)

loop

loop(agent: Any, condition: Callable[[str], bool], max_loops: int = 5, loop_prompt: str | None = None) -> LoopAgent

Create a loop agent.

Parameters:

Name	Type	Description	Default
agent	Any	Agent to run repeatedly	required
condition	Callable[[str], bool]	Function returning True when loop should stop	required
max_loops	int	Maximum iterations	5
loop_prompt	str | None	Template for loop iteration prompts	None

Source code in .sdk/src/tulip/agent/composition.py

def loop(
    agent: Any,
    condition: Callable[[str], bool],
    max_loops: int = 5,
    loop_prompt: str | None = None,
) -> LoopAgent:
    """Create a loop agent.

    Args:
        agent: Agent to run repeatedly
        condition: Function returning True when loop should stop
        max_loops: Maximum iterations
        loop_prompt: Template for loop iteration prompts
    """
    kwargs: dict[str, Any] = {
        "agent": agent,
        "condition": condition,
        "max_loops": max_loops,
    }
    if loop_prompt:
        kwargs["loop_prompt"] = loop_prompt
    return LoopAgent(**kwargs)