teff.harness

teff.harness

Agent harness — reusable model↔tool loop.

A harness owns the transport and provider plumbing for one model and drives the agent loop: call the model, execute requested tools, feed the results back into the conversation. It is shared by the :class:~teff.node.llm.LLM node (internal multi-round loop) and the :class:~teff.node.agent.ReActAgent (one step per graph round, so the loop stays visible as topology). Tools are ordinary :class:~teff.tool.Tool instances keyed by name, so MCP tools and built-in tools work unchanged.

Behaviour can be parameterised through the constructor / ``from_config``:

- ``max_rounds`` — stop the ``run()`` loop after this many model calls.
- ``stop_when(messages)`` — extra termination predicate.
- ``parse_text_tool_calls`` — decode tool calls embedded in plain text
  (local models often skip the structured ``tool_calls`` field).
- ``tool_error_mode`` — ``"message"`` (default, errors become tool
  messages) or ``"raise"`` (a tool failure propagates, e.g. into an
  ``__error__`` edge).
- ``tool_timeout`` — per-tool execution timeout in seconds.
- ``tool_retries`` — extra attempts per tool call after a failure.
- ``max_retries`` — retry failed HTTP requests (429/5xx/timeouts).
- ``retry_on`` — status codes / error types worth retrying.
- ``fallbacks`` — list of fallback model names used when the primary
  transport fails (provider failover).
- ``max_total_tokens`` — stop the loop once total prompt+completion
  tokens exceed this budget.
- ``max_context_tokens`` / ``max_context_messages`` — trim the
  conversation history before each model call to fit these limits.
- ``cache`` — cache model responses keyed by request so re-runs /
  checkpoint resumes do not pay for the same call twice.
- ``on_tool_call`` — async hook ``(name, args) -> Awaitable[None]``
  invoked before each tool executes (approval/auditing).
- ``on_step`` / ``on_llm`` / ``on_token`` — observability hooks.

Modules:

Name	Description
context	Context management — token estimation and message trimming.
formats	Response parsing and message-format normalisation for LLM providers.
loop	The Harness — transport + agent loop for a single model.
providers	Backward-compatible alias for :mod:teff.provider.providers.
schema	Tool schema conversion — Python signatures to OpenAI-style function schemas.
tools	Tool-approval resolution and parallel tool-call execution.

Classes:

Name	Description
ContextLimitError	Raised when a conversation cannot fit the configured context limits.
Harness	Transport + loop for one model, reusable across nodes and flows.
ModelReply	A single model call's result.
Provider	A named model endpoint: wire protocol + endpoint data.
Step	One iteration of the agent loop (model call + any tool execution).

Functions:

Name	Description
execute_tool_calls	Execute tool_calls against tools in parallel.
extract_content	Extract the assistant text from a response.
extract_message	Normalise response formats to {role, content, tool_calls}.
extract_usage	Extract (prompt_tokens, completion_tokens) from an LLM response.
normalize_text_tool_calls	Turn a text-embedded tool call into the structured tool_calls list.
parse_text_tool_call	Parse a tool call embedded in plain text content.
provider_concurrency	Return the current global concurrency limit for provider (if any).
resolve_approval	Resolve a tool-approval decision for one tool call.
resolve_provider	Resolve a provider key from an explicit value or a default name.
resolve_provider_entry	Resolve the effective :class:Provider for provider_key.
set_provider_concurrency	Globally cap concurrent model calls for provider.
tool_to_schema	Convert a :class:~teff.tool.Tool to an OpenAI-style function schema.
trim_messages	Trim messages down to fit context limits.

ContextLimitError

Bases: WorkflowError

Raised when a conversation cannot fit the configured context limits.

Source code in teff/harness/context.py

class ContextLimitError(WorkflowError):
    """Raised when a conversation cannot fit the configured context limits."""

Harness

Transport + loop for one model, reusable across nodes and flows.

Parameters:

Name	Type	Description	Default
model	str	Model name (e.g. gpt-4, llama3.1:8b).	required
provider	str | None	Provider name ("openai", "ollama", etc.). Falls back to default_provider when unset.	None
base_url / api_key_env / chat_path / auth_header / auth_prefix		Overrides for the provider defaults.	required
providers	'dict[str, Provider] | ProviderRegistry | None'	Optional {name: Provider} map or :class:~teff.provider.ProviderRegistry (custom providers declared in a workflow / passed to graph.run). Entries are resolved before the built-in presets.	None
timeout	float | None	HTTP timeout in seconds.	120
max_rounds	int	Maximum model calls for :meth:run.	10
parse_text_tool_calls	bool	Decode text-embedded tool calls.	True
tool_error_mode	str	"message" or "raise".	'message'
stop_when	Callable[[list[dict]], bool] | None	Optional (messages) -> bool termination predicate.	None
on_step	Callable[[Step], Awaitable[None]] | None	Async callback (Step) -> None after each iteration.	None
on_llm	Callable[[str, str, int, int, float], Awaitable[None]] | None	Async callback (provider, model, prompt_tokens, completion_tokens, latency_ms) -> None after each model call.	None
on_token	Callable[[str], Awaitable[None]] | None	Token callback for streaming.	None
temperature / max_tokens / response_format		Default body extras.	required
stream	bool	Stream tokens by default (disabled while tools are active).	False
default_provider	str | None	Fallback provider name (the graph-level default, e.g. Graph(default_provider="ollama") or a workflow default_provider:).	None

Methods:

Name	Description
call	One model call.
from_config	Build a harness from a node config dict.
manage_context	Trim messages to the configured context limits.
run	Loop :meth:step until a final answer, stop_when, or max_rounds.
step	One iteration: call the model, execute requested tools, feed back.

Source code in teff/harness/loop.py

