AgentRuntime.resume

Python · package toolnexus · SPEC §7D / §10 · python/src/toolnexus/agents/runtime.py

async def resume(self, answer: Answer) -> None

Manifest correction

The cross-language API surface manifest lists this entry point as Client.resume for every port. That symbol does not exist in python/src/toolnexus/client.py — nor in any of the other five ports’ bare-client source. The durable, answer-carrying entry point that actually ships in Python (and identically in JS/Java/C#/Elixir) is the §7D agent runtime’s AgentRuntime.resume(answer) documented on this page. Client-level run_with_answer / ask_with_answer (SPEC §10 “Durable resume”) is a Go-only preview (golang/relay.go), not part of the six-port conformance contract yet — see suspension/relay for the same Go-only caveat on its sibling feature.

The §7D counterpart to Client’s in-process wait_for: when a spawned agent’s turn halts because a tool suspended and no wait_for was reachable in its ancestor chain (durable), the Agent.run/AgentRuntime.wait result comes back status="pending" carrying both the Request and the live runtime that produced it. Call await runtime.resume(answer) once you have an Answer (answer.id should echo pending.id) — it routes to the deepest suspended handle, replays that handle’s exact suspended turn with ctx.answer set, and cascades upward: any suspended ancestor’s turn re-runs too, its retried task call reattaching to the same child by task key rather than spawning a duplicate.

When to use it

• A §7D agent (or team of agents) suspended durably — no wait_for anywhere in its ancestor chain — and you now have the human’s/system’s answer in hand and want the run to continue.
• You’re building a host around Agent.run/AgentRuntime that separates “detect a suspension” from “resolve it” — e.g. surface result.pending.prompt to a UI immediately, then call resume only once the UI reports back, potentially turns later.
• The suspension happened several levels deep in a delegated team: resume finds the deepest suspended handle for you — you never have to walk the tree yourself.

Why this and not the alternative

Reach for wait_for when you want it to just not hang

resume is the other half of the durable path from WaitFor: give an agent’s definition a wait_for and its own suspensions resolve inline, in the same await, no resume call needed. Reach for resume when there’s no wait_for in the chain — by design (you want the halt to surface immediately) or by omission — and something external now has the answer.

Because Request/Answer are plain, dataclasses-serializable data (SPEC §10), a host can hand result.pending off to a UI, a queue, or a channel message the moment it appears, then construct an Answer from whatever comes back — resume doesn’t care where the Answer originated, only that it echoes back to the same runtime that is still holding the suspended tree.

Examples

1. The smallest useful call — suspend, then resume in place

import asyncio

from toolnexus import Answer, ToolResult, define_tool, pending
from toolnexus.agents import agent


class ScriptedTransport:
    """Canned OpenAI-shaped replies, popped per model — no network, no real LLM."""

    def __init__(self, scripts):
        self._scripts = {k: list(v) for k, v in scripts.items()}

    def post(self, url, headers, payload, timeout):
        return self._scripts[payload.get("model")].pop(0)

    def open(self, url, headers, payload, timeout):
        raise NotImplementedError


def tool_call_reply():
    return {
        "choices": [{
            "message": {
                "role": "assistant", "content": None,
                "tool_calls": [{"id": "c1", "type": "function",
                                 "function": {"name": "approve", "arguments": "{}"}}],
            },
            "finish_reason": "tool_calls",
        }],
        "usage": {"prompt_tokens": 6, "completion_tokens": 2, "total_tokens": 8},
    }


def final_reply(text):
    return {
        "choices": [{"message": {"role": "assistant", "content": text}}],
        "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
    }


async def approve(args=None, ctx=None):
    if ctx is not None and ctx.answer is not None and ctx.answer.ok:
        return ToolResult(output="approved and deployed", is_error=False)
    return pending(kind="approval", prompt="OK to deploy?")


async def main():
    approve_tool = define_tool(approve, name="approve", description="Deploy after approval.")
    approver = agent("approver", does="deploys after human approval", uses={"tools": [approve_tool]})

    # The resumed turn REPLAYS from a rolled-back transcript (§10 idempotency), so the
    # model calls the tool a second time before it ever sees the final answer — that's
    # why `tool_call_reply` appears twice: once to trigger the original suspension, once
    # on the resumed replay, where `wait_for` now resolves it inline.
    scripts = {"stub-model": [tool_call_reply(), tool_call_reply(), final_reply("deployed.")]}
    r = await approver.run(
        "deploy the latest build",
        transport=ScriptedTransport(scripts),
        llm={"base_url": "http://mock.local", "style": "openai", "model": "stub-model", "api_key": "unused"},
    )

    assert r.status == "pending"
    assert r.pending.kind == "approval"
    assert r.runtime is not None  # the live runtime, holding the suspended tree

    answer = Answer(id=r.pending.id, ok=True)
    await r.runtime.resume(answer)

    # The spawned handle is the runtime root's only child (Agent.run spawned it there).
    handle = r.runtime.root.children[0]
    final = await r.runtime.wait(handle)

    assert final.status == "done"
    assert final.text == "deployed."

    await r.runtime.close(r.runtime.root)
    print("ok:", final.status, "|", final.text)


asyncio.run(main())

2. The realistic case — Request/Answer as plain data, round-tripped through JSON

import asyncio
import dataclasses
import json

from toolnexus import Answer, ToolResult, define_tool, pending
from toolnexus.agents import agent


class ScriptedTransport:
    def __init__(self, scripts):
        self._scripts = {k: list(v) for k, v in scripts.items()}

    def post(self, url, headers, payload, timeout):
        return self._scripts[payload.get("model")].pop(0)

    def open(self, url, headers, payload, timeout):
        raise NotImplementedError


def tool_call_reply():
    return {
        "choices": [{
            "message": {
                "role": "assistant", "content": None,
                "tool_calls": [{"id": "c1", "type": "function",
                                 "function": {"name": "refund", "arguments": "{}"}}],
            },
            "finish_reason": "tool_calls",
        }],
        "usage": {"prompt_tokens": 6, "completion_tokens": 2, "total_tokens": 8},
    }


def final_reply(text):
    return {
        "choices": [{"message": {"role": "assistant", "content": text}}],
        "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
    }


async def refund(args=None, ctx=None):
    if ctx is not None and ctx.answer is not None and ctx.answer.ok:
        amount = (ctx.answer.data or {}).get("amount", 0)
        return ToolResult(output=f"refunded ${amount}", is_error=False)
    return pending(kind="input", prompt="How much should be refunded?")


async def main():
    refund_tool = define_tool(refund, name="refund", description="Issue a refund.")
    support = agent("support", does="handles refunds", uses={"tools": [refund_tool]})

    # Replayed on resume — the model calls the tool again before the final answer.
    scripts = {"stub-model": [tool_call_reply(), tool_call_reply(), final_reply("refund processed.")]}
    r = await support.run(
        "refund the last order",
        transport=ScriptedTransport(scripts),
        llm={"base_url": "http://mock.local", "style": "openai", "model": "stub-model", "api_key": "unused"},
    )
    assert r.status == "pending"

    # Request/Answer are dataclasses — plain, JSON-serializable data (SPEC §10). A durable
    # host can persist exactly this and reconstruct it from a queue, a webhook, another
    # process's message — anything that eventually hands the same shape back.
    wire = json.dumps(dataclasses.asdict(r.pending))
    restored = json.loads(wire)
    answer = Answer(id=restored["id"], ok=True, data={"amount": 40})

    await r.runtime.resume(answer)
    handle = r.runtime.root.children[0]
    final = await r.runtime.wait(handle)

    assert final.status == "done"
    assert final.text == "refund processed."

    await r.runtime.close(r.runtime.root)
    print("ok:", wire[:40], "... ->", final.text)


asyncio.run(main())

3. The full surface — a TaskResult’s numbers are per-call; the handle’s are cumulative

TaskResult.turns/.total_tokens report the ask() call that just finished — for the resumed call, that means the replay’s own turns (the repeated tool call plus its continuation), not a running total since spawn. For a lifetime total, read runtime.inspect(handle).tokens — the Handle’s own counters, which really do only grow, across every turn since spawn.

import asyncio

from toolnexus import Answer, ToolResult, define_tool, pending
from toolnexus.agents import agent


class ScriptedTransport:
    def __init__(self, scripts):
        self._scripts = {k: list(v) for k, v in scripts.items()}

    def post(self, url, headers, payload, timeout):
        return self._scripts[payload.get("model")].pop(0)

    def open(self, url, headers, payload, timeout):
        raise NotImplementedError


def tool_call_reply():
    return {
        "choices": [{
            "message": {
                "role": "assistant", "content": None,
                "tool_calls": [{"id": "c1", "type": "function",
                                 "function": {"name": "gate", "arguments": "{}"}}],
            },
            "finish_reason": "tool_calls",
        }],
        "usage": {"prompt_tokens": 6, "completion_tokens": 2, "total_tokens": 8},
    }


def final_reply(text):
    return {
        "choices": [{"message": {"role": "assistant", "content": text}}],
        "usage": {"prompt_tokens": 3, "completion_tokens": 1, "total_tokens": 4},
    }


async def gate(args=None, ctx=None):
    if ctx is not None and ctx.answer is not None and ctx.answer.ok:
        return ToolResult(output="through the gate", is_error=False)
    return pending(kind="approval", prompt="let it through?")


async def main():
    gate_tool = define_tool(gate, name="gate", description="Gated action.")
    worker = agent("worker", does="does gated work", uses={"tools": [gate_tool]})

    # Replayed on resume: tool call, tool call again (resolved inline), final answer.
    scripts = {"stub-model": [tool_call_reply(), tool_call_reply(), final_reply("done.")]}
    r = await worker.run(
        "do the gated thing",
        transport=ScriptedTransport(scripts),
        llm={"base_url": "http://mock.local", "style": "openai", "model": "stub-model", "api_key": "unused"},
    )
    assert r.status == "pending"
    handle = r.runtime.root.children[0]
    tokens_before = r.runtime.inspect(handle).tokens  # the suspending attempt already rolled up

    await r.runtime.resume(Answer(id=r.pending.id, ok=True))
    final = await r.runtime.wait(handle)

    assert final.status == "done"
    assert final.turns == 2  # the resumed replay's OWN two turns — not a running total

    tokens_after = r.runtime.inspect(handle).tokens
    assert tokens_after > tokens_before  # the Handle's lifetime counter really does only grow

    await r.runtime.close(r.runtime.root)
    print("ok: handle tokens", tokens_before, "->", tokens_after, "| resumed turns:", final.turns)


asyncio.run(main())

TaskResult fields relevant to resume

Field	Type	What it is
status	str	"pending" when durably suspended — one of the closed §7D vocabulary (done/pending/incomplete/interrupted/closed/timeout/error).
pending	Request | None	Set iff status == "pending". Request.data["path"] carries the suspended subtree’s id path.
runtime	AgentRuntime | None	Set by Agent.run() — hold onto it to call resume later.
turns / total_tokens	int	The ask() call that just completed — the resumed replay’s own numbers, not a running total. Use runtime.inspect(handle).tokens for the handle’s lifetime total.

See also

• pending — Return a Pending from a tool to park the run until someone answers.
• auth_required — The auth-shaped suspension: hand back a URL, resume once the user has granted access.
• WaitFor — The single hook where the host resolves a suspension — in-process prompt or durable queue, same contract.
• pending_of — Detect that a RunResult is parked rather than finished, and get the Request that parked it.