ToolContext

Python · package toolnexus · SPEC §1 · python/src/toolnexus/types.py

@dataclass
class ToolContext:
    signal: Optional[Any] = None      # cancellation token — optional / advisory
    timeout: Optional[float] = None   # SECONDS (JS uses milliseconds)
    answer: Optional[Answer] = None   # present ONLY on a post-wait_for retry (§10)

The second, optional argument to execute. Every field is optional and the whole object may be None, so a tool that ignores it still works — but these three things are how a tool participates in cancellation, honours a deadline, and receives the answer to a question it asked.

When to use it

Read ctx when your tool does anything slow or interactive:

signal — long work that should stop when the run is cancelled.
timeout — the caller’s deadline for this specific call, in seconds.
answer — you returned a suspension on a previous attempt and the host has now resolved it.

A pure, fast, local computation can ignore ctx entirely.

Why it is optional, and what that costs you

Examples

1. Respecting cancellation

Check signal before starting work and between steps. A cancelled tool should return promptly rather than raising.

import asyncio
from toolnexus import Tool, ToolResult, ToolContext


async def _crunch(args, ctx=None):
    done = 0
    for _ in range(int(args.get("steps", 3))):
        # signal is advisory in Python — any object exposing a truthy cancellation works.
        if ctx is not None and ctx.signal is not None and ctx.signal.is_set():
            return ToolResult(output=f"cancelled after {done} step(s)", is_error=True)
        done += 1
    return ToolResult(output=f"completed {done} step(s)", is_error=False)


crunch = Tool(
    name="crunch",
    description="Do some work in steps, stopping if cancelled",
    input_schema={"type": "object", "properties": {"steps": {"type": "number"}}},
    source="custom",
    execute=_crunch,
)


async def main():
    # No context at all — the tool still runs.
    plain = await crunch.execute({"steps": 3})
    assert plain.output == "completed 3 step(s)"

    # Cancelled before it starts.
    flag = asyncio.Event()
    flag.set()
    stopped = await crunch.execute({"steps": 3}, ToolContext(signal=flag))
    assert stopped.is_error is True
    assert stopped.output == "cancelled after 0 step(s)"

    print("ok:", plain.output, "|", stopped.output)


asyncio.run(main())

2. Honouring the caller’s timeout

timeout is seconds. It is the caller’s budget for this call — treat it as a ceiling, not a suggestion.

import asyncio
from toolnexus import Tool, ToolResult, ToolContext


async def _fetchish(args, ctx=None):
    # Fall back to your own default when the caller gave no budget.
    budget = ctx.timeout if ctx is not None and ctx.timeout is not None else 30.0
    if budget < 0.1:
        return ToolResult(output=f"budget {budget}s is too small to try", is_error=True)
    return ToolResult(output=f"fetched {args['url']} within {budget}s", is_error=False)


fetchish = Tool(
    name="fetchish",
    description="Pretend to fetch, bounded by the caller's timeout",
    input_schema={"type": "object", "properties": {"url": {"type": "string"}}},
    source="custom",
    execute=_fetchish,
)


async def main():
    generous = await fetchish.execute({"url": "/a"}, ToolContext(timeout=5.0))
    assert generous.output == "fetched /a within 5.0s"

    stingy = await fetchish.execute({"url": "/a"}, ToolContext(timeout=0.01))
    assert stingy.is_error is True

    defaulted = await fetchish.execute({"url": "/a"})
    assert "30.0s" in defaulted.output

    print("ok:", generous.output, "|", stingy.output)


asyncio.run(main())

3. `answer` — the second half of a suspension

This is the field that makes the human-in-the-loop contract work. On the first call the tool returns a pending. The host resolves it, then calls the same tool again with ctx.answer set. The tool branches on whether the answer is there.

import asyncio
from toolnexus import Tool, ToolResult, ToolContext, Answer, pending, pending_of


async def _deploy(args, ctx=None):
    # Second pass: the host resolved the question and handed the answer back.
    if ctx is not None and ctx.answer is not None:
        if not ctx.answer.ok:
            reason = ctx.answer.reason or "no reason"
            return ToolResult(output=f"declined: {reason}", is_error=True)
        env = (ctx.answer.data or {}).get("env", "unknown")
        return ToolResult(output=f"deployed to {env}", is_error=False)
    # First pass: park the run and ask.
    return pending(kind="input", prompt="Which environment?")


deploy = Tool(
    name="deploy",
    description="Deploy, asking which environment first",
    input_schema={"type": "object", "properties": {}},
    source="custom",
    execute=_deploy,
)


async def main():
    # First pass — a suspension, not an answer.
    first = await deploy.execute({})
    req = pending_of(first)
    assert req is not None
    assert req.kind == "input"

    # Second pass — the host supplies the resolution, echoing the request id.
    second = await deploy.execute({}, ToolContext(answer=Answer(id=req.id, ok=True, data={"env": "staging"})))
    assert second.is_error is False
    assert second.output == "deployed to staging"

    # A refusal is a normal outcome, not a crash.
    refused = await deploy.execute({}, ToolContext(answer=Answer(id=req.id, ok=False, reason="declined")))
    assert refused.is_error is True

    print("ok:", second.output, "|", refused.output)


asyncio.run(main())

Fields

Field	Type	What it is
`signal`	`Any \| None`	Cancellation token — advisory. Guard with `ctx is not None`.
`timeout`	`float \| None`	This call’s budget, in seconds.
`answer`	`Answer \| None`	Present only on a post-`wait_for` retry — the resolution of a prior suspension.