class Harness:
    """Transport + loop for one model, reusable across nodes and flows.

    Args:
        model: Model name (e.g. ``gpt-4``, ``llama3.1:8b``).
        provider: Provider name (``"openai"``, ``"ollama"``, etc.).
            Falls back to *default_provider* when unset.
        base_url / api_key_env / chat_path / auth_header / auth_prefix:
            Overrides for the provider defaults.
        providers: Optional ``{name: Provider}`` map or
            :class:`~teff.provider.ProviderRegistry` (custom providers
            declared in a workflow / passed to ``graph.run``).  Entries
            are resolved before the built-in presets.
        timeout: HTTP timeout in seconds.
        max_rounds: Maximum model calls for :meth:`run`.
        parse_text_tool_calls: Decode text-embedded tool calls.
        tool_error_mode: ``"message"`` or ``"raise"``.
        stop_when: Optional ``(messages) -> bool`` termination predicate.
        on_step: Async callback ``(Step) -> None`` after each iteration.
        on_llm: Async callback ``(provider, model, prompt_tokens,
            completion_tokens, latency_ms) -> None`` after each model call.
        on_token: Token callback for streaming.
        temperature / max_tokens / response_format: Default body extras.
        stream: Stream tokens by default (disabled while tools are active).
        default_provider: Fallback provider name (the graph-level default,
            e.g. ``Graph(default_provider="ollama")`` or a workflow
            ``default_provider:``).
    """

    def __init__(
        self,
        *,
        model: str,
        provider: str | None = None,
        providers: "dict[str, Provider] | ProviderRegistry | None" = None,
        base_url: str = "",
        api_key_env: str = "",
        chat_path: str = "",
        auth_header: str = "",
        auth_prefix: str = "",
        timeout: float | None = 120,
        max_rounds: int = 10,
        parse_text_tool_calls: bool = True,
        tool_error_mode: str = "message",
        tool_timeout: float | None = None,
        tool_retries: int = 0,
        max_retries: int = 2,
        retry_on: tuple[int, ...] = (429, 500, 502, 503, 504),
        fallbacks: list[str] | None = None,
        tool_approval: typing.Any = None,
        max_total_tokens: int | None = None,
        max_context_tokens: int | None = None,
        max_context_messages: int | None = None,
        max_parallel: int | None = None,
        stop_when: Callable[[list[dict]], bool] | None = None,
        on_step: Callable[[Step], Awaitable[None]] | None = None,
        on_llm: Callable[[str, str, int, int, float], Awaitable[None]] | None = None,
        on_llm_payload: Callable[
            [str, str, list[dict], str, dict, float, bool], Awaitable[None]
        ]
        | None = None,
        on_token: Callable[[str], Awaitable[None]] | None = None,
        on_tool_call: Callable[[str, dict], Awaitable[None]] | None = None,
        temperature: float | None = None,
        max_tokens: int | None = None,
        response_format: dict | None = None,
        stream: bool = False,
        cache: "MutableMapping[str, str] | bool | None" = None,
        default_provider: str | None = None,
    ):
        self.model = model
        self.timeout = timeout or 120
        self.max_rounds = max_rounds
        self.parse_text_tool_calls = parse_text_tool_calls
        self.tool_error_mode = tool_error_mode
        self.tool_timeout = tool_timeout
        self.tool_retries = tool_retries
        self.max_retries = max_retries
        self.retry_on = tuple(retry_on or ())
        self.fallbacks = list(fallbacks or [])
        self.max_parallel: int | None = max_parallel
        self._tool_approval = tool_approval
        self.max_total_tokens = max_total_tokens
        self.max_context_tokens = max_context_tokens
        self.max_context_messages = max_context_messages
        self.stop_when = stop_when
        self.on_step = on_step
        self.on_llm = on_llm
        self.on_llm_payload = on_llm_payload
        self.on_token = on_token
        self.on_tool_call = on_tool_call
        self.stream = stream
        self._cache: MutableMapping[str, str] | None = None
        if isinstance(cache, bool):
            if cache:
                # ``True`` shares one process-wide store so distinct harnesses
                # (per-node/per-run instances) reuse the same responses.
                self._cache = _DEFAULT_CACHE
        elif cache is not None:
            self._cache = cache

        self.provider_key = resolve_provider(provider, default_provider)
        entry = resolve_provider_entry(self.provider_key, providers)
        self.type = entry.type
        if not self.timeout or self.timeout <= 0:
            self.timeout = entry.timeout or 120

        resolved_url = base_url or os.environ.get(
            f"{self.provider_key.upper()}_BASE_URL", entry.base_url
        )
        resolved_env = api_key_env or entry.api_key_env
        resolved_path = chat_path or entry.chat_path

        api_key = ""
        if resolved_env:
            api_key = os.environ.get(resolved_env, "")
        if not api_key:
            api_key = os.environ.get("LLM_API_KEY", "")

        headers = {"Content-Type": "application/json"}
        hdr_name = auth_header or entry.auth_header
        if hdr_name and api_key:
            hdr_prefix = auth_prefix or entry.auth_prefix
            headers[hdr_name] = f"{hdr_prefix}{api_key}"

        self._url = f"{resolved_url}{resolved_path}"
        self._headers = headers

        self._body_extra: dict = {}
        if temperature is not None:
            self._body_extra["temperature"] = temperature
        if max_tokens is not None:
            self._body_extra["max_tokens"] = max_tokens
        if response_format is not None:
            self._body_extra["response_format"] = response_format

        # Provider failover: build fallback transports lazily.  Each
        # fallback is described by (model, url, headers) and tried in
        # order when the primary request fails after all retries.
        self._fallback_transports: list[tuple[str, str, dict]] = []
        for fb_model in self.fallbacks:
            fb_provider = self.provider_key
            fb_entry = resolve_provider_entry(fb_provider, providers)
            fb_url = base_url or os.environ.get(
                f"{fb_provider.upper()}_BASE_URL", fb_entry.base_url
            )
            fb_env = api_key_env or fb_entry.api_key_env
            fb_path = chat_path or fb_entry.chat_path
            fb_key = ""
            if fb_env:
                fb_key = os.environ.get(fb_env, "")
            if not fb_key:
                fb_key = os.environ.get("LLM_API_KEY", "")
            fb_headers = {"Content-Type": "application/json"}
            fb_hdr = auth_header or fb_entry.auth_header
            if fb_hdr and fb_key:
                fb_prefix = auth_prefix or fb_entry.auth_prefix
                fb_headers[fb_hdr] = f"{fb_prefix}{fb_key}"
            self._fallback_transports.append(
                (fb_model, f"{fb_url}{fb_path}", fb_headers)
            )

        # Token budget tracking across calls.
        self.total_tokens = 0

        # Register the per-provider concurrency guard (grows global cap).
        self._concurrency_semaphore()

    @classmethod
    def from_config(
        cls,
        cfg: dict,
        *,
        default_provider: str | None = None,
        default_model: str | None = None,
        providers: "dict[str, Provider] | ProviderRegistry | None" = None,
    ) -> "Harness":
        """Build a harness from a node config dict.

        Recognises the transport keys shared by ``LLM`` and
        ``ReActAgent`` plus the loop knobs ``max_tool_rounds``,
        ``tool_error_mode``, ``parse_text_tool_calls``, ``tool_timeout``,
        ``tool_retries``, ``max_retries``, ``fallbacks``,
        ``max_total_tokens``, ``max_context_tokens`` and
        ``max_context_messages``.

        *providers* is an optional ``{name: Provider}`` map or
        :class:`~teff.provider.ProviderRegistry` (custom providers from
        the workflow) consulted before the built-in presets.

        The model name comes from ``cfg["model"]`` or, when absent,
        *default_model* (the graph-level default).  When neither is set a
        :class:`ConfigError` is raised — there is no silent model default.
        """
        model = cfg.get("model") or default_model
        if not model:
            from teff.errors import ConfigError

            raise ConfigError(
                "no model configured: set `model=` on the node or pass "
                "`default_model=` to the graph / `default_model:` in the "
                "workflow"
            )
        return cls(
            model=str(model),
            provider=cfg.get("provider"),
            providers=providers,
            base_url=cfg.get("base_url") or "",
            api_key_env=cfg.get("api_key_env") or "",
            chat_path=cfg.get("chat_path") or "",
            auth_header=cfg.get("auth_header") or "",
            auth_prefix=cfg.get("auth_prefix") or "",
            timeout=_opt_float(cfg.get("timeout")),
            max_rounds=_cfg_int(cfg, "max_tool_rounds", 10),
            parse_text_tool_calls=bool(cfg.get("parse_text_tool_calls", True)),
            tool_error_mode=str(cfg.get("tool_error_mode", "message")),
            tool_timeout=_opt_float(cfg.get("tool_timeout")),
            tool_retries=_cfg_int(cfg, "tool_retries", 0),
            max_retries=_cfg_int(cfg, "max_retries", 2),
            retry_on=tuple(int(x) for x in cfg.get("retry_on") or ())
            or (429, 500, 502, 503, 504),
            fallbacks=cfg.get("fallbacks"),
            tool_approval=cfg.get("tool_approval"),
            max_total_tokens=_opt_int(cfg.get("max_total_tokens")),
            max_context_tokens=_opt_int(cfg.get("max_context_tokens")),
            max_context_messages=_opt_int(cfg.get("max_context_messages")),
            max_parallel=_opt_int(cfg.get("max_parallel")),
            stop_when=cfg.get("stop_when"),
            on_step=cfg.get("on_step"),
            on_llm=cfg.get("on_llm"),
            on_token=cfg.get("on_token"),
            on_tool_call=cfg.get("on_tool_call"),
            temperature=cfg.get("temperature"),
            max_tokens=cfg.get("max_tokens"),
            response_format=cfg.get("response_format"),
            stream=bool(cfg.get("stream", False)),
            cache=cfg.get("cache"),
            default_provider=default_provider,
        )

    def _body(self, messages: list[dict], tools: list[dict] | None = None) -> dict:
        if self.type == "anthropic_compatible":
            return self._anthropic_body(messages, tools)
        body: dict = {"model": self.model, "messages": messages, **self._body_extra}
        if tools:
            body["tools"] = tools
        return body

    def _anthropic_body(self, messages: list[dict], tools: list[dict] | None) -> dict:
        """Build an Anthropic ``/messages`` request body.

        Splits ``system`` out to the top level, converts tool results and
        assistant ``tool_calls`` into content blocks, and rewrites the
        tool schemas into Anthropic's ``input_schema`` shape.
        """
        system = "\n".join(
            str(m.get("content", "")) for m in messages if m.get("role") == "system"
        )
        body: dict = {
            "model": self.model,
            "messages": [
                self._to_anthropic_message(m)
                for m in messages
                if m.get("role") != "system"
            ],
            "max_tokens": self._body_extra.get("max_tokens") or 1024,
        }
        if self._body_extra.get("temperature") is not None:
            body["temperature"] = self._body_extra["temperature"]
        if self._body_extra.get("response_format") is not None:
            body["response_format"] = self._body_extra["response_format"]
        if system:
            body["system"] = system
        if tools:
            body["tools"] = [self._to_anthropic_tool(t) for t in tools]
        return body

    @staticmethod
    def _to_anthropic_message(msg: dict) -> dict:
        """Convert an OpenAI-shaped message into an Anthropic one."""
        role = msg.get("role")
        if role == "tool":
            return {
                "role": "user",
                "content": [
                    {
                        "type": "tool_result",
                        "tool_use_id": msg.get("tool_call_id", ""),
                        "content": msg.get("content", ""),
                    }
                ],
            }
        if role == "assistant" and msg.get("tool_calls"):
            blocks: list[dict] = []
            content = msg.get("content")
            if content:
                blocks.append({"type": "text", "text": str(content)})
            for tc in msg["tool_calls"]:
                name, raw, call_id = _tool_call_parts(tc)
                try:
                    args = json.loads(raw) if raw else {}
                except json.JSONDecodeError:
                    args = {}
                blocks.append(
                    {"type": "tool_use", "id": call_id, "name": name, "input": args}
                )
            return {"role": "assistant", "content": blocks}
        return {"role": role, "content": msg.get("content", "")}

    @staticmethod
    def _to_anthropic_tool(tool: dict) -> dict:
        """Convert an OpenAI function schema into an Anthropic tool schema."""
        fn = tool.get("function", tool)
        return {
            "name": fn.get("name", ""),
            "description": fn.get("description", ""),
            "input_schema": fn.get("parameters", {"type": "object", "properties": {}}),
        }

    def _is_retryable(self, exc: Exception) -> bool:
        """Whether an HTTP exception should be retried per *retry_on*."""
        if isinstance(exc, (httpx.TransportError, httpx.TimeoutException)):
            return True
        status = getattr(exc, "response", None)
        code = getattr(status, "status_code", None)
        return code is not None and code in self.retry_on

    def _concurrency_semaphore(self) -> asyncio.Semaphore | None:
        """Global semaphore for this provider.

        An explicit cap (``set_provider_concurrency``) is authoritative;
        otherwise the shared semaphore grows to the largest ``max_parallel``
        any harness has configured for the provider.  Growth only replaces an
        idle semaphore, so in-flight requests never exceed the new cap
        (replacing a contended semaphore would let old + new holders run
        concurrently past the limit).
        """
        key = self.provider_key
        if key in _EXPLICIT_LIMITS:
            return _PROVIDER_SEMAPHORES.get(key)
        if self.max_parallel and self.max_parallel > 0:
            sem = _PROVIDER_SEMAPHORES.get(key)
            current = _PROVIDER_LIMITS.get(key, 0)
            if current >= self.max_parallel:
                return sem
            # Grow only while no request is in flight (idle).  Replacing a
            # contended semaphore would let old + new holders run past the cap.
            if sem is None or _PROVIDER_ACTIVE.get(key, 0) == 0:
                sem = asyncio.Semaphore(self.max_parallel)
                _PROVIDER_SEMAPHORES[key] = sem
                _PROVIDER_LIMITS[key] = self.max_parallel
                _PROVIDER_ACTIVE[key] = 0
            return _PROVIDER_SEMAPHORES[key]
        return None

    async def _post_with_retries(
        self, url: str, headers: dict, body: dict, *, allow_fallback: bool = True
    ) -> tuple[dict, bool]:
        """POST *body* with retries + backoff + failover.

        Returns ``(data, used_fallback)`` where *used_fallback* is ``True``
        when the response came from a fallback model rather than the primary.

        The provider semaphore is held for the whole retry cycle (including
        the backoff sleeps) — a slow/failing request keeps one slot so healthy
        in-flight calls never exceed the agreed cap.
        """
        key = self.provider_key
        sem = self._concurrency_semaphore()
        if sem is None:
            return await self._post_with_retries_impl(
                url, headers, body, allow_fallback=allow_fallback
            )
        _PROVIDER_ACTIVE[key] = _PROVIDER_ACTIVE.get(key, 0) + 1
        try:
            async with sem:
                return await self._post_with_retries_impl(
                    url, headers, body, allow_fallback=allow_fallback
                )
        finally:
            _PROVIDER_ACTIVE[key] = max(0, _PROVIDER_ACTIVE.get(key, 0) - 1)

    async def _post_with_retries_impl(
        self, url: str, headers: dict, body: dict, *, allow_fallback: bool = True
    ) -> tuple[dict, bool]:
        last_exc: Exception | None = None
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            for attempt in range(self.max_retries + 1):
                try:
                    response = await client.post(url, headers=headers, json=body)
                    response.raise_for_status()
                    return response.json(), False
                except Exception as exc:  # noqa: BLE001 — retry policy drives handling
                    last_exc = exc
                    if not self._is_retryable(exc) or attempt >= self.max_retries:
                        break
                    await asyncio.sleep(min(4.0, 0.5 * (2**attempt)))
        # Primary transport exhausted — try fallback models (once each).
        if allow_fallback and last_exc is not None and self._fallback_transports:
            for fb_model, fb_url, fb_headers in self._fallback_transports:
                try:
                    data, _ = await self._post_with_retries_impl(
                        fb_url,
                        fb_headers,
                        {**body, "model": fb_model},
                        allow_fallback=False,
                    )
                    return data, True
                except Exception as exc:  # noqa: BLE001
                    last_exc = exc
        assert last_exc is not None
        raise last_exc

    async def _post_stream_with_retries(
        self, url: str, headers: dict, body: dict, *, allow_fallback: bool = True
    ) -> tuple[str, dict]:
        """Stream a POST response with retries + backoff + failover.

        Returns ``(content, usage)``; *usage* carries the provider-reported
        token counts when the final chunk included them, otherwise ``{}``.
        """
        key = self.provider_key
        sem = self._concurrency_semaphore()
        if sem is None:
            return await self._post_stream_with_retries_impl(
                url, headers, body, allow_fallback=allow_fallback
            )
        _PROVIDER_ACTIVE[key] = _PROVIDER_ACTIVE.get(key, 0) + 1
        try:
            async with sem:
                return await self._post_stream_with_retries_impl(
                    url, headers, body, allow_fallback=allow_fallback
                )
        finally:
            _PROVIDER_ACTIVE[key] = max(0, _PROVIDER_ACTIVE.get(key, 0) - 1)

    async def _post_stream_with_retries_impl(
        self, url: str, headers: dict, body: dict, *, allow_fallback: bool = True
    ) -> tuple[str, dict]:
        last_exc: Exception | None = None
        for attempt in range(self.max_retries + 1):
            try:
                return await self._stream_once(url, headers, body)
            except Exception as exc:  # noqa: BLE001
                last_exc = exc
                if not self._is_retryable(exc) or attempt >= self.max_retries:
                    break
                await asyncio.sleep(min(4.0, 0.5 * (2**attempt)))
        if allow_fallback and last_exc is not None and self._fallback_transports:
            for fb_model, fb_url, fb_headers in self._fallback_transports:
                try:
                    return await self._post_stream_with_retries_impl(
                        fb_url,
                        fb_headers,
                        {**body, "model": fb_model},
                        allow_fallback=False,
                    )
                except Exception as exc:  # noqa: BLE001
                    last_exc = exc
        assert last_exc is not None
        raise last_exc

    async def _stream_once(
        self, url: str, headers: dict, body: dict
    ) -> tuple[str, dict]:
        content = ""
        usage: dict = {}
        async with httpx.AsyncClient(timeout=self.timeout) as client:
            async with client.stream("POST", url, headers=headers, json=body) as resp:
                resp.raise_for_status()
                async for line in resp.aiter_lines():
                    if line.startswith("data: "):
                        data = line[6:].strip()
                    elif line.startswith("{"):
                        data = line.strip()
                    else:
                        continue
                    if data == "[DONE]":
                        break
                    if not data:
                        continue
                    try:
                        chunk = json.loads(data)
                    except json.JSONDecodeError:
                        continue
                    chunk_usage = chunk.get("usage")
                    if isinstance(chunk_usage, dict) and chunk_usage:
                        usage = chunk_usage
                    token = self._stream_token(chunk)
                    if token:
                        content += token
                        if self.on_token:
                            result = self.on_token(token)
                            if inspect.isawaitable(result):
                                await result
        return content, usage

    def _stream_token(self, chunk: dict) -> str:
        """Extract a text delta from a streaming chunk (provider-aware)."""
        if self.type == "anthropic_compatible":
            delta = chunk.get("delta") or {}
            if delta.get("type") == "text_delta":
                return str(delta.get("text", ""))
            return ""
        delta = (chunk.get("choices") or [{}])[0].get("delta", {})
        token = delta.get("content", "")
        if not token:
            token = (chunk.get("message") or {}).get("content", "")
        return str(token)

    async def _post(self, body: dict) -> tuple[dict, bool]:
        return await self._post_with_retries(
            self._url, self._headers, {**body, "stream": False}
        )

    async def _post_stream(self, body: dict) -> tuple[str, dict]:
        body = {**body, "stream": True}
        return await self._post_stream_with_retries(self._url, self._headers, body)

    def _cache_key(self, body: dict) -> str:
        """Hash of the request that identifies a cacheable model call."""
        payload = json.dumps(body, sort_keys=True, default=str)
        digest = hashlib.sha256(payload.encode()).hexdigest()
        return f"{self.provider_key}:{self.model}:{digest}"

    @staticmethod
    def _estimate_message_tokens(messages: list[dict]) -> int:
        """Rough prompt-token estimate (~4 chars per token)."""
        total = 0
        for m in messages:
            content = m.get("content")
            if isinstance(content, list):
                for block in content:
                    total += max(1, len(str(block.get("text", ""))) // 4)
            else:
                total += max(1, len(str(content)) // 4)
        return total

    def _stream_tokens(
        self, messages: list[dict], content: str, stream_usage: dict
    ) -> tuple[int, int]:
        """Token counts for a streamed call.

        Uses provider-reported usage from the streamed chunks when present;
        otherwise falls back to rough estimates so budgets/hooks still work.
        """
        prompt = stream_usage.get("prompt_tokens") or stream_usage.get("input_tokens")
        completion = stream_usage.get("completion_tokens") or stream_usage.get(
            "output_tokens"
        )
        if prompt is None:
            prompt = self._estimate_message_tokens(messages)
        if completion is None:
            completion = len(content) // 4
        return int(prompt or 0), int(completion or 0)

    async def call(
        self,
        messages: list[dict],
        *,
        tools: list[dict] | None = None,
        stream: bool | None = None,
        content_path: str = "",
    ) -> ModelReply:
        """One model call.

        Args:
            messages: Message history.
            tools: Tool schemas to attach (disables streaming).
            stream: Force streaming on/off (defaults to *self.stream* and
                automatically off when *tools* are attached).
            content_path: Dot-separated path for content extraction.

        Returns:
            A :class:`ModelReply` (``cached=True`` when served from cache).
        """
        body = self._body(messages, tools)
        use_stream = stream if stream is not None else (self.stream and not tools)
        t0 = time.monotonic()

        cached = False
        cache_key: str | None = None
        data: dict = {}
        if not use_stream and self._cache is not None:
            cache_key = self._cache_key(body)
            hit = self._cache.get(cache_key)
            if hit is not None:
                data = json.loads(hit) if isinstance(hit, str) else hit
                cached = True

        if use_stream:
            content, stream_usage = await self._post_stream(body)
            msg: dict = {"role": "assistant", "content": content}
            data = {"message": msg}
            prompt, completion = self._stream_tokens(messages, content, stream_usage)
            usage = {"prompt": prompt, "completion": completion}
            if self.on_llm:
                await self.on_llm(
                    self.provider_key, self.model, prompt, completion, _ms(t0)
                )
        else:
            if not cached:
                data, used_fallback = await self._post(body)
                if (
                    self._cache is not None
                    and cache_key is not None
                    and not used_fallback
                ):
                    # Never cache a fallback model's reply under the primary's key,
                    # otherwise a recovered primary keeps serving stale fallback output.
                    self._cache[cache_key] = json.dumps(data, default=str)
            if self.type == "anthropic_compatible":
                msg = _anthropic_to_message(data)
                content = msg.get("content", "")
            else:
                msg = extract_message(data)
                content = extract_content(
                    data, self.type, content_path, msg.get("content", "")
                )
            prompt, completion = extract_usage(data)
            usage = {"prompt": prompt, "completion": completion}
            if self.on_llm:
                await self.on_llm(
                    self.provider_key, self.model, prompt, completion, _ms(t0)
                )
        log.info(
            "llm_call model=%s provider=%s prompt_tokens=%s completion_tokens=%s latency_ms=%s",
            self.model,
            self.provider_key,
            usage.get("prompt"),
            usage.get("completion"),
            f"{_ms(t0):.0f}",
        )
        log.debug("llm_request %s", _truncate(redact(_last_user_message(messages))))
        log.debug("llm_response %s", _truncate(redact(content)))
        if self.on_llm_payload is not None:
            await self.on_llm_payload(
                self.provider_key,
                self.model,
                messages,
                content,
                usage,
                _ms(t0),
                cached,
            )
        self.total_tokens += int(usage.get("prompt", 0)) + int(
            usage.get("completion", 0)
        )
        return ModelReply(
            data=data,
            message=msg,
            content=content,
            usage=usage,
            latency_ms=_ms(t0),
            cached=cached,
        )

    async def step(
        self, messages: list[dict], tools: Mapping[str, Tool] | None
    ) -> Step:
        """One iteration: call the model, execute requested tools, feed back.

        Returns a :class:`Step` whose ``messages`` is the updated history
        (assistant message plus any ``tool`` responses).  History is
        trimmed to *max_context_tokens* / *max_context_messages* before
        the call.
        """
        messages = self.manage_context(messages)
        tool_defs = [tool_to_schema(t) for t in tools.values()] if tools else []
        reply = await self.call(messages, tools=tool_defs or None)
        tool_calls = reply.message.get("tool_calls")

        if tool_defs and not tool_calls and self.parse_text_tool_calls:
            tool_calls, reply.message = normalize_text_tool_calls(
                reply.content, reply.message, seq=len(messages)
            )

        new_messages = list(messages)
        if tool_calls:
            log.info("tool_call count=%s", len(tool_calls))
            if self.on_tool_call is not None:
                for tc in tool_calls:
                    name, raw, _ = _tool_call_parts(tc)
                    try:
                        args = json.loads(raw) if raw else {}
                    except json.JSONDecodeError:
                        args = {}
                    log.info(
                        "tool_call tool=%s args=%s",
                        name,
                        _truncate(json.dumps(redact(args), default=str)),
                    )
                    result = self.on_tool_call(name, args)
                    if inspect.isawaitable(result):
                        await result
            new_messages.append(reply.message)
            results = await execute_tool_calls(
                tool_calls,
                tools or {},
                self.tool_error_mode,
                self.tool_timeout,
                self.tool_retries,
                self._tool_approval,
            )
            for tc, res in zip(tool_calls, results):
                new_messages.append(
                    {"role": "tool", "tool_call_id": tc.get("id", ""), "content": res}
                )
            step = Step(
                messages=new_messages,
                content=reply.content,
                tool_calls=tool_calls,
                reply=reply,
            )
        else:
            step = Step(
                messages=new_messages,
                content=reply.content,
                tool_calls=[],
                reply=reply,
            )
        if self.on_step:
            await self.on_step(step)
        return step

    def manage_context(self, messages: list[dict]) -> list[dict]:
        """Trim *messages* to the configured context limits.

        Applies ``max_context_tokens`` / ``max_context_messages``
        (whichever is set).  The leading ``system`` message is preserved.
        """
        if self.max_context_tokens is None and self.max_context_messages is None:
            return messages
        return trim_messages(
            messages,
            max_tokens=self.max_context_tokens,
            max_messages=self.max_context_messages,
        )

    async def run(self, messages: list[dict], tools: Mapping[str, Tool] | None) -> Step:
        """Loop :meth:`step` until a final answer, *stop_when*, or *max_rounds*.

        Stops early when the cumulative token budget (*max_total_tokens*)
        is exceeded.

        Returns the final :class:`Step` (its ``content`` holds the answer;
        ``messages`` holds the full history).
        """
        step = await self.step(messages, tools)
        for _ in range(1, self.max_rounds):
            if not step.wants_tool:
                break
            if self.stop_when is not None and self.stop_when(step.messages):
                break
            if (
                self.max_total_tokens is not None
                and self.total_tokens >= self.max_total_tokens
            ):
                break
            step = await self.step(step.messages, tools)
        return step

call async

call(messages, *, tools=None, stream=None, content_path='')

One model call.

Parameters:

Name	Type	Description	Default
messages	list[dict]	Message history.	required
tools	list[dict] | None	Tool schemas to attach (disables streaming).	None
stream	bool | None	Force streaming on/off (defaults to self.stream and automatically off when tools are attached).	None
content_path	str	Dot-separated path for content extraction.	''

Returns:

Name	Type	Description
A	ModelReply	class:ModelReply (cached=True when served from cache).

Source code in teff/harness/loop.py

async def call(
    self,
    messages: list[dict],
    *,
    tools: list[dict] | None = None,
    stream: bool | None = None,
    content_path: str = "",
) -> ModelReply:
    """One model call.

    Args:
        messages: Message history.
        tools: Tool schemas to attach (disables streaming).
        stream: Force streaming on/off (defaults to *self.stream* and
            automatically off when *tools* are attached).
        content_path: Dot-separated path for content extraction.

    Returns:
        A :class:`ModelReply` (``cached=True`` when served from cache).
    """
    body = self._body(messages, tools)
    use_stream = stream if stream is not None else (self.stream and not tools)
    t0 = time.monotonic()

    cached = False
    cache_key: str | None = None
    data: dict = {}
    if not use_stream and self._cache is not None:
        cache_key = self._cache_key(body)
        hit = self._cache.get(cache_key)
        if hit is not None:
            data = json.loads(hit) if isinstance(hit, str) else hit
            cached = True

    if use_stream:
        content, stream_usage = await self._post_stream(body)
        msg: dict = {"role": "assistant", "content": content}
        data = {"message": msg}
        prompt, completion = self._stream_tokens(messages, content, stream_usage)
        usage = {"prompt": prompt, "completion": completion}
        if self.on_llm:
            await self.on_llm(
                self.provider_key, self.model, prompt, completion, _ms(t0)
            )
    else:
        if not cached:
            data, used_fallback = await self._post(body)
            if (
                self._cache is not None
                and cache_key is not None
                and not used_fallback
            ):
                # Never cache a fallback model's reply under the primary's key,
                # otherwise a recovered primary keeps serving stale fallback output.
                self._cache[cache_key] = json.dumps(data, default=str)
        if self.type == "anthropic_compatible":
            msg = _anthropic_to_message(data)
            content = msg.get("content", "")
        else:
            msg = extract_message(data)
            content = extract_content(
                data, self.type, content_path, msg.get("content", "")
            )
        prompt, completion = extract_usage(data)
        usage = {"prompt": prompt, "completion": completion}
        if self.on_llm:
            await self.on_llm(
                self.provider_key, self.model, prompt, completion, _ms(t0)
            )
    log.info(
        "llm_call model=%s provider=%s prompt_tokens=%s completion_tokens=%s latency_ms=%s",
        self.model,
        self.provider_key,
        usage.get("prompt"),
        usage.get("completion"),
        f"{_ms(t0):.0f}",
    )
    log.debug("llm_request %s", _truncate(redact(_last_user_message(messages))))
    log.debug("llm_response %s", _truncate(redact(content)))
    if self.on_llm_payload is not None:
        await self.on_llm_payload(
            self.provider_key,
            self.model,
            messages,
            content,
            usage,
            _ms(t0),
            cached,
        )
    self.total_tokens += int(usage.get("prompt", 0)) + int(
        usage.get("completion", 0)
    )
    return ModelReply(
        data=data,
        message=msg,
        content=content,
        usage=usage,
        latency_ms=_ms(t0),
        cached=cached,
    )

from_config classmethod

from_config(cfg, *, default_provider=None, default_model=None, providers=None)

Build a harness from a node config dict.

Recognises the transport keys shared by LLM and ReActAgent plus the loop knobs max_tool_rounds, tool_error_mode, parse_text_tool_calls, tool_timeout, tool_retries, max_retries, fallbacks, max_total_tokens, max_context_tokens and max_context_messages.

providers is an optional {name: Provider} map or :class:~teff.provider.ProviderRegistry (custom providers from the workflow) consulted before the built-in presets.

The model name comes from cfg["model"] or, when absent, default_model (the graph-level default). When neither is set a :class:ConfigError is raised — there is no silent model default.

Source code in teff/harness/loop.py

@classmethod
def from_config(
    cls,
    cfg: dict,
    *,
    default_provider: str | None = None,
    default_model: str | None = None,
    providers: "dict[str, Provider] | ProviderRegistry | None" = None,
) -> "Harness":
    """Build a harness from a node config dict.

    Recognises the transport keys shared by ``LLM`` and
    ``ReActAgent`` plus the loop knobs ``max_tool_rounds``,
    ``tool_error_mode``, ``parse_text_tool_calls``, ``tool_timeout``,
    ``tool_retries``, ``max_retries``, ``fallbacks``,
    ``max_total_tokens``, ``max_context_tokens`` and
    ``max_context_messages``.

    *providers* is an optional ``{name: Provider}`` map or
    :class:`~teff.provider.ProviderRegistry` (custom providers from
    the workflow) consulted before the built-in presets.

    The model name comes from ``cfg["model"]`` or, when absent,
    *default_model* (the graph-level default).  When neither is set a
    :class:`ConfigError` is raised — there is no silent model default.
    """
    model = cfg.get("model") or default_model
    if not model:
        from teff.errors import ConfigError

        raise ConfigError(
            "no model configured: set `model=` on the node or pass "
            "`default_model=` to the graph / `default_model:` in the "
            "workflow"
        )
    return cls(
        model=str(model),
        provider=cfg.get("provider"),
        providers=providers,
        base_url=cfg.get("base_url") or "",
        api_key_env=cfg.get("api_key_env") or "",
        chat_path=cfg.get("chat_path") or "",
        auth_header=cfg.get("auth_header") or "",
        auth_prefix=cfg.get("auth_prefix") or "",
        timeout=_opt_float(cfg.get("timeout")),
        max_rounds=_cfg_int(cfg, "max_tool_rounds", 10),
        parse_text_tool_calls=bool(cfg.get("parse_text_tool_calls", True)),
        tool_error_mode=str(cfg.get("tool_error_mode", "message")),
        tool_timeout=_opt_float(cfg.get("tool_timeout")),
        tool_retries=_cfg_int(cfg, "tool_retries", 0),
        max_retries=_cfg_int(cfg, "max_retries", 2),
        retry_on=tuple(int(x) for x in cfg.get("retry_on") or ())
        or (429, 500, 502, 503, 504),
        fallbacks=cfg.get("fallbacks"),
        tool_approval=cfg.get("tool_approval"),
        max_total_tokens=_opt_int(cfg.get("max_total_tokens")),
        max_context_tokens=_opt_int(cfg.get("max_context_tokens")),
        max_context_messages=_opt_int(cfg.get("max_context_messages")),
        max_parallel=_opt_int(cfg.get("max_parallel")),
        stop_when=cfg.get("stop_when"),
        on_step=cfg.get("on_step"),
        on_llm=cfg.get("on_llm"),
        on_token=cfg.get("on_token"),
        on_tool_call=cfg.get("on_tool_call"),
        temperature=cfg.get("temperature"),
        max_tokens=cfg.get("max_tokens"),
        response_format=cfg.get("response_format"),
        stream=bool(cfg.get("stream", False)),
        cache=cfg.get("cache"),
        default_provider=default_provider,
    )

manage_context

manage_context(messages)

Trim messages to the configured context limits.

Applies max_context_tokens / max_context_messages (whichever is set). The leading system message is preserved.

Source code in teff/harness/loop.py

def manage_context(self, messages: list[dict]) -> list[dict]:
    """Trim *messages* to the configured context limits.

    Applies ``max_context_tokens`` / ``max_context_messages``
    (whichever is set).  The leading ``system`` message is preserved.
    """
    if self.max_context_tokens is None and self.max_context_messages is None:
        return messages
    return trim_messages(
        messages,
        max_tokens=self.max_context_tokens,
        max_messages=self.max_context_messages,
    )

run async

run(messages, tools)

Loop :meth:step until a final answer, stop_when, or max_rounds.

Stops early when the cumulative token budget (max_total_tokens) is exceeded.

Returns the final :class:Step (its content holds the answer; messages holds the full history).

Source code in teff/harness/loop.py

async def run(self, messages: list[dict], tools: Mapping[str, Tool] | None) -> Step:
    """Loop :meth:`step` until a final answer, *stop_when*, or *max_rounds*.

    Stops early when the cumulative token budget (*max_total_tokens*)
    is exceeded.

    Returns the final :class:`Step` (its ``content`` holds the answer;
    ``messages`` holds the full history).
    """
    step = await self.step(messages, tools)
    for _ in range(1, self.max_rounds):
        if not step.wants_tool:
            break
        if self.stop_when is not None and self.stop_when(step.messages):
            break
        if (
            self.max_total_tokens is not None
            and self.total_tokens >= self.max_total_tokens
        ):
            break
        step = await self.step(step.messages, tools)
    return step

step async

step(messages, tools)

One iteration: call the model, execute requested tools, feed back.

Returns a :class:Step whose messages is the updated history (assistant message plus any tool responses). History is trimmed to max_context_tokens / max_context_messages before the call.

Source code in teff/harness/loop.py

async def step(
    self, messages: list[dict], tools: Mapping[str, Tool] | None
) -> Step:
    """One iteration: call the model, execute requested tools, feed back.

    Returns a :class:`Step` whose ``messages`` is the updated history
    (assistant message plus any ``tool`` responses).  History is
    trimmed to *max_context_tokens* / *max_context_messages* before
    the call.
    """
    messages = self.manage_context(messages)
    tool_defs = [tool_to_schema(t) for t in tools.values()] if tools else []
    reply = await self.call(messages, tools=tool_defs or None)
    tool_calls = reply.message.get("tool_calls")

    if tool_defs and not tool_calls and self.parse_text_tool_calls:
        tool_calls, reply.message = normalize_text_tool_calls(
            reply.content, reply.message, seq=len(messages)
        )

    new_messages = list(messages)
    if tool_calls:
        log.info("tool_call count=%s", len(tool_calls))
        if self.on_tool_call is not None:
            for tc in tool_calls:
                name, raw, _ = _tool_call_parts(tc)
                try:
                    args = json.loads(raw) if raw else {}
                except json.JSONDecodeError:
                    args = {}
                log.info(
                    "tool_call tool=%s args=%s",
                    name,
                    _truncate(json.dumps(redact(args), default=str)),
                )
                result = self.on_tool_call(name, args)
                if inspect.isawaitable(result):
                    await result
        new_messages.append(reply.message)
        results = await execute_tool_calls(
            tool_calls,
            tools or {},
            self.tool_error_mode,
            self.tool_timeout,
            self.tool_retries,
            self._tool_approval,
        )
        for tc, res in zip(tool_calls, results):
            new_messages.append(
                {"role": "tool", "tool_call_id": tc.get("id", ""), "content": res}
            )
        step = Step(
            messages=new_messages,
            content=reply.content,
            tool_calls=tool_calls,
            reply=reply,
        )
    else:
        step = Step(
            messages=new_messages,
            content=reply.content,
            tool_calls=[],
            reply=reply,
        )
    if self.on_step:
        await self.on_step(step)
    return step

ModelReply dataclass

A single model call's result.

Source code in teff/harness/loop.py

@dataclass
class ModelReply:
    """A single model call's result."""

    data: dict
    message: dict
    content: str
    usage: dict = field(default_factory=dict)
    latency_ms: float = 0.0
    cached: bool = False

Provider

A named model endpoint: wire protocol + endpoint data.

name is the registry key used by provider= references. type is the protocol discriminator — openai_compatible / anthropic_compatible / ollama — and decides the request body, streaming chunk parsing, and response extraction held by :class:~teff.harness.Harness.

Built-in presets subclass this and set name (and the other fields) once; a custom provider is a plain instance. Fields may be overridden at construction:

Provider(name="my-vllm", type="openai_compatible", base_url="http://vllm:8000/v1")

type is deliberately a distinct concept from name: the name is just a key and never carries protocol meaning.

Methods:

Name	Description
from_mapping	Build from a config dict, keeping only known fields.
to_dict	All provider fields as a plain dict (for YAML serialisation).

Source code in teff/provider/builtin/base.py

class Provider:
    """A named model endpoint: wire protocol + endpoint data.

    ``name`` is the registry key used by ``provider=`` references.
    ``type`` is the protocol discriminator — ``openai_compatible`` /
    ``anthropic_compatible`` / ``ollama`` — and decides the request body,
    streaming chunk parsing, and response extraction held by
    :class:`~teff.harness.Harness`.

    Built-in presets subclass this and set ``name`` (and the other fields)
    once; a custom provider is a plain instance.  Fields may be overridden
    at construction:

        Provider(name="my-vllm", type="openai_compatible", base_url="http://vllm:8000/v1")

    ``type`` is deliberately a distinct concept from ``name``: the name is
    just a key and never carries protocol meaning.
    """

    name: str = ""
    type: str = "openai_compatible"
    base_url: str = ""
    chat_path: str = "/chat/completions"
    api_key_env: str = ""
    auth_header: str = "Authorization"
    auth_prefix: str = "Bearer "
    timeout: float = 120.0

    def __init__(self, **overrides):
        unknown = set(overrides) - set(PROVIDER_FIELDS)
        if unknown:
            raise TypeError(f"unknown Provider field(s): {', '.join(sorted(unknown))}")
        for field in PROVIDER_FIELDS:
            if field in overrides:
                setattr(self, field, overrides[field])

    @classmethod
    def from_mapping(cls, cfg: dict) -> "Provider":
        """Build from a config dict, keeping only known fields."""
        return cls(**{f: cfg[f] for f in PROVIDER_FIELDS if f in cfg})

    def to_dict(self) -> dict:
        """All provider fields as a plain dict (for YAML serialisation)."""
        return {f: getattr(self, f) for f in PROVIDER_FIELDS}

    def __repr__(self) -> str:
        shown = ", ".join(
            f"{f}={getattr(self, f)!r}" for f in PROVIDER_FIELDS if getattr(self, f)
        )
        return f"{type(self).__name__}({shown})"

from_mapping classmethod

from_mapping(cfg)

Build from a config dict, keeping only known fields.

Source code in teff/provider/builtin/base.py

@classmethod
def from_mapping(cls, cfg: dict) -> "Provider":
    """Build from a config dict, keeping only known fields."""
    return cls(**{f: cfg[f] for f in PROVIDER_FIELDS if f in cfg})

to_dict

to_dict()

All provider fields as a plain dict (for YAML serialisation).

Source code in teff/provider/builtin/base.py

def to_dict(self) -> dict:
    """All provider fields as a plain dict (for YAML serialisation)."""
    return {f: getattr(self, f) for f in PROVIDER_FIELDS}

Step dataclass

One iteration of the agent loop (model call + any tool execution).

Attributes:

Name	Type	Description
wants_tool	bool	Whether the step ended requesting more tool execution.

Source code in teff/harness/loop.py

@dataclass
class Step:
    """One iteration of the agent loop (model call + any tool execution)."""

    messages: list[dict]
    content: str
    tool_calls: list[dict]
    reply: ModelReply

    @property
    def wants_tool(self) -> bool:
        """Whether the step ended requesting more tool execution."""
        return bool(self.tool_calls)

wants_tool property

wants_tool

Whether the step ended requesting more tool execution.

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

extract_content

extract_content(data, provider_type, path='', fallback='')

Extract the assistant text from a response.

path is a dot-separated path into data; otherwise the extraction follows the wire protocol provider_type (Anthropic content blocks, Ollama root message).

Source code in teff/harness/formats.py

def extract_content(
    data: dict, provider_type: str, path: str = "", fallback: str = ""
) -> str:
    """Extract the assistant text from a response.

    *path* is a dot-separated path into *data*; otherwise the extraction
    follows the wire protocol *provider_type* (Anthropic content blocks,
    Ollama root ``message``).
    """
    if path:
        parts = path.split(".")
        val: typing.Any = data
        try:
            for p in parts:
                if p.isdigit():
                    val = val[int(p)]
                else:
                    val = val.get(p, "")
        except (AttributeError, IndexError, KeyError, TypeError, ValueError):
            return ""
        return str(val) if val else ""

    if provider_type == "anthropic_compatible":
        for block in data.get("content", []):
            if block.get("type") == "text":
                return block.get("text", "")
        return ""

    if provider_type == "ollama":
        return data.get("message", {}).get("content", "")

    return fallback

extract_message

extract_message(data)

Normalise response formats to {role, content, tool_calls}.

Handles OpenAI (data["choices"][0]["message"]) and Ollama (data["message"] at root).

Source code in teff/harness/formats.py

def extract_message(data: dict) -> dict:
    """Normalise response formats to ``{role, content, tool_calls}``.

    Handles OpenAI (``data["choices"][0]["message"]``) and
    Ollama (``data["message"]`` at root).
    """
    choice = (data.get("choices") or [{}])[0]
    msg = choice.get("message", {})
    if not msg and "message" in data:
        msg = data["message"]
    return msg

extract_usage

extract_usage(data)

Extract (prompt_tokens, completion_tokens) from an LLM response.

Handles both OpenAI-style (data["usage"]) and Ollama-style (data["prompt_eval_count"] / data["eval_count"]) formats.

Source code in teff/harness/formats.py

def extract_usage(data: dict) -> tuple[int, int]:
    """Extract ``(prompt_tokens, completion_tokens)`` from an LLM response.

    Handles both OpenAI-style (``data["usage"]``) and Ollama-style
    (``data["prompt_eval_count"]`` / ``data["eval_count"]``) formats.
    """
    usage = data.get("usage") or {}
    prompt = usage.get("prompt_tokens")
    completion = usage.get("completion_tokens")
    if prompt is None:
        prompt = data.get("prompt_eval_count")
    if completion is None:
        completion = data.get("eval_count")
    if prompt is None:
        prompt = usage.get("input_tokens")
    if completion is None:
        completion = usage.get("output_tokens")
    return int(prompt or 0), int(completion or 0)

normalize_text_tool_calls

normalize_text_tool_calls(content, msg, *, seq=0)

Turn a text-embedded tool call into the structured tool_calls list.

When content parses as a single {name, arguments|parameters} object, returns ([tool_call], msg_with_tool_calls); otherwise returns ([], msg) unchanged. The generated call_id is derived from seq + the tool name so it is unique within a run.

Returns:

Type	Description
list[dict]	A (tool_calls, message) pair. message is msg with
dict	tool_calls attached when a text call was found.

Source code in teff/harness/formats.py

def normalize_text_tool_calls(
    content: str, msg: dict, *, seq: int = 0
) -> tuple[list[dict], dict]:
    """Turn a text-embedded tool call into the structured ``tool_calls`` list.

    When ``content`` parses as a single ``{name, arguments|parameters}``
    object, returns ``([tool_call], msg_with_tool_calls)``; otherwise
    returns ``([], msg)`` unchanged.  The generated ``call_id`` is derived
    from *seq* + the tool name so it is unique within a run.

    Returns:
        A ``(tool_calls, message)`` pair.  *message* is *msg* with
        ``tool_calls`` attached when a text call was found.
    """
    parsed = parse_text_tool_call(content)
    if not parsed:
        return [], msg
    name, args = parsed
    call_id = f"call_{seq}_{name}"
    tool_calls = [
        {
            "id": call_id,
            "type": "function",
            "function": {"name": name, "arguments": json.dumps(args)},
        }
    ]
    return tool_calls, {**msg, "tool_calls": tool_calls}

parse_text_tool_call

parse_text_tool_call(content)

Parse a tool call embedded in plain text content.

Local models sometimes emit {"name": "rag", "parameters": {...}} or {"name": "rag", "arguments": {...}} as text instead of using the structured tool_calls field. Returns (name, args) if found.

Source code in teff/harness/formats.py

def parse_text_tool_call(content: str) -> tuple[str, dict] | None:
    """Parse a tool call embedded in plain text content.

    Local models sometimes emit ``{"name": "rag", "parameters": {...}}``
    or ``{"name": "rag", "arguments": {...}}`` as text instead of using
    the structured ``tool_calls`` field. Returns ``(name, args)`` if found.
    """
    m = re.search(r'"name"\s*:\s*"([^"]+)"', content)
    if not m:
        return None
    name = m.group(1)
    for key in ("parameters", "arguments"):
        idx = content.find(f'"{key}"')
        if idx == -1:
            continue
        brace = content.find("{", content.find(":", idx))
        if brace == -1:
            continue
        obj = extract_json_object(content, brace)
        if obj is None:
            continue
        try:
            args = json.loads(obj)
        except json.JSONDecodeError:
            args = {}
        return name, args
    return name, {}

provider_concurrency

provider_concurrency(provider)

Return the current global concurrency limit for provider (if any).

Returns the active semaphore's capacity (explicit or auto-grown via max_parallel), or None when the provider has no semaphore.

Source code in teff/provider/concurrency.py

def provider_concurrency(provider: str) -> int | None:
    """Return the current global concurrency limit for *provider* (if any).

    Returns the active semaphore's capacity (explicit or auto-grown via
    ``max_parallel``), or ``None`` when the provider has no semaphore.
    """
    return _PROVIDER_LIMITS.get(provider.lower())

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"

resolve_provider

resolve_provider(provider=None, default_provider=None)

Resolve a provider key from an explicit value or a default name.

The explicit provider (node-level) wins; otherwise default_provider (the graph-level default) is used. Model-name auto-detection was removed — a provider must be stated explicitly.

Raises:

Type	Description
ConfigError	When neither a provider nor a default is configured.

Source code in teff/provider/resolve.py

def resolve_provider(
    provider: str | None = None, default_provider: str | None = None
) -> str:
    """Resolve a provider key from an explicit value or a default name.

    The explicit *provider* (node-level) wins; otherwise *default_provider*
    (the graph-level default) is used.  Model-name auto-detection was removed
    — a provider must be stated explicitly.

    Raises:
        ConfigError: When neither a provider nor a default is configured.
    """
    p = provider or default_provider
    if not p:
        raise ConfigError(
            "no provider configured: set `provider=` on the node, pass "
            "`default_provider=` to the graph, or declare a top-level "
            "`default_provider:` in the workflow"
        )
    return p.lower()

resolve_provider_entry

resolve_provider_entry(provider_key, providers=None)

Resolve the effective :class:Provider for provider_key.

When providers is a :class:ProviderRegistry or dict it is authoritative — provider_key must be declared in it. With None (a bare, standalone Harness) a built-in preset is used. Unknown names raise a :class:ConfigError — there is no silent fallback to the OpenAI shape, so typos surface early instead of silently routing to the wrong wire protocol.

Raises:

Type	Description
ConfigError	When provider_key is neither declared in providers nor (with providers=None) a built-in preset name.

Source code in teff/provider/resolve.py

def resolve_provider_entry(
    provider_key: str,
    providers: "dict[str, Provider] | ProviderRegistry | None" = None,
) -> Provider:
    """Resolve the effective :class:`Provider` for *provider_key*.

    When *providers* is a :class:`ProviderRegistry` or dict it is
    authoritative — *provider_key* must be declared in it.  With ``None``
    (a bare, standalone ``Harness``) a built-in preset is used.  Unknown
    names raise a :class:`ConfigError` — there is no silent fallback to the
    OpenAI shape, so typos surface early instead of silently routing to the
    wrong wire protocol.

    Raises:
        ConfigError: When *provider_key* is neither declared in *providers*
            nor (with ``providers=None``) a built-in preset name.
    """
    if isinstance(providers, ProviderRegistry):
        return providers.resolve(provider_key)
    if providers and provider_key in providers:
        return providers[provider_key]
    if providers is None:
        preset = BUILTINS.get(provider_key)
        if preset is not None:
            return preset()
    raise ConfigError(
        f"unknown provider: {provider_key!r} — declare it in the `providers=` "
        f"map / `providers:` block, or name a built-in preset "
        f"({', '.join(BUILTINS)})"
    )

set_provider_concurrency

set_provider_concurrency(provider, limit)

Globally cap concurrent model calls for provider.

Overrides any per-harness max_parallel for that provider. Pass limit <= 0 to remove the cap.

Source code in teff/provider/concurrency.py

def set_provider_concurrency(provider: str, limit: int) -> None:
    """Globally cap concurrent model calls for *provider*.

    Overrides any per-harness ``max_parallel`` for that provider.
    Pass ``limit <= 0`` to remove the cap.
    """
    provider = provider.lower()
    if limit <= 0:
        _EXPLICIT_LIMITS.pop(provider, None)
        _PROVIDER_SEMAPHORES.pop(provider, None)
        _PROVIDER_LIMITS.pop(provider, None)
        _PROVIDER_ACTIVE.pop(provider, None)
    else:
        _EXPLICIT_LIMITS[provider] = limit
        _PROVIDER_SEMAPHORES[provider] = asyncio.Semaphore(limit)
        _PROVIDER_LIMITS[provider] = limit
        _PROVIDER_ACTIVE[provider] = 0

tool_to_schema

tool_to_schema(tool)

Convert a :class:~teff.tool.Tool to an OpenAI-style function schema.

Uses the tool's schema attribute when set (e.g. by MCP tools); otherwise infers parameters from the run/arun signature. Nested type hints (list[dict], dict[str, str], dataclasses, TypedDict) expand to nested JSON Schemas.

Source code in teff/harness/schema.py

def tool_to_schema(tool: Tool) -> dict:
    """Convert a :class:`~teff.tool.Tool` to an OpenAI-style function schema.

    Uses the tool's ``schema`` attribute when set (e.g. by MCP tools);
    otherwise infers ``parameters`` from the ``run``/``arun`` signature.
    Nested type hints (``list[dict]``, ``dict[str, str]``, dataclasses,
    ``TypedDict``) expand to nested JSON Schemas.
    """
    provider_schema = tool.schema
    if isinstance(provider_schema, dict):
        return {
            "type": "function",
            "function": {
                "name": tool.name,
                "description": tool.description or "",
                "parameters": provider_schema,
            },
        }
    run_method = tool.run
    if type(tool).run is Tool.run and type(tool).arun is not Tool.arun:
        run_method = tool.arun
    sig = inspect.signature(run_method)
    try:
        hints = typing.get_type_hints(run_method)
    except Exception:
        hints = {}

    properties: dict = {}
    required: list[str] = []
    for pname, param in sig.parameters.items():
        if pname in ("self", "kwargs", "args"):
            continue
        if pname.startswith("__"):
            # Internal runtime kwargs (e.g. ``__state__``/``__ctx__``) injected
            # by the executor — never exposed to the model as call arguments.
            continue
        prop: dict = _py_type_to_schema(hints.get(pname, str))
        if param.default is not inspect.Parameter.empty:
            if param.default is not None:
                prop["default"] = param.default
        else:
            required.append(pname)
        properties[pname] = prop

    return {
        "type": "function",
        "function": {
            "name": tool.name,
            "description": tool.description or "",
            "parameters": {
                "type": "object",
                "properties": properties,
                "required": required,
            },
        },
    }

trim_messages

trim_messages(messages, max_tokens=None, max_messages=None)

Trim messages down to fit context limits.

The leading system message (if any) is always preserved; older messages are dropped from the front of the conversation until the estimated token count and message count fit the limits.

A limit <= 0 keeps only the system message(s). When the system message alone cannot fit max_tokens (and dropping it is not allowed) a :class:ContextLimitError is raised.

Parameters:

Name	Type	Description	Default
messages	list[dict]	The conversation history.	required
max_tokens	int | None	Maximum estimated tokens to keep.	None
max_messages	int | None	Maximum number of messages to keep.	None

Returns:

Type	Description
list[dict]	A new list of messages, trimmed from the front (system kept).

Raises:

Type	Description
ContextLimitError	When even the system message alone would exceed max_tokens (system is never dropped).

Source code in teff/harness/context.py

def trim_messages(
    messages: list[dict],
    max_tokens: int | None = None,
    max_messages: int | None = None,
) -> list[dict]:
    """Trim *messages* down to fit context limits.

    The leading ``system`` message (if any) is always preserved; older
    messages are dropped from the front of the conversation until the
    estimated token count and message count fit the limits.

    A limit ``<= 0`` keeps only the system message(s).  When the system
    message alone cannot fit *max_tokens* (and dropping it is not allowed)
    a :class:`ContextLimitError` is raised.

    Args:
        messages: The conversation history.
        max_tokens: Maximum estimated tokens to keep.
        max_messages: Maximum number of messages to keep.

    Returns:
        A new list of messages, trimmed from the front (system kept).

    Raises:
        ContextLimitError: When even the system message alone would exceed
            *max_tokens* (system is never dropped).
    """
    if not messages:
        return []

    system: list[dict] = []
    body: list[dict] = []
    for msg in messages:
        if msg.get("role") == "system":
            system.append(msg)
        else:
            body.append(msg)

    if max_messages is not None and max_messages <= 0:
        return system
    if max_tokens is not None and max_tokens <= 0:
        return system

    if max_messages is not None and len(body) > max_messages:
        body = body[-max_messages:]
    if max_tokens is not None and _estimate_tokens(messages) > max_tokens:
        while body and _estimate_tokens(system + body) > max_tokens:
            body.pop(0)
        if not body and _estimate_tokens(system) > max_tokens:
            raise ContextLimitError(
                "conversation system message exceeds max_context_tokens"
            )
    return system + body