teff.harness.tools

teff.harness.tools

Tool-approval resolution and parallel tool-call execution.

Functions:

Name	Description
execute_tool_calls	Execute tool_calls against tools in parallel.
resolve_approval	Resolve a tool-approval decision for one tool call.

execute_tool_calls async

execute_tool_calls(
    tool_calls,
    tools,
    tool_error_mode="message",
    timeout=None,
    tool_retries=0,
    approver=None,
    state=None,
    ctx=None,
)

Execute tool_calls against tools in parallel.

Each call resolves to a result string (errors become "Error ..." messages unless tool_error_mode is "raise"). Each call is retried up to tool_retries times on failure and bounded by timeout seconds when set. An optional approver gates each call before it runs (see :func:resolve_approval); non-"approve" decisions short-circuit the call with a "not approved" message.

state / ctx are injected into tools that declare __state__ / __ctx__ runtime kwargs (sub-agent tools), so they can read/write the enclosing workflow state and forward tracing.

Parameters:

Name	Type	Description	Default
tool_calls	list[dict]	List of tool-call dicts.	required
tools	Mapping[str, Tool]	Tool registry (name -> Tool).	required
tool_error_mode	str	"message" or "raise".	'message'
timeout	float | None	Per-tool timeout in seconds (None = no limit).	None
tool_retries	int	Extra attempts per tool call after a failure.	0
approver	Any	Approval policy (string or callable).	None
state	dict | None	Workflow state dict to expose to state-aware tools.	None
ctx	Any	:class:~teff.node.context.ExecContext to expose to tools.	None

Source code in teff/harness/tools.py

async def execute_tool_calls(
    tool_calls: list[dict],
    tools: Mapping[str, Tool],
    tool_error_mode: str = "message",
    timeout: float | None = None,
    tool_retries: int = 0,
    approver: typing.Any = None,
    state: dict | None = None,
    ctx: typing.Any = None,
) -> list[str]:
    """Execute *tool_calls* against *tools* in parallel.

    Each call resolves to a result string (errors become ``"Error ..."``
    messages unless *tool_error_mode* is ``"raise"``).  Each call is
    retried up to *tool_retries* times on failure and bounded by *timeout*
    seconds when set.  An optional *approver* gates each call before it
    runs (see :func:`resolve_approval`); non-``"approve"`` decisions
    short-circuit the call with a "not approved" message.

    *state* / *ctx* are injected into tools that declare ``__state__`` /
    ``__ctx__`` runtime kwargs (sub-agent tools), so they can read/write the
    enclosing workflow state and forward tracing.

    Args:
        tool_calls: List of tool-call dicts.
        tools: Tool registry (name -> ``Tool``).
        tool_error_mode: ``"message"`` or ``"raise"``.
        timeout: Per-tool timeout in seconds (``None`` = no limit).
        tool_retries: Extra attempts per tool call after a failure.
        approver: Approval policy (string or callable).
        state: Workflow state dict to expose to state-aware tools.
        ctx: :class:`~teff.node.context.ExecContext` to expose to tools.
    """
    if not tool_calls:
        return []
    return await gather_or_cancel(
        *(
            _run_one_tool_call(
                tc,
                tools,
                tool_error_mode,
                timeout,
                tool_retries,
                approver,
                state,
                ctx,
            )
            for tc in tool_calls
        )
    )

resolve_approval async

resolve_approval(approver, name, args)

Resolve a tool-approval decision for one tool call.

approver may be:

• "auto" (or None) → "approve"
• "deny" → "deny" (no call ever runs)
• "interactive" → prompt the operator on stdin
• a callable (name, args) -> str | bool (sync or async) returning "approve"/"deny"/"pause" (or True/False).

Returns one of "approve", "deny", "pause".

Source code in teff/harness/tools.py

async def resolve_approval(approver: typing.Any, name: str, args: dict) -> str:
    """Resolve a tool-approval decision for one tool call.

    *approver* may be:

    - ``"auto"`` (or ``None``) → ``"approve"``
    - ``"deny"`` → ``"deny"`` (no call ever runs)
    - ``"interactive"`` → prompt the operator on stdin
    - a callable ``(name, args) -> str | bool`` (sync or async)
      returning ``"approve"``/``"deny"``/``"pause"`` (or ``True``/``False``).

    Returns one of ``"approve"``, ``"deny"``, ``"pause"``.
    """
    if approver is None or approver == "auto":
        return "approve"
    if approver == "deny":
        return "deny"
    if approver == "interactive":
        import sys

        sys.stderr.write(
            f"\n[teff] approve tool call '{name}' with args {json.dumps(args)}? [y/N] "
        )
        sys.stderr.flush()
        answer = input().strip().lower()
        return "approve" if answer in ("y", "yes") else "deny"
    if callable(approver):
        result = approver(name, args)
        if inspect.isawaitable(result):
            result = await result
        if isinstance(result, bool):
            return "approve" if result else "deny"
        return str(result).lower()
    return "approve"