Agent

The optional agent layer (pip install "agribound[agent]"). See Agent layer.

agent

The Agribound agent: a tool-use loop with a human confirmation gate.

:func:agent sends a natural-language request to an LLM together with the typed tools of :mod:agribound.agent.tools, runs the tools the model calls, and returns an :class:AgentResult. Its autonomy is deliberately low:

• the model can only propose a run (propose_run); a human approves or denies the exact plan (hash-bound, single-use approval; :mod:agribound.agent.gate);
• the gate enforces a hard execution limit (max_executions, default 1);
• the loop stops right after an execution attempt or a denial, without another model turn, so at most one plan runs per call and the model cannot re-run or re-tune anything; a follow-up run needs a new call by the human;
• dry_run=True removes execute_plan altogether and leaves a plan YAML for agribound delineate --config.

Every model turn and tool call is recorded in a JSON transcript (:mod:agribound.agent.session).

This module imports only the standard library at import time; the tools (pydantic) and the backend SDK are imported when :func:agent runs.

STATUSES module-attribute

STATUSES = ('completed', 'executed', 'execution_failed', 'denied', 'refused', 'max_tokens', 'max_turns', 'error')

AgentResult dataclass

Outcome of one :func:agent session.

Attributes:

Name	Type	Description
status	str	One of :data:STATUSES.
final_text	str	The most recent non-empty text written by the model in a complete turn. Partial text of a refused or max_tokens-truncated response is kept only in the transcript. After an execution attempt or a denial the model gets no further turn, so this is text written before the execution.
report	str	Deterministic summary written by Agribound (not by the model): plans, gate decisions, execution results and the transcript path.
plans	list of dict	Every plan proposed in the session (:meth:Plan.to_dict).
plan_yaml_paths	list of str	YAML files of the plans (agribound delineate --config <file>).
executions	list of dict	Execution attempts with their summaries or errors.
transcript_path	str	The JSON transcript.
session_id, workdir	str	Session identifier and directory.
error	str or None	Error message for "error", "max_tokens" and failed executions.
stop_details	dict or None	Provider detail for "refused".

Source code in agribound/agent/agent.py

@dataclass
class AgentResult:
    """Outcome of one :func:`agent` session.

    Attributes
    ----------
    status : str
        One of :data:`STATUSES`.
    final_text : str
        The most recent non-empty text written by the model in a complete
        turn. Partial text of a refused or ``max_tokens``-truncated response
        is kept only in the transcript. After an execution attempt or a
        denial the model gets no further turn, so this is text written
        *before* the execution.
    report : str
        Deterministic summary written by Agribound (not by the model): plans,
        gate decisions, execution results and the transcript path.
    plans : list of dict
        Every plan proposed in the session (:meth:`Plan.to_dict`).
    plan_yaml_paths : list of str
        YAML files of the plans (``agribound delineate --config <file>``).
    executions : list of dict
        Execution attempts with their summaries or errors.
    transcript_path : str
        The JSON transcript.
    session_id, workdir : str
        Session identifier and directory.
    error : str or None
        Error message for ``"error"``, ``"max_tokens"`` and failed executions.
    stop_details : dict or None
        Provider detail for ``"refused"``.
    """

    status: str
    final_text: str
    report: str
    plans: list[dict[str, Any]] = field(default_factory=list)
    plan_yaml_paths: list[str] = field(default_factory=list)
    executions: list[dict[str, Any]] = field(default_factory=list)
    transcript_path: str = ""
    session_id: str = ""
    workdir: str = ""
    error: str | None = None
    stop_details: dict[str, Any] | None = None

    @property
    def executed(self) -> bool:
        """True if a plan ran successfully."""
        return self.status == "executed"

executed property

executed: bool

True if a plan ran successfully.

agent

agent(request: str, *, study_area: str | None = None, gee_project: str | None = None, workdir: str | Path | None = None, backend: Any = 'anthropic', model: str | None = None, base_url: str | None = None, confirm: Any = None, dry_run: bool = False, max_turns: int = 20, max_executions: int = 1, reference_boundaries: str | None = None, allow_network: bool = True) -> AgentResult

Plan (and, after human approval, run) an Agribound delineation from a request.

Parameters:

Name	Type	Description	Default
request	str	Natural-language request, e.g. "Delineate fields in this AOI for 2024 with a label-free approach".	required
study_area	str or None	Default study area for the tools (vector file, GEE asset ID, "bbox:..." or WKT).	None
gee_project	str or None	Earth Engine project used in plans and live availability checks.	None
workdir	(str, Path or None)	Session directory (default ./agribound_agent/<session_id>). Plans, their YAML files and outputs, the shared cache and the transcript are written here.	None
backend	str or LLMBackend	"anthropic" (default) or an object implementing :class:~agribound.agent.backends.base.LLMBackend.	'anthropic'
model	str or None	Model ID for a named backend (default $AGRIBOUND_AGENT_MODEL or "claude-opus-5").	None
base_url	str or None	Anthropic-compatible endpoint for a named backend (e.g. a local Ollama or vLLM server).	None
confirm	(None, 'prompt', 'deny' or callable)	Confirmation callback of the gate. None: typed yes prompt when standard input is a terminal, otherwise every plan is denied. "prompt" always prompts (e.g. in notebooks), "deny" never approves, and a callable confirm(plan) -> bool decides itself.	None
dry_run	bool	Disable execution: execute_plan is not offered; plans are written as YAML.	False
max_turns	int	Maximum number of model responses (default 20).	20
max_executions	int	Execution limit of the confirmation gate (default 1). The loop stops after the first execution attempt, so at most one plan runs per call whatever the value; 0 keeps execute_plan offered but lets the gate refuse it without asking the reviewer.	1
reference_boundaries	str or None	Default reference layer for resolvability/evaluation tools.	None
allow_network	bool	Allow tools to contact Earth Engine, TESSERA, Source Cooperative or the USGS NAIP Plus ImageServer. With False, the read-only tools skip or refuse their network parts and execute_plan refuses, without asking the reviewer, every plan whose pipeline run needs one of these services. Not affected: the LLM backend, and model-weight downloads from Hugging Face during a run (prefetch them with agribound prefetch).	True

Returns:

Type	Description
AgentResult

Raises:

Type	Description
ValueError	For invalid arguments.
ImportError	If the backend's SDK or pydantic is not installed (pip install "agribound[agent]").

Source code in agribound/agent/agent.py

def agent(
    request: str,
    *,
    study_area: str | None = None,
    gee_project: str | None = None,
    workdir: str | Path | None = None,
    backend: Any = "anthropic",
    model: str | None = None,
    base_url: str | None = None,
    confirm: Any = None,
    dry_run: bool = False,
    max_turns: int = 20,
    max_executions: int = 1,
    reference_boundaries: str | None = None,
    allow_network: bool = True,
) -> AgentResult:
    """Plan (and, after human approval, run) an Agribound delineation from a request.

    Parameters
    ----------
    request : str
        Natural-language request, e.g. ``"Delineate fields in this AOI for
        2024 with a label-free approach"``.
    study_area : str or None
        Default study area for the tools (vector file, GEE asset ID,
        ``"bbox:..."`` or WKT).
    gee_project : str or None
        Earth Engine project used in plans and live availability checks.
    workdir : str, Path or None
        Session directory (default ``./agribound_agent/<session_id>``). Plans,
        their YAML files and outputs, the shared cache and the transcript are
        written here.
    backend : str or LLMBackend
        ``"anthropic"`` (default) or an object implementing
        :class:`~agribound.agent.backends.base.LLMBackend`.
    model : str or None
        Model ID for a named backend (default ``$AGRIBOUND_AGENT_MODEL`` or
        ``"claude-opus-5"``).
    base_url : str or None
        Anthropic-compatible endpoint for a named backend (e.g. a local
        Ollama or vLLM server).
    confirm : None, "prompt", "deny" or callable
        Confirmation callback of the gate. *None*: typed ``yes`` prompt when
        standard input is a terminal, otherwise every plan is denied.
        ``"prompt"`` always prompts (e.g. in notebooks), ``"deny"`` never
        approves, and a callable ``confirm(plan) -> bool`` decides itself.
    dry_run : bool
        Disable execution: ``execute_plan`` is not offered; plans are written
        as YAML.
    max_turns : int
        Maximum number of model responses (default 20).
    max_executions : int
        Execution limit of the confirmation gate (default 1). The loop stops
        after the first execution attempt, so at most one plan runs per call
        whatever the value; ``0`` keeps ``execute_plan`` offered but lets the
        gate refuse it without asking the reviewer.
    reference_boundaries : str or None
        Default reference layer for resolvability/evaluation tools.
    allow_network : bool
        Allow tools to contact Earth Engine, TESSERA, Source Cooperative or
        the USGS NAIP Plus ImageServer. With *False*, the read-only tools skip
        or refuse their network parts and ``execute_plan`` refuses, without
        asking the reviewer, every plan whose pipeline run needs one of these
        services. Not affected: the LLM backend, and model-weight downloads
        from Hugging Face during a run (prefetch them with ``agribound
        prefetch``).

    Returns
    -------
    AgentResult

    Raises
    ------
    ValueError
        For invalid arguments.
    ImportError
        If the backend's SDK or ``pydantic`` is not installed
        (``pip install "agribound[agent]"``).
    """
    if not str(request or "").strip():
        raise ValueError("request must be a non-empty string")
    if int(max_turns) < 1:
        raise ValueError(f"max_turns must be >= 1, got {max_turns}")

    from agribound._repro import new_run_id
    from agribound.agent.backends.base import ToolResult
    from agribound.agent.errors import AgentDependencyError
    from agribound.agent.gate import ConfirmationGate
    from agribound.agent.prompts import SYSTEM_PROMPT, session_preamble
    from agribound.agent.session import AgentSession

    try:
        from agribound.agent.tools import ToolContext, ToolRegistry
    except ModuleNotFoundError as exc:
        if exc.name != "pydantic":
            raise
        raise AgentDependencyError(
            'The agent tools need pydantic: pip install "agribound[agent]"'
        ) from exc

    llm = _make_backend(backend, model, base_url)
    session_id = new_run_id()
    root = Path(workdir) if workdir is not None else Path("agribound_agent") / session_id
    gate = ConfirmationGate(_make_callback(confirm), max_executions=max_executions)
    ctx = ToolContext(
        workdir=root,
        study_area=study_area,
        gee_project=gee_project,
        reference_boundaries=reference_boundaries,
        allow_network=allow_network,
        execution_enabled=not dry_run,
        gate=gate,
    )
    session = AgentSession(
        request,
        workdir=ctx.workdir,
        session_id=session_id,
        backend_info=llm.info(),
        options={
            "study_area": study_area,
            "gee_project": gee_project,
            "reference_boundaries": reference_boundaries,
            "dry_run": dry_run,
            "max_turns": int(max_turns),
            "max_executions": int(max_executions),
            "allow_network": allow_network,
            "workdir": str(ctx.workdir),
        },
    )
    ctx.session = session
    registry = ToolRegistry(ctx)
    tools = registry.definitions()
    session.options["tools"] = [t.name for t in tools]

    messages: list[Any] = [
        llm.user_message(
            session_preamble(
                request,
                study_area=study_area,
                gee_project=gee_project,
                reference_boundaries=reference_boundaries,
                dry_run=dry_run,
                workdir=str(ctx.workdir),
                allow_network=allow_network,
            )
        )
    ]
    status, final_text, error, stop_details = "max_turns", "", None, None
    session.write()

    for _ in range(int(max_turns)):
        t0 = time.perf_counter()
        try:
            turn = llm.complete(system=SYSTEM_PROMPT, tools=tools, messages=messages)
        except Exception as exc:
            logger.error("LLM request failed: %s", exc)
            status, error = "error", f"LLM request failed: {type(exc).__name__}: {exc}"
            if not session.turns:
                error += (
                    " (The first request failed: check the backend's credentials, endpoint and "
                    "network access, e.g. ANTHROPIC_API_KEY or `ant auth login` for the "
                    "Anthropic API, or base_url / --base-url for a local server.)"
                )
            break
        entry = session.record_turn(turn, duration_s=time.perf_counter() - t0)

        # Check the stop reason before using the content: a refused or truncated
        # response is partial and is neither run nor reported as the final text.
        if turn.stop_reason == "refusal":
            status, stop_details = "refused", turn.stop_details
            break
        if turn.stop_reason == "max_tokens":
            status = "max_tokens"
            error = "The model response hit max_tokens before finishing; no tools were run."
            break
        if turn.stop_reason == "model_context_window_exceeded":
            status = "error"
            error = (
                "The conversation exceeded the model's context window; no tools were run. "
                "Start a new, narrower request."
            )
            break
        if turn.text:
            final_text = turn.text
        if turn.stop_reason == "pause_turn":
            # Resend the paused assistant content unchanged so the model continues.
            messages.append(llm.assistant_message(turn))
            session.write()
            continue
        if turn.stop_reason in ("end_turn", "stop_sequence"):
            status = "completed"
            break
        if turn.stop_reason != "tool_use":
            status = "error"
            error = f"Unexpected stop_reason {turn.stop_reason!r}"
            break
        if not turn.tool_calls:
            status, error = "error", "stop_reason 'tool_use' without tool_use blocks"
            break

        messages.append(llm.assistant_message(turn))
        results, stop_status = _run_tool_calls(registry, ctx, session, entry["index"], turn)
        messages.append(
            llm.tool_results_message(
                [ToolResult(tool_call_id=cid, content=c, is_error=e) for cid, c, e in results]
            )
        )
        session.set_gate(gate)
        session.write()
        if stop_status is not None:
            status = stop_status
            if stop_status == "execution_failed":
                error = next(
                    (x.get("error") for x in reversed(ctx.executions) if x.get("error")), None
                )
            break

    session.set_gate(gate)
    report = _render_report(session, status, error)
    session.finish(
        status, final_text=final_text, report=report, error=error, stop_details=stop_details
    )
    path = session.write()
    return AgentResult(
        status=status,
        final_text=final_text,
        report=report,
        plans=list(session.plans),
        plan_yaml_paths=[p["yaml_path"] for p in session.plans if p.get("yaml_path")],
        executions=list(session.executions),
        transcript_path=str(path),
        session_id=session.session_id,
        workdir=str(ctx.workdir),
        error=error,
        stop_details=stop_details,
    )

Tools

tools

Provider-neutral, typed tools of the Agribound agent.

Every tool is a plain function tool(context, validated_input) -> output with a pydantic input model and a pydantic output model. The same :class:ToolRegistry drives the local tool-use loop (:mod:agribound.agent.agent) and the MCP server (:mod:agribound.agent.mcp_server). This module imports neither anthropic nor mcp.

Read-only tools list_sources, list_engines, describe_study_area, check_availability, estimate_resolvability, recommend_configurations, evaluate_against_reference and query_published_ftw. None of them runs the pipeline or modifies user files. query_published_ftw writes the downloaded polygons into the session work directory, and the network-backed checks may fill download caches (published FTW tiles under <workdir>/ftw_cache, the TESSERA tile manifest in the geotessera cache).

Gated tools propose_run freezes a validated configuration into a :class:~agribound.agent.plans.Plan (and writes its YAML); it never runs anything. execute_plan runs :func:agribound.pipeline.delineate for a plan only after the :class:~agribound.agent.gate.ConfirmationGate approved that plan's exact hash. Before the reviewer is asked, :func:preflight_execution refuses plans that cannot run in this session (execution disabled, no executions left, changed inputs, or remote services needed while network access is off).

Network access With ToolContext.allow_network=False no tool contacts Earth Engine, TESSERA, Source Cooperative or the USGS NAIP Plus ImageServer: the read-only tools refuse (or skip) their network parts, and execute_plan refuses every plan whose pipeline run would need one of these services (:func:plan_network_services). Model weights are a separate matter: engines download them from Hugging Face unless they are already cached (agribound prefetch); this flag does not block that.

Expected failures raise :class:~agribound.agent.errors.AgentToolError.

ToolContext dataclass

State shared by the tools of one agent session or MCP server process.

Parameters:

Name	Type	Description	Default
workdir	str or Path	Session directory. Plans, plan outputs, downloaded FTW polygons and the shared cache (<workdir>/cache) are written here.	required
study_area	str or None	Default study area for tools called without one.	None
gee_project	str or None	Default Earth Engine project (plans and live checks).	None
gee_service_account_key	str or None	Service-account key injected into plans (never set by the model).	None
reference_boundaries	str or None	Default reference layer for resolvability and evaluation tools.	None
allow_network	bool	Whether tools may contact Earth Engine, TESSERA, Source Cooperative or the USGS NAIP Plus ImageServer. False disables the live availability checks, published FTW polygons and GEE-asset study areas of the read-only tools, and makes execute_plan refuse plans whose pipeline run needs one of these services (:func:plan_network_services; see the module docstring for model-weight downloads).	True
execution_enabled	bool	Whether execute_plan may run at all (False for dry runs).	False
gate	ConfirmationGate or None	The confirmation gate (required for execution).	None
embedding_cache_dir	str or None	Cache directory for the TESSERA manifest used by live checks.	None

Source code in agribound/agent/tools.py

@dataclass
class ToolContext:
    """State shared by the tools of one agent session or MCP server process.

    Parameters
    ----------
    workdir : str or Path
        Session directory. Plans, plan outputs, downloaded FTW polygons and
        the shared cache (``<workdir>/cache``) are written here.
    study_area : str or None
        Default study area for tools called without one.
    gee_project : str or None
        Default Earth Engine project (plans and live checks).
    gee_service_account_key : str or None
        Service-account key injected into plans (never set by the model).
    reference_boundaries : str or None
        Default reference layer for resolvability and evaluation tools.
    allow_network : bool
        Whether tools may contact Earth Engine, TESSERA, Source Cooperative
        or the USGS NAIP Plus ImageServer. *False* disables the live
        availability checks, published FTW polygons and GEE-asset study areas
        of the read-only tools, and makes ``execute_plan`` refuse plans whose
        pipeline run needs one of these services (:func:`plan_network_services`;
        see the module docstring for model-weight downloads).
    execution_enabled : bool
        Whether ``execute_plan`` may run at all (*False* for dry runs).
    gate : ConfirmationGate or None
        The confirmation gate (required for execution).
    embedding_cache_dir : str or None
        Cache directory for the TESSERA manifest used by live checks.
    """

    workdir: Path
    study_area: str | None = None
    gee_project: str | None = None
    gee_service_account_key: str | None = None
    reference_boundaries: str | None = None
    allow_network: bool = True
    execution_enabled: bool = False
    gate: Any = None
    embedding_cache_dir: str | None = None
    session: Any = None
    plans: dict[str, Any] = field(default_factory=dict)
    executions: list[dict[str, Any]] = field(default_factory=list)

    def __post_init__(self) -> None:
        self.workdir = Path(self.workdir).expanduser().resolve()
        self.workdir.mkdir(parents=True, exist_ok=True)

    @property
    def cache_dir(self) -> Path:
        return self.workdir / "cache"

    def resolve_study_area(self, value: str | None) -> str:
        area = value or self.study_area
        if not area:
            raise AgentToolError(
                "No study area was given and the session has no default study area. Pass "
                "study_area (a vector file, GEE asset ID, 'bbox:minx,miny,maxx,maxy' or WKT)."
            )
        return str(area)

    def require_network(self, what: str) -> None:
        if not self.allow_network:
            raise NetworkDisabledError(
                f"{what} needs network access, which is disabled for this session."
            )

    def get_plan(self, plan_id: str) -> Any:
        plan = self.plans.get(plan_id)
        if plan is None:
            known = sorted(self.plans) or "none"
            raise UnknownPlanError(f"Unknown plan_id {plan_id!r}. Plans in this session: {known}")
        return plan

ToolRegistry

Validating dispatcher over :data:TOOL_SPECS for one :class:ToolContext.

Parameters:

Name	Type	Description	Default
context	ToolContext	Shared session state.	required
include_execute	bool or None	Offer execute_plan; default context.execution_enabled.	None

Source code in agribound/agent/tools.py

class ToolRegistry:
    """Validating dispatcher over :data:`TOOL_SPECS` for one :class:`ToolContext`.

    Parameters
    ----------
    context : ToolContext
        Shared session state.
    include_execute : bool or None
        Offer ``execute_plan``; default ``context.execution_enabled``.
    """

    def __init__(self, context: ToolContext, *, include_execute: bool | None = None) -> None:
        self.context = context
        include = context.execution_enabled if include_execute is None else include_execute
        self.specs: dict[str, ToolSpec] = {
            s.name: s for s in TOOL_SPECS if include or s.name != "execute_plan"
        }

    def definitions(self) -> list[Any]:
        """Tool definitions for a backend, in a stable order."""
        return [spec.definition() for spec in self.specs.values()]

    def call(self, name: str, arguments: Any) -> ToolOutcome:
        """Validate *arguments* and run tool *name*; never raises for tool failures."""
        spec = self.specs.get(name)
        if spec is None:
            return ToolOutcome(
                name=name,
                ok=False,
                error=f"Unknown tool {name!r}. Available: {list(self.specs)}",
                arguments=arguments,
            )
        if not isinstance(arguments, dict):
            return ToolOutcome(
                name=name,
                ok=False,
                error=f"Arguments for {name} must be a JSON object, got {type(arguments).__name__}",
                arguments=arguments,
            )
        try:
            validated = spec.input_model.model_validate(arguments)
        except ValidationError as exc:
            return ToolOutcome(
                name=name, ok=False, error=format_validation_error(name, exc), arguments=arguments
            )
        dumped = validated.model_dump(mode="json")
        try:
            result = spec.func(self.context, validated)
        except AgentToolError as exc:
            return ToolOutcome(
                name=name, ok=False, error=str(exc), arguments=dumped, arguments_valid=True
            )
        except Exception as exc:
            logger.warning("Tool %s raised an unexpected error", name, exc_info=True)
            return ToolOutcome(
                name=name,
                ok=False,
                error=f"Unexpected error in {name}: {type(exc).__name__}: {exc}",
                arguments=dumped,
                arguments_valid=True,
            )
        return ToolOutcome(
            name=name,
            ok=True,
            output=result.model_dump(mode="json"),
            arguments=dumped,
            arguments_valid=True,
        )

definitions

definitions() -> list[Any]

Tool definitions for a backend, in a stable order.

Source code in agribound/agent/tools.py

def definitions(self) -> list[Any]:
    """Tool definitions for a backend, in a stable order."""
    return [spec.definition() for spec in self.specs.values()]

call

call(name: str, arguments: Any) -> ToolOutcome

Validate arguments and run tool name; never raises for tool failures.

Source code in agribound/agent/tools.py

def call(self, name: str, arguments: Any) -> ToolOutcome:
    """Validate *arguments* and run tool *name*; never raises for tool failures."""
    spec = self.specs.get(name)
    if spec is None:
        return ToolOutcome(
            name=name,
            ok=False,
            error=f"Unknown tool {name!r}. Available: {list(self.specs)}",
            arguments=arguments,
        )
    if not isinstance(arguments, dict):
        return ToolOutcome(
            name=name,
            ok=False,
            error=f"Arguments for {name} must be a JSON object, got {type(arguments).__name__}",
            arguments=arguments,
        )
    try:
        validated = spec.input_model.model_validate(arguments)
    except ValidationError as exc:
        return ToolOutcome(
            name=name, ok=False, error=format_validation_error(name, exc), arguments=arguments
        )
    dumped = validated.model_dump(mode="json")
    try:
        result = spec.func(self.context, validated)
    except AgentToolError as exc:
        return ToolOutcome(
            name=name, ok=False, error=str(exc), arguments=dumped, arguments_valid=True
        )
    except Exception as exc:
        logger.warning("Tool %s raised an unexpected error", name, exc_info=True)
        return ToolOutcome(
            name=name,
            ok=False,
            error=f"Unexpected error in {name}: {type(exc).__name__}: {exc}",
            arguments=dumped,
            arguments_valid=True,
        )
    return ToolOutcome(
        name=name,
        ok=True,
        output=result.model_dump(mode="json"),
        arguments=dumped,
        arguments_valid=True,
    )

preflight_execution

preflight_execution(ctx: ToolContext, plan: Any) -> None

Checks made before the reviewer is asked to approve plan.

None of them involves the reviewer, so a plan that cannot run is refused without asking for an approval that could not be used.

Raises:

Type	Description
ExecutionDisabledError	Execution is disabled (dry run) or no gate is configured.
ExecutionLimitError	The gate has no executions left.
PlanChangedError	The plan no longer matches its hash (:func:agribound.agent.gate.check_plan_current).
NetworkDisabledError	Network access is off and the plan needs a remote service (:func:plan_network_services).

Source code in agribound/agent/tools.py

def preflight_execution(ctx: ToolContext, plan: Any) -> None:
    """Checks made before the reviewer is asked to approve *plan*.

    None of them involves the reviewer, so a plan that cannot run is refused
    without asking for an approval that could not be used.

    Raises
    ------
    ExecutionDisabledError
        Execution is disabled (dry run) or no gate is configured.
    ExecutionLimitError
        The gate has no executions left.
    PlanChangedError
        The plan no longer matches its hash
        (:func:`agribound.agent.gate.check_plan_current`).
    NetworkDisabledError
        Network access is off and the plan needs a remote service
        (:func:`plan_network_services`).
    """
    from agribound.agent.gate import check_plan_current

    if not ctx.execution_enabled:
        raise ExecutionDisabledError("Execution is disabled in this session (dry run).")
    if ctx.gate is None:
        raise ExecutionDisabledError("No confirmation gate is configured; execution is disabled.")
    ctx.gate.check_limit()
    check_plan_current(plan)
    _require_plan_network(ctx, plan)

plan_network_services

plan_network_services(config: Any) -> list[str]

Remote data services the pipeline contacts for config.

Judged from the configuration alone: caches are not inspected, so a plan is listed as needing a service even if, for example, its composite is already cached. Model-weight downloads (Hugging Face) are not listed.

Parameters:

Name	Type	Description	Default
config	AgriboundConfig	Validated configuration.	required

Returns:

Type	Description
list of str	One entry per service and stage, e.g. "Earth Engine (composite)"; empty for a local run without the LULC filter and with a file study area.

Source code in agribound/agent/tools.py

def plan_network_services(config: Any) -> list[str]:
    """Remote data services the pipeline contacts for *config*.

    Judged from the configuration alone: caches are not inspected, so a plan
    is listed as needing a service even if, for example, its composite is
    already cached. Model-weight downloads (Hugging Face) are not listed.

    Parameters
    ----------
    config : AgriboundConfig
        Validated configuration.

    Returns
    -------
    list of str
        One entry per service and stage, e.g. ``"Earth Engine (composite)"``;
        empty for a ``local`` run without the LULC filter and with a file
        study area.
    """
    services: list[str] = []
    if str(config.study_area or "").startswith(("projects/", "users/")):
        services.append("Earth Engine (study-area asset)")
    if config.is_gee_source():
        services.append("Earth Engine (composite)")
    if config.source == "google-embedding":
        if config.google_embedding_backend == "gee":
            services.append("Earth Engine (Google embeddings)")
        else:
            services.append("Source Cooperative (Google embeddings)")
    if config.source == "tessera-embedding":
        services.append("TESSERA (embedding tiles)")
    if config.source == "usgs-naip-plus":
        from agribound.config import AgriboundConfig

        default_url = AgriboundConfig.__dataclass_fields__["usgs_service_url"].default
        if config.usgs_service_url == default_url:
            services.append("USGS NAIP Plus ImageServer (composite)")
        else:
            from urllib.parse import urlparse

            host = urlparse(str(config.usgs_service_url)).netloc or str(config.usgs_service_url)
            services.append(f"ImageServer at {host} (usgs_service_url; composite)")
    if config.is_gee_source() and config.export_method == "gcs":
        services.append(f"Google Cloud Storage bucket {config.gcs_bucket!r} (batch export)")
    elif config.is_gee_source() and config.export_method == "gdrive":
        services.append("Google Drive of the Earth Engine account (batch export)")
    if config.lulc_filter:
        services.append(f"Earth Engine (LULC filter, lulc_mode={config.lulc_mode!r})")
    return services

Plans and the confirmation gate

plans

Run plans: frozen, hash-identified Agribound configurations proposed by the agent.

A :class:Plan is created by the propose_run tool. It stores the fully validated :class:~agribound.config.AgriboundConfig as canonical JSON, a fingerprint of the run's inputs, and a SHA-256 plan hash over both. The confirmation gate (:mod:agribound.agent.gate) binds every approval to that hash, and recomputes it immediately before execution, so an approval can never authorise a different configuration, a different study-area geometry, or a modified reference/raster file.

Input fingerprints

• study_area: :func:agribound._cache.aoi_fingerprint (geometry-based for files, bbox: strings and WKT; the asset ID string for GEE assets).
• reference_boundaries and local_tif_path: absolute path, size and modification time (the file contents are not hashed).

Plan dataclass

A proposed Agribound run, identified by the hash of its content.

Instances are immutable; a changed configuration is a new plan with a new plan_id and plan_hash and needs its own approval.

Attributes:

Name	Type	Description
plan_id	str	"plan-" + the first 12 hex characters of plan_hash.
config_json	str	Canonical JSON of AgriboundConfig.to_dict().
inputs_json	str	Canonical JSON of :func:input_fingerprints at proposal time.
plan_hash	str	:func:compute_plan_hash of the two fields above.
created_utc	str	ISO-8601 creation time (UTC).
rationale, limitations, alternatives		The agent's explanation, shown to the reviewer.
warnings	tuple of str	Warnings generated by Agribound (not by the model).
non_default_fields_json	str	Canonical JSON of :func:non_default_fields.
estimated_cost_json	str	Canonical JSON of the size/resource estimate.
yaml_path	str or None	Where the plan's configuration YAML was written.

Source code in agribound/agent/plans.py

@dataclass(frozen=True)
class Plan:
    """A proposed Agribound run, identified by the hash of its content.

    Instances are immutable; a changed configuration is a new plan with a new
    ``plan_id`` and ``plan_hash`` and needs its own approval.

    Attributes
    ----------
    plan_id : str
        ``"plan-"`` + the first 12 hex characters of *plan_hash*.
    config_json : str
        Canonical JSON of ``AgriboundConfig.to_dict()``.
    inputs_json : str
        Canonical JSON of :func:`input_fingerprints` at proposal time.
    plan_hash : str
        :func:`compute_plan_hash` of the two fields above.
    created_utc : str
        ISO-8601 creation time (UTC).
    rationale, limitations, alternatives
        The agent's explanation, shown to the reviewer.
    warnings : tuple of str
        Warnings generated by Agribound (not by the model).
    non_default_fields_json : str
        Canonical JSON of :func:`non_default_fields`.
    estimated_cost_json : str
        Canonical JSON of the size/resource estimate.
    yaml_path : str or None
        Where the plan's configuration YAML was written.
    """

    plan_id: str
    config_json: str
    inputs_json: str
    plan_hash: str
    created_utc: str
    rationale: str = ""
    limitations: tuple[str, ...] = ()
    alternatives: tuple[str, ...] = ()
    warnings: tuple[str, ...] = ()
    non_default_fields_json: str = "{}"
    estimated_cost_json: str = "{}"
    yaml_path: str | None = None

    # -- accessors -------------------------------------------------------------

    @property
    def config(self) -> dict[str, Any]:
        """A fresh copy of the frozen configuration dictionary."""
        return json.loads(self.config_json)

    @property
    def inputs(self) -> dict[str, Any]:
        """A fresh copy of the input fingerprints recorded at proposal time."""
        return json.loads(self.inputs_json)

    @property
    def non_default_fields(self) -> dict[str, Any]:
        """Fields that differ from the package defaults."""
        return json.loads(self.non_default_fields_json)

    @property
    def estimated_cost(self) -> dict[str, Any]:
        """Size and resource estimate."""
        return json.loads(self.estimated_cost_json)

    def to_config(self) -> Any:
        """Rebuild the validated :class:`~agribound.config.AgriboundConfig`."""
        from agribound.config import AgriboundConfig

        return AgriboundConfig.from_dict(self.config)

    def current_hash(self) -> str:
        """Recompute the plan hash from the stored configuration and *current* inputs.

        Differs from :attr:`plan_hash` when the stored configuration was
        altered or when an input file or the study-area geometry changed.
        """
        current_inputs = canonical_json(input_fingerprints(self.to_config()))
        return compute_plan_hash(self.config_json, current_inputs)

    def stored_hash_is_consistent(self) -> bool:
        """True if :attr:`plan_hash` matches the stored JSON (no input re-check)."""
        return compute_plan_hash(self.config_json, self.inputs_json) == self.plan_hash

    def config_yaml(self) -> str:
        """The configuration as YAML, in :class:`AgriboundConfig` field order."""
        import yaml

        from agribound.config import AgriboundConfig

        data = self.config
        order = AgriboundConfig.field_names()
        ordered = {k: data[k] for k in order if k in data}
        return yaml.safe_dump(ordered, sort_keys=False, default_flow_style=False)

    def to_dict(self) -> dict[str, Any]:
        """JSON-serialisable view (used in tool results and the session transcript)."""
        return {
            "plan_id": self.plan_id,
            "plan_hash": self.plan_hash,
            "created_utc": self.created_utc,
            "config": self.config,
            "inputs": self.inputs,
            "rationale": self.rationale,
            "limitations": list(self.limitations),
            "alternatives": list(self.alternatives),
            "warnings": list(self.warnings),
            "non_default_fields": self.non_default_fields,
            "estimated_cost": self.estimated_cost,
            "yaml_path": self.yaml_path,
        }

    def render(self) -> str:
        """Human-readable description shown by the confirmation gate.

        The sections generated by Agribound (warnings, fields that differ from
        the defaults, estimate) come first. The text written by the agent
        (rationale, limitations, alternatives) follows, with **every** line
        prefixed by ``"  | "``, so agent text cannot pass itself off as an
        Agribound section. The full configuration YAML comes last. Control
        characters in any value (terminal escape sequences, line breaks in
        a single-line field, bidirectional overrides) are shown escaped
        (:func:`display_safe`), so no value can move the cursor, erase or
        reorder the screen, or add lines of its own.
        """
        cfg = self.config
        safe = display_safe
        lines = [
            f"Plan {self.plan_id} (sha256 {self.plan_hash})",
            f"  source={safe(cfg.get('source'))}  engine={safe(cfg.get('engine'))}  "
            f"year={safe(cfg.get('year'))}",
            f"  study_area={safe(cfg.get('study_area'))}",
            f"  output_path={safe(cfg.get('output_path'))}",
            "",
            "Warnings (from Agribound):",
        ]
        lines += [f"  - {safe(item)}" for item in self.warnings] or ["  (none)"]
        changes = self.non_default_fields
        lines += ["", "Fields that differ from the AgriboundConfig defaults:"]
        if changes:
            for name, change in changes.items():
                lines.append(
                    f"  {safe(name)}: {safe(repr(change['default']))} -> "
                    f"{safe(repr(change['value']))}"
                )
        else:
            lines.append("  (none)")
        cost = self.estimated_cost
        if cost:
            lines += ["", "Estimate:"]
            lines += [f"  {safe(key)}: {safe(value)}" for key, value in cost.items()]
        if self.rationale or self.limitations or self.alternatives:
            lines += [
                "",
                f"Text written by the agent (not checked by Agribound; lines marked "
                f"{_AGENT_MARK.strip()!r}):",
            ]
            if self.rationale:
                lines.append("  Rationale:")
                lines += _marked(self.rationale)
            if self.limitations:
                lines.append("  Limitations:")
                for item in self.limitations:
                    lines += _marked(item, first="- ", cont="  ")
            if self.alternatives:
                lines.append("  Alternatives:")
                for item in self.alternatives:
                    lines += _marked(item, first="- ", cont="  ")
        lines += ["", "Full configuration (YAML):"]
        lines += [safe(line) for line in self.config_yaml().rstrip().split("\n")]
        return "\n".join(lines)

config property

config: dict[str, Any]

A fresh copy of the frozen configuration dictionary.

inputs property

inputs: dict[str, Any]

A fresh copy of the input fingerprints recorded at proposal time.

non_default_fields property

non_default_fields: dict[str, Any]

Fields that differ from the package defaults.

estimated_cost property

estimated_cost: dict[str, Any]

Size and resource estimate.

to_config

to_config() -> Any

Rebuild the validated :class:~agribound.config.AgriboundConfig.

Source code in agribound/agent/plans.py

def to_config(self) -> Any:
    """Rebuild the validated :class:`~agribound.config.AgriboundConfig`."""
    from agribound.config import AgriboundConfig

    return AgriboundConfig.from_dict(self.config)

current_hash

current_hash() -> str

Recompute the plan hash from the stored configuration and current inputs.

Differs from :attr:plan_hash when the stored configuration was altered or when an input file or the study-area geometry changed.

Source code in agribound/agent/plans.py

def current_hash(self) -> str:
    """Recompute the plan hash from the stored configuration and *current* inputs.

    Differs from :attr:`plan_hash` when the stored configuration was
    altered or when an input file or the study-area geometry changed.
    """
    current_inputs = canonical_json(input_fingerprints(self.to_config()))
    return compute_plan_hash(self.config_json, current_inputs)

stored_hash_is_consistent

stored_hash_is_consistent() -> bool

True if :attr:plan_hash matches the stored JSON (no input re-check).

Source code in agribound/agent/plans.py

def stored_hash_is_consistent(self) -> bool:
    """True if :attr:`plan_hash` matches the stored JSON (no input re-check)."""
    return compute_plan_hash(self.config_json, self.inputs_json) == self.plan_hash

config_yaml

config_yaml() -> str

The configuration as YAML, in :class:AgriboundConfig field order.

Source code in agribound/agent/plans.py

def config_yaml(self) -> str:
    """The configuration as YAML, in :class:`AgriboundConfig` field order."""
    import yaml

    from agribound.config import AgriboundConfig

    data = self.config
    order = AgriboundConfig.field_names()
    ordered = {k: data[k] for k in order if k in data}
    return yaml.safe_dump(ordered, sort_keys=False, default_flow_style=False)

to_dict

to_dict() -> dict[str, Any]

JSON-serialisable view (used in tool results and the session transcript).

Source code in agribound/agent/plans.py

def to_dict(self) -> dict[str, Any]:
    """JSON-serialisable view (used in tool results and the session transcript)."""
    return {
        "plan_id": self.plan_id,
        "plan_hash": self.plan_hash,
        "created_utc": self.created_utc,
        "config": self.config,
        "inputs": self.inputs,
        "rationale": self.rationale,
        "limitations": list(self.limitations),
        "alternatives": list(self.alternatives),
        "warnings": list(self.warnings),
        "non_default_fields": self.non_default_fields,
        "estimated_cost": self.estimated_cost,
        "yaml_path": self.yaml_path,
    }

render

render() -> str

Human-readable description shown by the confirmation gate.

The sections generated by Agribound (warnings, fields that differ from the defaults, estimate) come first. The text written by the agent (rationale, limitations, alternatives) follows, with every line prefixed by " | ", so agent text cannot pass itself off as an Agribound section. The full configuration YAML comes last. Control characters in any value (terminal escape sequences, line breaks in a single-line field, bidirectional overrides) are shown escaped (:func:display_safe), so no value can move the cursor, erase or reorder the screen, or add lines of its own.

Source code in agribound/agent/plans.py

def render(self) -> str:
    """Human-readable description shown by the confirmation gate.

    The sections generated by Agribound (warnings, fields that differ from
    the defaults, estimate) come first. The text written by the agent
    (rationale, limitations, alternatives) follows, with **every** line
    prefixed by ``"  | "``, so agent text cannot pass itself off as an
    Agribound section. The full configuration YAML comes last. Control
    characters in any value (terminal escape sequences, line breaks in
    a single-line field, bidirectional overrides) are shown escaped
    (:func:`display_safe`), so no value can move the cursor, erase or
    reorder the screen, or add lines of its own.
    """
    cfg = self.config
    safe = display_safe
    lines = [
        f"Plan {self.plan_id} (sha256 {self.plan_hash})",
        f"  source={safe(cfg.get('source'))}  engine={safe(cfg.get('engine'))}  "
        f"year={safe(cfg.get('year'))}",
        f"  study_area={safe(cfg.get('study_area'))}",
        f"  output_path={safe(cfg.get('output_path'))}",
        "",
        "Warnings (from Agribound):",
    ]
    lines += [f"  - {safe(item)}" for item in self.warnings] or ["  (none)"]
    changes = self.non_default_fields
    lines += ["", "Fields that differ from the AgriboundConfig defaults:"]
    if changes:
        for name, change in changes.items():
            lines.append(
                f"  {safe(name)}: {safe(repr(change['default']))} -> "
                f"{safe(repr(change['value']))}"
            )
    else:
        lines.append("  (none)")
    cost = self.estimated_cost
    if cost:
        lines += ["", "Estimate:"]
        lines += [f"  {safe(key)}: {safe(value)}" for key, value in cost.items()]
    if self.rationale or self.limitations or self.alternatives:
        lines += [
            "",
            f"Text written by the agent (not checked by Agribound; lines marked "
            f"{_AGENT_MARK.strip()!r}):",
        ]
        if self.rationale:
            lines.append("  Rationale:")
            lines += _marked(self.rationale)
        if self.limitations:
            lines.append("  Limitations:")
            for item in self.limitations:
                lines += _marked(item, first="- ", cont="  ")
        if self.alternatives:
            lines.append("  Alternatives:")
            for item in self.alternatives:
                lines += _marked(item, first="- ", cont="  ")
    lines += ["", "Full configuration (YAML):"]
    lines += [safe(line) for line in self.config_yaml().rstrip().split("\n")]
    return "\n".join(lines)

gate

Human confirmation gate for agent-proposed runs.

No plan is executed unless a :class:ConfirmationGate holds an unused approval bound to that plan's hash:

• An approval is created either by the gate's callback (called from :meth:ConfirmationGate.request_approval) or explicitly with :meth:ConfirmationGate.record_approval (used by the MCP server after an elicitation, or by Python code that approves a plan it has inspected).
• :meth:ConfirmationGate.authorize_execution recomputes the plan hash from the stored configuration and the current input files, requires a matching unused approval, marks it used (approvals are single-use) and counts the execution against max_executions (default 1 per session). When the hash check fails, the unused approvals of that hash are revoked, and :meth:ConfirmationGate.request_approval always asks the callback anew, so an approval that could not be used never authorises a later run.
• :func:check_plan_current performs the same hash check on its own; the execute_plan tools call it before asking the reviewer, so nobody is asked to approve a plan whose inputs already changed.

Callbacks

A callback receives the :class:~agribound.agent.plans.Plan and returns True to approve. Ready-made callbacks:

• :func:prompt_confirm -- prints the full resolved configuration and requires the reviewer to type yes (anything else denies).
• :func:deny_all -- denies every plan (the default for non-interactive sessions).

:func:default_callback picks :func:prompt_confirm when standard input is an interactive terminal and :func:deny_all otherwise.

ConfirmationGate

Single-use, hash-bound approvals and a hard execution limit.

Parameters:

Name	Type	Description	Default
callback	callable or None	callback(plan) -> bool consulted by every :meth:request_approval call. None uses :func:default_callback.	None
max_executions	int	Maximum number of plans this gate lets run (default 1).	1
approver	str or None	Name recorded with callback approvals (default: the OS user name).	None

Examples:

>>> gate = ConfirmationGate(callback=lambda plan: plan.config["year"] == 2024)

Source code in agribound/agent/gate.py

class ConfirmationGate:
    """Single-use, hash-bound approvals and a hard execution limit.

    Parameters
    ----------
    callback : callable or None
        ``callback(plan) -> bool`` consulted by every :meth:`request_approval`
        call. *None* uses :func:`default_callback`.
    max_executions : int
        Maximum number of plans this gate lets run (default 1).
    approver : str or None
        Name recorded with callback approvals (default: the OS user name).

    Examples
    --------
    >>> gate = ConfirmationGate(callback=lambda plan: plan.config["year"] == 2024)
    """

    def __init__(
        self,
        callback: ConfirmCallback | None = None,
        *,
        max_executions: int = 1,
        approver: str | None = None,
    ) -> None:
        if int(max_executions) < 0:
            raise ValueError(f"max_executions must be >= 0, got {max_executions}")
        self.callback: ConfirmCallback = callback if callback is not None else default_callback()
        self.max_executions = int(max_executions)
        self.approver = approver
        self.approvals: list[Approval] = []
        self.denials: list[Denial] = []
        self.executions = 0
        # Serialises bookkeeping; MCP runs tool calls in worker threads.
        self._lock = threading.RLock()

    # -- queries ---------------------------------------------------------------

    @property
    def remaining_executions(self) -> int:
        return max(0, self.max_executions - self.executions)

    def _unused_approval(self, plan: Any) -> Approval | None:
        for approval in self.approvals:
            if (
                approval.usable
                and approval.plan_id == plan.plan_id
                and approval.plan_hash == plan.plan_hash
            ):
                return approval
        return None

    def _revoke(self, plan: Any, reason: str) -> int:
        """Revoke every usable approval of *plan*'s hash; return how many (lock held)."""
        n = 0
        for approval in self.approvals:
            if approval.usable and approval.plan_hash == plan.plan_hash:
                approval.revoked_utc = _utc_now()
                approval.revoked_reason = reason
                n += 1
        if n:
            logger.info("Revoked %d unused approval(s) of plan %s: %s", n, plan.plan_id, reason)
        return n

    def check_limit(self) -> None:
        """Raise :class:`ExecutionLimitError` if no executions remain."""
        if self.executions >= self.max_executions:
            raise ExecutionLimitError(
                f"This session already ran {self.executions} plan(s) "
                f"(max_executions={self.max_executions}). Any further run needs a new request "
                "from the human user."
            )

    # -- approvals -------------------------------------------------------------

    def record_approval(self, plan: Any, *, approver: str, method: str) -> Approval:
        """Record an approval obtained outside the callback (e.g. MCP elicitation).

        Parameters
        ----------
        plan : Plan
            The approved plan; the approval is bound to ``plan.plan_hash``.
        approver : str
            Who approved (free text, recorded in the transcript).
        method : str
            How the approval was obtained (recorded in the transcript).

        Returns
        -------
        Approval
        """
        approval = Approval(
            plan_id=plan.plan_id,
            plan_hash=plan.plan_hash,
            approver=str(approver),
            method=str(method),
            approved_utc=_utc_now(),
        )
        with self._lock:
            self.approvals.append(approval)
        logger.info("Plan %s approved by %s (%s)", plan.plan_id, approver, method)
        return approval

    def record_denial(self, plan: Any, *, method: str, reason: str) -> Denial:
        """Record a denial (for the transcript)."""
        denial = Denial(
            plan_id=plan.plan_id,
            plan_hash=plan.plan_hash,
            method=str(method),
            reason=str(reason),
            denied_utc=_utc_now(),
        )
        with self._lock:
            self.denials.append(denial)
        logger.info("Plan %s denied (%s): %s", plan.plan_id, method, reason)
        return denial

    def request_approval(self, plan: Any) -> Approval:
        """Ask the callback to approve *plan* and return the new approval.

        The callback is asked on every call. An earlier approval of the same
        plan hash that was never used (for example because
        :meth:`authorize_execution` refused it) is revoked first, so it can
        never authorise a later run without a new answer from the reviewer.

        Raises
        ------
        ExecutionLimitError
            If no executions remain (the reviewer is not asked).
        ExecutionDeniedError
            If the callback returns anything but *True* or raises.
        """
        with self._lock:
            self.check_limit()
            self._revoke(plan, reason="superseded by a new approval request")
        method = _callback_method(self.callback)
        try:
            decision = self.callback(plan)
        except Exception as exc:
            self.record_denial(plan, method=method, reason=f"callback raised {exc!r}")
            raise ExecutionDeniedError(
                f"Plan {plan.plan_id} was not approved: the confirmation callback failed ({exc})."
            ) from exc
        if decision is True:
            return self.record_approval(
                plan, approver=self.approver or _current_user(), method=method
            )
        self.record_denial(plan, method=method, reason="not approved by the reviewer")
        raise ExecutionDeniedError(
            f"Plan {plan.plan_id} was not approved ({method}). Do not modify the plan to obtain "
            "approval; report the plan and its limitations to the user and stop."
        )

    def authorize_execution(self, plan: Any) -> Approval:
        """Consume the approval for *plan* immediately before it runs.

        Recomputes the plan hash from the stored configuration and the
        current input fingerprints, then marks the matching approval used and
        counts the execution.

        Raises
        ------
        PlanChangedError
            If the recomputed hash differs from ``plan.plan_hash`` or the inputs
            can no longer be fingerprinted (:func:`check_plan_current`). The
            unused approvals of that hash are revoked: the reviewer approved
            what they were shown, which is no longer what would run.
        ExecutionLimitError
            If no executions remain.
        ExecutionNotApprovedError
            If there is no unused approval bound to the plan hash.
        """
        try:
            check_plan_current(plan)
        except PlanChangedError as exc:
            with self._lock:
                self._revoke(plan, reason=f"plan changed before it ran: {exc}")
            raise
        with self._lock:  # check, consume and count atomically
            self.check_limit()
            approval = self._unused_approval(plan)
            if approval is None:
                raise ExecutionNotApprovedError(
                    f"Plan {plan.plan_id} (sha256 {plan.plan_hash[:12]}...) has no unused approval."
                )
            approval.used_utc = _utc_now()
            self.executions += 1
        return approval

    def to_dict(self) -> dict[str, Any]:
        """Transcript view of the gate's state."""
        return {
            "callback": _callback_method(self.callback),
            "max_executions": self.max_executions,
            "executions": self.executions,
            "approvals": [a.to_dict() for a in self.approvals],
            "denials": [d.to_dict() for d in self.denials],
        }

check_limit

check_limit() -> None

Raise :class:ExecutionLimitError if no executions remain.

Source code in agribound/agent/gate.py

def check_limit(self) -> None:
    """Raise :class:`ExecutionLimitError` if no executions remain."""
    if self.executions >= self.max_executions:
        raise ExecutionLimitError(
            f"This session already ran {self.executions} plan(s) "
            f"(max_executions={self.max_executions}). Any further run needs a new request "
            "from the human user."
        )

record_approval

record_approval(plan: Any, *, approver: str, method: str) -> Approval

Record an approval obtained outside the callback (e.g. MCP elicitation).

Parameters:

Name	Type	Description	Default
plan	Plan	The approved plan; the approval is bound to plan.plan_hash.	required
approver	str	Who approved (free text, recorded in the transcript).	required
method	str	How the approval was obtained (recorded in the transcript).	required

Returns:

Type	Description
Approval

Source code in agribound/agent/gate.py

def record_approval(self, plan: Any, *, approver: str, method: str) -> Approval:
    """Record an approval obtained outside the callback (e.g. MCP elicitation).

    Parameters
    ----------
    plan : Plan
        The approved plan; the approval is bound to ``plan.plan_hash``.
    approver : str
        Who approved (free text, recorded in the transcript).
    method : str
        How the approval was obtained (recorded in the transcript).

    Returns
    -------
    Approval
    """
    approval = Approval(
        plan_id=plan.plan_id,
        plan_hash=plan.plan_hash,
        approver=str(approver),
        method=str(method),
        approved_utc=_utc_now(),
    )
    with self._lock:
        self.approvals.append(approval)
    logger.info("Plan %s approved by %s (%s)", plan.plan_id, approver, method)
    return approval

record_denial

record_denial(plan: Any, *, method: str, reason: str) -> Denial

Record a denial (for the transcript).

Source code in agribound/agent/gate.py

def record_denial(self, plan: Any, *, method: str, reason: str) -> Denial:
    """Record a denial (for the transcript)."""
    denial = Denial(
        plan_id=plan.plan_id,
        plan_hash=plan.plan_hash,
        method=str(method),
        reason=str(reason),
        denied_utc=_utc_now(),
    )
    with self._lock:
        self.denials.append(denial)
    logger.info("Plan %s denied (%s): %s", plan.plan_id, method, reason)
    return denial

request_approval

request_approval(plan: Any) -> Approval

Ask the callback to approve plan and return the new approval.

The callback is asked on every call. An earlier approval of the same plan hash that was never used (for example because :meth:authorize_execution refused it) is revoked first, so it can never authorise a later run without a new answer from the reviewer.

Raises:

Type	Description
ExecutionLimitError	If no executions remain (the reviewer is not asked).
ExecutionDeniedError	If the callback returns anything but True or raises.

Source code in agribound/agent/gate.py

def request_approval(self, plan: Any) -> Approval:
    """Ask the callback to approve *plan* and return the new approval.

    The callback is asked on every call. An earlier approval of the same
    plan hash that was never used (for example because
    :meth:`authorize_execution` refused it) is revoked first, so it can
    never authorise a later run without a new answer from the reviewer.

    Raises
    ------
    ExecutionLimitError
        If no executions remain (the reviewer is not asked).
    ExecutionDeniedError
        If the callback returns anything but *True* or raises.
    """
    with self._lock:
        self.check_limit()
        self._revoke(plan, reason="superseded by a new approval request")
    method = _callback_method(self.callback)
    try:
        decision = self.callback(plan)
    except Exception as exc:
        self.record_denial(plan, method=method, reason=f"callback raised {exc!r}")
        raise ExecutionDeniedError(
            f"Plan {plan.plan_id} was not approved: the confirmation callback failed ({exc})."
        ) from exc
    if decision is True:
        return self.record_approval(
            plan, approver=self.approver or _current_user(), method=method
        )
    self.record_denial(plan, method=method, reason="not approved by the reviewer")
    raise ExecutionDeniedError(
        f"Plan {plan.plan_id} was not approved ({method}). Do not modify the plan to obtain "
        "approval; report the plan and its limitations to the user and stop."
    )

authorize_execution

authorize_execution(plan: Any) -> Approval

Consume the approval for plan immediately before it runs.

Recomputes the plan hash from the stored configuration and the current input fingerprints, then marks the matching approval used and counts the execution.

Raises:

Type	Description
PlanChangedError	If the recomputed hash differs from plan.plan_hash or the inputs can no longer be fingerprinted (:func:check_plan_current). The unused approvals of that hash are revoked: the reviewer approved what they were shown, which is no longer what would run.
ExecutionLimitError	If no executions remain.
ExecutionNotApprovedError	If there is no unused approval bound to the plan hash.

Source code in agribound/agent/gate.py

def authorize_execution(self, plan: Any) -> Approval:
    """Consume the approval for *plan* immediately before it runs.

    Recomputes the plan hash from the stored configuration and the
    current input fingerprints, then marks the matching approval used and
    counts the execution.

    Raises
    ------
    PlanChangedError
        If the recomputed hash differs from ``plan.plan_hash`` or the inputs
        can no longer be fingerprinted (:func:`check_plan_current`). The
        unused approvals of that hash are revoked: the reviewer approved
        what they were shown, which is no longer what would run.
    ExecutionLimitError
        If no executions remain.
    ExecutionNotApprovedError
        If there is no unused approval bound to the plan hash.
    """
    try:
        check_plan_current(plan)
    except PlanChangedError as exc:
        with self._lock:
            self._revoke(plan, reason=f"plan changed before it ran: {exc}")
        raise
    with self._lock:  # check, consume and count atomically
        self.check_limit()
        approval = self._unused_approval(plan)
        if approval is None:
            raise ExecutionNotApprovedError(
                f"Plan {plan.plan_id} (sha256 {plan.plan_hash[:12]}...) has no unused approval."
            )
        approval.used_utc = _utc_now()
        self.executions += 1
    return approval

to_dict

to_dict() -> dict[str, Any]

Transcript view of the gate's state.

Source code in agribound/agent/gate.py

def to_dict(self) -> dict[str, Any]:
    """Transcript view of the gate's state."""
    return {
        "callback": _callback_method(self.callback),
        "max_executions": self.max_executions,
        "executions": self.executions,
        "approvals": [a.to_dict() for a in self.approvals],
        "denials": [d.to_dict() for d in self.denials],
    }

prompt_confirm

prompt_confirm(plan: Any, *, input_fn: Callable[[str], str] | None = None, out: TextIO | None = None) -> bool

Show the full plan and approve only if the reviewer types yes.

Parameters:

Name	Type	Description	Default
plan	Plan	Plan to review.	required
input_fn	callable or None	Reads the answer; it is called with an empty prompt string because the question is printed to out (default: the built-in :func:input).	None
out	text stream or None	Where the plan and the question are printed (default sys.stderr, so standard output stays free for results and a redirected standard output does not hide the question).	None

Returns:

Type	Description
bool	True only for the exact answer yes (case-insensitive, surrounding whitespace ignored).

Source code in agribound/agent/gate.py

def prompt_confirm(
    plan: Any,
    *,
    input_fn: Callable[[str], str] | None = None,
    out: TextIO | None = None,
) -> bool:
    """Show the full plan and approve only if the reviewer types ``yes``.

    Parameters
    ----------
    plan : Plan
        Plan to review.
    input_fn : callable or None
        Reads the answer; it is called with an empty prompt string because
        the question is printed to *out* (default: the built-in
        :func:`input`).
    out : text stream or None
        Where the plan and the question are printed (default ``sys.stderr``,
        so standard output stays free for results and a redirected standard
        output does not hide the question).

    Returns
    -------
    bool
        *True* only for the exact answer ``yes`` (case-insensitive, surrounding
        whitespace ignored).
    """
    stream = out if out is not None else sys.stderr
    print("", file=stream)
    print("=" * 72, file=stream)
    print("The agent asks to run the following plan.", file=stream)
    print("=" * 72, file=stream)
    print(plan.render(), file=stream)
    print("=" * 72, file=stream)
    print("Type 'yes' to run this plan (anything else cancels): ", end="", file=stream)
    stream.flush()
    reader = input_fn if input_fn is not None else input
    try:
        answer = reader("")
    except (EOFError, KeyboardInterrupt):
        return False
    return str(answer).strip().lower() == "yes"

deny_all

deny_all(plan: Any) -> bool

Deny every plan (non-interactive default).

Source code in agribound/agent/gate.py

def deny_all(plan: Any) -> bool:
    """Deny every plan (non-interactive default)."""
    return False

check_plan_current

check_plan_current(plan: Any) -> None

Raise :class:PlanChangedError unless plan still matches its hash.

Recomputes the hash from the stored configuration and the current input fingerprints (:meth:Plan.current_hash). An input that can no longer be fingerprinted (for example a deleted study-area or reference file) also raises :class:PlanChangedError.

Source code in agribound/agent/gate.py

def check_plan_current(plan: Any) -> None:
    """Raise :class:`PlanChangedError` unless *plan* still matches its hash.

    Recomputes the hash from the stored configuration and the *current*
    input fingerprints (:meth:`Plan.current_hash`). An input that can no
    longer be fingerprinted (for example a deleted study-area or reference
    file) also raises :class:`PlanChangedError`.
    """
    try:
        current = plan.current_hash()
    except (OSError, ValueError, TypeError) as exc:
        raise PlanChangedError(
            f"Plan {plan.plan_id}: its inputs could not be fingerprinted again "
            f"({type(exc).__name__}: {exc}). Propose a new plan; it needs a new approval."
        ) from exc
    if current != plan.plan_hash:
        raise PlanChangedError(
            f"Plan {plan.plan_id} no longer matches its hash (the configuration, the "
            "study-area geometry or an input file changed since it was proposed). "
            "Propose a new plan; it needs a new approval."
        )

Errors

errors

Exceptions of the agent layer.

Every expected failure of an agent tool (invalid arguments, a missing file, no network permission, a plan that was not approved) is an :class:AgentToolError. The local tool-use loop returns its message to the model as an is_error tool result, and the MCP server maps it to mcp.server.mcpserver.exceptions.ToolError so that MCP hosts show the message instead of a generic failure.

This module imports only the standard library.

AgentToolError

Bases: Exception

An anticipated tool failure whose message is shown to the model.

Source code in agribound/agent/errors.py

class AgentToolError(Exception):
    """An anticipated tool failure whose message is shown to the model."""

GateError

Bases: AgentToolError

Base class of confirmation-gate refusals.

Source code in agribound/agent/errors.py

class GateError(AgentToolError):
    """Base class of confirmation-gate refusals."""

ExecutionDisabledError

Bases: GateError

execute_plan was called while execution is disabled (dry run).

Source code in agribound/agent/errors.py

class ExecutionDisabledError(GateError):
    """``execute_plan`` was called while execution is disabled (dry run)."""

ExecutionDeniedError

Bases: GateError

The human reviewer (or the non-interactive default) denied a plan.

Source code in agribound/agent/errors.py

class ExecutionDeniedError(GateError):
    """The human reviewer (or the non-interactive default) denied a plan."""

ExecutionNotApprovedError

Bases: GateError

No unused approval exists for the plan's hash.

Source code in agribound/agent/errors.py

class ExecutionNotApprovedError(GateError):
    """No unused approval exists for the plan's hash."""

ExecutionLimitError

Bases: GateError

The session already ran max_executions plans.

Source code in agribound/agent/errors.py

class ExecutionLimitError(GateError):
    """The session already ran ``max_executions`` plans."""

PlanChangedError

Bases: GateError

The plan's configuration or inputs no longer match its hash.

Source code in agribound/agent/errors.py

class PlanChangedError(GateError):
    """The plan's configuration or inputs no longer match its hash."""

UnknownPlanError

Bases: AgentToolError

No plan with the requested ID exists in this session.

Source code in agribound/agent/errors.py

class UnknownPlanError(AgentToolError):
    """No plan with the requested ID exists in this session."""

NetworkDisabledError

Bases: AgentToolError

A tool (or a plan's pipeline run) needs a remote service while network access is off.

Source code in agribound/agent/errors.py

class NetworkDisabledError(AgentToolError):
    """A tool (or a plan's pipeline run) needs a remote service while network access is off."""

AgentDependencyError

Bases: ImportError

An optional dependency of the agent layer is not installed.

Source code in agribound/agent/errors.py

class AgentDependencyError(ImportError):
    """An optional dependency of the agent layer is not installed."""

Session transcript

session

Agent session transcript.

:class:AgentSession records one agent session as JSON (<workdir>/agent_session_<session_id>.json): the request, backend and model ID, package/SDK versions, every model turn (stop reason, text, tool calls, token usage, duration), every tool call (validated arguments, error or result summary, duration), the plans, the confirmation gate's approvals and denials (who, how, when), the executions, and the final status. The file is rewritten after every turn and once more when an approved plan starts to run (with the approval and a "running" execution record), so an interrupted session, including one that crashes during a long pipeline run, still leaves a record.

AgentSession

Collects and writes the transcript of one agent session.

Parameters:

Name	Type	Description	Default
request	str	The user's natural-language request.	required
workdir	str or Path	Session directory; the transcript is written there.	required
session_id	str or None	Identifier (default :func:agribound._repro.new_run_id).	None
backend_info	dict or None	Backend description (name, model, base URL, request options).	None
options	dict or None	Session options (dry run, limits, study area, ...).	None

Source code in agribound/agent/session.py

class AgentSession:
    """Collects and writes the transcript of one agent session.

    Parameters
    ----------
    request : str
        The user's natural-language request.
    workdir : str or Path
        Session directory; the transcript is written there.
    session_id : str or None
        Identifier (default :func:`agribound._repro.new_run_id`).
    backend_info : dict or None
        Backend description (name, model, base URL, request options).
    options : dict or None
        Session options (dry run, limits, study area, ...).
    """

    def __init__(
        self,
        request: str,
        *,
        workdir: str | Path,
        session_id: str | None = None,
        backend_info: dict[str, Any] | None = None,
        options: dict[str, Any] | None = None,
    ) -> None:
        from agribound._repro import collect_versions, new_run_id

        self.request = str(request)
        self.workdir = Path(workdir)
        self.session_id = session_id or new_run_id()
        self.path = self.workdir / f"agent_session_{self.session_id}.json"
        self.backend_info = dict(backend_info or {})
        self.options = dict(options or {})
        self.versions = collect_versions(extra=("anthropic", "mcp", "mcp-types", "pydantic"))
        self.started_utc = _utc_now()
        self.finished_utc: str | None = None
        self._t0 = time.perf_counter()
        self.turns: list[dict[str, Any]] = []
        self.tool_calls: list[dict[str, Any]] = []
        self.plans: list[dict[str, Any]] = []
        self.executions: list[dict[str, Any]] = []
        self.warnings: list[str] = []
        self.gate_state: dict[str, Any] | None = None
        self.status = "running"
        self.final_text = ""
        self.report = ""
        self.error: str | None = None
        self.stop_details: dict[str, Any] | None = None

    # -- recording -------------------------------------------------------------

    def record_turn(self, turn: Any, *, duration_s: float) -> dict[str, Any]:
        """Record one model response (a :class:`~agribound.agent.backends.base.ModelTurn`)."""
        entry = {
            "index": len(self.turns),
            "stop_reason": turn.stop_reason,
            "model": turn.model,
            "request_id": turn.request_id,
            "text": turn.text,
            "tool_calls": [{"id": c.id, "name": c.name} for c in turn.tool_calls],
            "usage": dict(turn.usage or {}),
            "iterations": [dict(it) for it in getattr(turn, "iterations", None) or []],
            "stop_details": turn.stop_details,
            "fallbacks": list(turn.fallbacks or []),
            "served_by_fallback": bool(getattr(turn, "served_by_fallback", False)),
            "duration_s": round(float(duration_s), 3),
        }
        self.turns.append(entry)
        return entry

    def record_tool_call(
        self,
        *,
        turn_index: int,
        call_id: str,
        name: str,
        outcome: Any,
        duration_s: float,
    ) -> dict[str, Any]:
        """Record one tool call and its :class:`~agribound.agent.tools.ToolOutcome`."""
        entry = {
            "turn": turn_index,
            "id": call_id,
            "name": name,
            "arguments": outcome.arguments,
            "arguments_valid": outcome.arguments_valid,
            "is_error": not outcome.ok,
            "error": outcome.error,
            "result_summary": summarize_result(outcome.output) if outcome.ok else None,
            "duration_s": round(float(duration_s), 3),
        }
        self.tool_calls.append(entry)
        return entry

    def record_plan(self, plan: Any) -> None:
        """Record a proposed plan.

        A re-proposal of the same configuration has the same plan ID (the ID
        hashes the configuration and inputs, not the agent's text). Its entry
        is replaced by the latest proposal, which is the version the
        confirmation gate shows to the reviewer, and ``n_proposals`` counts
        how often it was proposed.
        """
        entry = plan.to_dict()
        for i, existing in enumerate(self.plans):
            if existing["plan_id"] == plan.plan_id:
                entry["n_proposals"] = int(existing.get("n_proposals", 1)) + 1
                self.plans[i] = entry
                return
        entry["n_proposals"] = 1
        self.plans.append(entry)

    def record_execution(self, execution: dict[str, Any]) -> None:
        """Record an execution attempt, or update it.

        An entry with the same ``"attempt"`` number is replaced (a
        ``"running"`` record is written when the run starts and replaced by
        the final ``"success"``/``"failed"`` record).
        """
        entry = dict(execution)
        attempt = entry.get("attempt")
        if attempt is not None:
            for i, existing in enumerate(self.executions):
                if existing.get("attempt") == attempt:
                    self.executions[i] = entry
                    return
        self.executions.append(entry)

    def add_warning(self, message: str) -> None:
        self.warnings.append(str(message))

    def set_gate(self, gate: Any) -> None:
        """Snapshot the confirmation gate's approvals, denials and counters."""
        self.gate_state = gate.to_dict() if gate is not None else None

    def finish(
        self,
        status: str,
        *,
        final_text: str = "",
        report: str = "",
        error: str | None = None,
        stop_details: dict[str, Any] | None = None,
    ) -> None:
        """Set the final status and timing."""
        self.status = status
        self.final_text = final_text
        self.report = report
        self.error = error
        self.stop_details = stop_details
        self.finished_utc = _utc_now()

    # -- output ----------------------------------------------------------------

    def usage_totals(self) -> dict[str, int]:
        """Token usage summed over all turns.

        For a turn that reports per-attempt ``iterations`` (e.g. a declined
        attempt followed by a fallback model), the iterations are summed,
        because the top-level usage then covers only the final attempt;
        otherwise the top-level usage is used.
        """
        totals = dict.fromkeys(_USAGE_KEYS, 0)
        for turn in self.turns:
            entries = turn.get("iterations") or [turn.get("usage") or {}]
            for entry in entries:
                for key in _USAGE_KEYS:
                    value = entry.get(key)
                    if isinstance(value, int):
                        totals[key] += value
        return totals

    def to_dict(self) -> dict[str, Any]:
        return {
            "schema_version": SESSION_SCHEMA_VERSION,
            "session_id": self.session_id,
            "request": self.request,
            "status": self.status,
            "error": self.error,
            "stop_details": self.stop_details,
            "backend": self.backend_info,
            "options": self.options,
            "versions": self.versions,
            "platform": platform.platform(),
            "started_utc": self.started_utc,
            "finished_utc": self.finished_utc,
            "wall_s": round(time.perf_counter() - self._t0, 3),
            "usage_totals": self.usage_totals(),
            "turns": self.turns,
            "tool_calls": self.tool_calls,
            "plans": self.plans,
            "gate": self.gate_state,
            "executions": self.executions,
            "warnings": self.warnings,
            "final_text": self.final_text,
            "report": self.report,
        }

    def write(self) -> Path:
        """Write the transcript atomically and return its path."""
        from agribound.provenance import to_jsonable

        self.workdir.mkdir(parents=True, exist_ok=True)
        tmp = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp")
        with open(tmp, "w") as f:
            json.dump(to_jsonable(self.to_dict()), f, indent=2)
            f.write("\n")
        os.replace(tmp, self.path)
        return self.path

record_turn

record_turn(turn: Any, *, duration_s: float) -> dict[str, Any]

Record one model response (a :class:~agribound.agent.backends.base.ModelTurn).

Source code in agribound/agent/session.py

def record_turn(self, turn: Any, *, duration_s: float) -> dict[str, Any]:
    """Record one model response (a :class:`~agribound.agent.backends.base.ModelTurn`)."""
    entry = {
        "index": len(self.turns),
        "stop_reason": turn.stop_reason,
        "model": turn.model,
        "request_id": turn.request_id,
        "text": turn.text,
        "tool_calls": [{"id": c.id, "name": c.name} for c in turn.tool_calls],
        "usage": dict(turn.usage or {}),
        "iterations": [dict(it) for it in getattr(turn, "iterations", None) or []],
        "stop_details": turn.stop_details,
        "fallbacks": list(turn.fallbacks or []),
        "served_by_fallback": bool(getattr(turn, "served_by_fallback", False)),
        "duration_s": round(float(duration_s), 3),
    }
    self.turns.append(entry)
    return entry

record_tool_call

record_tool_call(*, turn_index: int, call_id: str, name: str, outcome: Any, duration_s: float) -> dict[str, Any]

Record one tool call and its :class:~agribound.agent.tools.ToolOutcome.

Source code in agribound/agent/session.py

def record_tool_call(
    self,
    *,
    turn_index: int,
    call_id: str,
    name: str,
    outcome: Any,
    duration_s: float,
) -> dict[str, Any]:
    """Record one tool call and its :class:`~agribound.agent.tools.ToolOutcome`."""
    entry = {
        "turn": turn_index,
        "id": call_id,
        "name": name,
        "arguments": outcome.arguments,
        "arguments_valid": outcome.arguments_valid,
        "is_error": not outcome.ok,
        "error": outcome.error,
        "result_summary": summarize_result(outcome.output) if outcome.ok else None,
        "duration_s": round(float(duration_s), 3),
    }
    self.tool_calls.append(entry)
    return entry

record_plan

record_plan(plan: Any) -> None

Record a proposed plan.

A re-proposal of the same configuration has the same plan ID (the ID hashes the configuration and inputs, not the agent's text). Its entry is replaced by the latest proposal, which is the version the confirmation gate shows to the reviewer, and n_proposals counts how often it was proposed.

Source code in agribound/agent/session.py

def record_plan(self, plan: Any) -> None:
    """Record a proposed plan.

    A re-proposal of the same configuration has the same plan ID (the ID
    hashes the configuration and inputs, not the agent's text). Its entry
    is replaced by the latest proposal, which is the version the
    confirmation gate shows to the reviewer, and ``n_proposals`` counts
    how often it was proposed.
    """
    entry = plan.to_dict()
    for i, existing in enumerate(self.plans):
        if existing["plan_id"] == plan.plan_id:
            entry["n_proposals"] = int(existing.get("n_proposals", 1)) + 1
            self.plans[i] = entry
            return
    entry["n_proposals"] = 1
    self.plans.append(entry)

record_execution

record_execution(execution: dict[str, Any]) -> None

Record an execution attempt, or update it.

An entry with the same "attempt" number is replaced (a "running" record is written when the run starts and replaced by the final "success"/"failed" record).

Source code in agribound/agent/session.py

def record_execution(self, execution: dict[str, Any]) -> None:
    """Record an execution attempt, or update it.

    An entry with the same ``"attempt"`` number is replaced (a
    ``"running"`` record is written when the run starts and replaced by
    the final ``"success"``/``"failed"`` record).
    """
    entry = dict(execution)
    attempt = entry.get("attempt")
    if attempt is not None:
        for i, existing in enumerate(self.executions):
            if existing.get("attempt") == attempt:
                self.executions[i] = entry
                return
    self.executions.append(entry)

set_gate

set_gate(gate: Any) -> None

Snapshot the confirmation gate's approvals, denials and counters.

Source code in agribound/agent/session.py

def set_gate(self, gate: Any) -> None:
    """Snapshot the confirmation gate's approvals, denials and counters."""
    self.gate_state = gate.to_dict() if gate is not None else None

finish

finish(status: str, *, final_text: str = '', report: str = '', error: str | None = None, stop_details: dict[str, Any] | None = None) -> None

Set the final status and timing.

Source code in agribound/agent/session.py

def finish(
    self,
    status: str,
    *,
    final_text: str = "",
    report: str = "",
    error: str | None = None,
    stop_details: dict[str, Any] | None = None,
) -> None:
    """Set the final status and timing."""
    self.status = status
    self.final_text = final_text
    self.report = report
    self.error = error
    self.stop_details = stop_details
    self.finished_utc = _utc_now()

usage_totals

usage_totals() -> dict[str, int]

Token usage summed over all turns.

For a turn that reports per-attempt iterations (e.g. a declined attempt followed by a fallback model), the iterations are summed, because the top-level usage then covers only the final attempt; otherwise the top-level usage is used.

Source code in agribound/agent/session.py

def usage_totals(self) -> dict[str, int]:
    """Token usage summed over all turns.

    For a turn that reports per-attempt ``iterations`` (e.g. a declined
    attempt followed by a fallback model), the iterations are summed,
    because the top-level usage then covers only the final attempt;
    otherwise the top-level usage is used.
    """
    totals = dict.fromkeys(_USAGE_KEYS, 0)
    for turn in self.turns:
        entries = turn.get("iterations") or [turn.get("usage") or {}]
        for entry in entries:
            for key in _USAGE_KEYS:
                value = entry.get(key)
                if isinstance(value, int):
                    totals[key] += value
    return totals

write

write() -> Path

Write the transcript atomically and return its path.

Source code in agribound/agent/session.py

def write(self) -> Path:
    """Write the transcript atomically and return its path."""
    from agribound.provenance import to_jsonable

    self.workdir.mkdir(parents=True, exist_ok=True)
    tmp = self.path.with_name(f".{self.path.name}.{os.getpid()}.tmp")
    with open(tmp, "w") as f:
        json.dump(to_jsonable(self.to_dict()), f, indent=2)
        f.write("\n")
    os.replace(tmp, self.path)
    return self.path

MCP server

mcp_server

MCP server exposing the Agribound agent tools (mcp >= 2.2).

agribound mcp serve starts an :class:mcp.server.mcpserver.MCPServer named "agribound" over stdio (default) or Streamable HTTP. Any MCP host (Claude Desktop/Code or a local-LLM MCP host) can then call the same typed tools that the built-in agent loop uses (:mod:agribound.agent.tools).

Tools and annotations

• Read-only tools carry ToolAnnotations(read_only_hint=True). query_published_ftw and propose_run write files into the server's work directory (downloaded polygons, plan YAML), so they are annotated read_only_hint=False, destructive_hint=False.
• propose_run is always registered; it never runs anything.
• execute_plan is registered only when the server is started with --allow-execute. At most max_executions (default 1) plans run per server process.
• Arguments are validated by the same pydantic models as in the local loop, including the rejection of unknown argument names (MCPServer's own argument model would silently drop them; the raw request arguments are checked).

Work directory

Default: $XDG_DATA_HOME/agribound/mcp or ~/.local/share/agribound/mcp (:func:default_workdir), not a path relative to the working directory the MCP host happens to start the server in. --workdir overrides it.

Human confirmation of execute_plan

• confirm="elicit" (default): before anything runs, the server sends an MCP elicitation showing the full plan and asks the user to type yes. Clients that did not declare the elicitation capability get an error explaining the alternative. MCP lets a client answer elicitations itself (for example an automated host), so the server cannot prove that a human answered; the answer and method are recorded with the approval.
• confirm="host": no elicitation; the server relies on the host's own per-tool approval prompt. Use it only with hosts that ask the user before every tool call. The approval is recorded with the method "host tool-approval prompt" and an approver the server cannot verify.

In both modes a plan that cannot run (unknown plan, no executions left, changed inputs, or remote services needed while allow_network=False) is refused before the user is asked (:func:agribound.agent.tools.preflight_execution).

Streamable HTTP

The streamable-http transport has no authentication: any process or user that can reach the host and port can call the tools, and with --allow-execute such a client can answer the execute_plan elicitation itself. :func:serve therefore refuses (:func:check_http_exposure, before the server is built) --allow-execute over streamable-http and a --host that is not a loopback address, unless allow_unauthenticated_http=True (--allow-unauthenticated-http), which logs a WARNING. The default host is 127.0.0.1.

Expected failures are raised as ToolError so the model reads the reason. For execute_plan every failure after the approval (the pipeline, the gate's final checks, or an unexpected exception) is reported as a ToolError with its type and message. For the other tools, an exception that is not an :class:~agribound.agent.errors.AgentToolError is reported by MCPServer as a generic "Error executing tool" and logged by the server. While a tool runs, sys.stdout is redirected to sys.stderr; the mcp stdio transport additionally points file descriptor 1 at stderr while serving (best effort, mcp.server.stdio), so stray prints, including those of native code, do not corrupt the protocol stream. Long runs send report_progress notifications every 15 s.

build_server

build_server(*, allow_execute: bool = False, confirm: Literal['elicit', 'host'] = 'elicit', workdir: str | Path | None = None, study_area: str | None = None, gee_project: str | None = None, reference_boundaries: str | None = None, allow_network: bool = True, max_executions: int = 1) -> Any

Create the Agribound MCP server.

Parameters:

Name	Type	Description	Default
allow_execute	bool	Register execute_plan (default False: plans can be proposed but not run through MCP).	False
confirm	('elicit', 'host')	How execute_plan obtains human confirmation (module docstring).	"elicit"
workdir	(str, Path or None)	Directory for plans, outputs, downloads and the cache (default :func:default_workdir).	None
study_area	str or None	Session defaults for the tools.	None
gee_project	str or None	Session defaults for the tools.	None
reference_boundaries	str or None	Session defaults for the tools.	None
allow_network	bool	Allow tools to contact Earth Engine, TESSERA, Source Cooperative or the USGS NAIP Plus ImageServer. With False, execute_plan also refuses plans whose pipeline run needs one of them (:attr:agribound.agent.tools.ToolContext.allow_network).	True
max_executions	int	Maximum number of plans executed by this server process.	1

Returns:

Type	Description
MCPServer	The server; its tool context is available as server.agribound_context.

Source code in agribound/agent/mcp_server.py

def build_server(
    *,
    allow_execute: bool = False,
    confirm: Literal["elicit", "host"] = "elicit",
    workdir: str | Path | None = None,
    study_area: str | None = None,
    gee_project: str | None = None,
    reference_boundaries: str | None = None,
    allow_network: bool = True,
    max_executions: int = 1,
) -> Any:
    """Create the Agribound MCP server.

    Parameters
    ----------
    allow_execute : bool
        Register ``execute_plan`` (default *False*: plans can be proposed but
        not run through MCP).
    confirm : {"elicit", "host"}
        How ``execute_plan`` obtains human confirmation (module docstring).
    workdir : str, Path or None
        Directory for plans, outputs, downloads and the cache (default
        :func:`default_workdir`).
    study_area, gee_project, reference_boundaries : str or None
        Session defaults for the tools.
    allow_network : bool
        Allow tools to contact Earth Engine, TESSERA, Source Cooperative or
        the USGS NAIP Plus ImageServer. With *False*, ``execute_plan`` also
        refuses plans whose pipeline run needs one of them
        (:attr:`agribound.agent.tools.ToolContext.allow_network`).
    max_executions : int
        Maximum number of plans executed by this server process.

    Returns
    -------
    mcp.server.mcpserver.MCPServer
        The server; its tool context is available as ``server.agribound_context``.
    """
    if confirm not in CONFIRM_MODES:
        raise ValueError(f"confirm must be one of {CONFIRM_MODES}, got {confirm!r}")
    m = _import_mcp()
    from agribound._version import __version__
    from agribound.agent.gate import ConfirmationGate, deny_all
    from agribound.agent.prompts import MCP_INSTRUCTIONS
    from agribound.agent.tools import TOOL_SPECS, ToolContext

    context = ToolContext(
        workdir=Path(workdir) if workdir is not None else default_workdir(),
        study_area=study_area,
        gee_project=gee_project,
        reference_boundaries=reference_boundaries,
        allow_network=allow_network,
        execution_enabled=bool(allow_execute),
        # Approvals come only from record_approval below; the callback never approves.
        gate=ConfirmationGate(deny_all, max_executions=max_executions),
    )
    logger.info("Agribound MCP work directory: %s", context.workdir)
    server = m["MCPServer"](name="agribound", instructions=MCP_INSTRUCTIONS, version=__version__)
    tool_error = m["ToolError"]
    for spec in TOOL_SPECS:
        if spec.name == "execute_plan":
            continue
        server.add_tool(
            _wrap_tool(spec, context, tool_error, m["Context"]),
            name=spec.name,
            description=spec.description,
            annotations=_annotations_for(spec, m["ToolAnnotations"]),
            structured_output=True,
        )
    if allow_execute:
        spec = next(s for s in TOOL_SPECS if s.name == "execute_plan")
        server.add_tool(
            _execute_tool(context, confirm, m),
            name=spec.name,
            description=spec.description
            + (
                " The user confirms through an MCP elicitation."
                if confirm == "elicit"
                else " Confirmation relies on the host's tool-approval prompt."
            ),
            annotations=_annotations_for(spec, m["ToolAnnotations"]),
            structured_output=True,
        )
    server.agribound_context = context
    return server

default_workdir

default_workdir() -> Path

$XDG_DATA_HOME/agribound/mcp, else ~/.local/share/agribound/mcp.

Source code in agribound/agent/mcp_server.py

def default_workdir() -> Path:
    """``$XDG_DATA_HOME/agribound/mcp``, else ``~/.local/share/agribound/mcp``."""
    base = os.environ.get("XDG_DATA_HOME")
    return (Path(base) if base else Path.home() / ".local" / "share") / "agribound" / "mcp"

Anthropic backend

anthropic_backend

Anthropic Messages API backend (manual tool-use loop).

Requests

First-party API (no custom base URL): client.beta.messages.create with

• model (default "claude-opus-5", overridable by argument or the AGRIBOUND_AGENT_MODEL environment variable),
• max_tokens=16000, thinking={"type": "adaptive"}, output_config={"effort": "high"}, tool_choice={"type": "auto"},
• the system prompt as one text block with cache_control (ephemeral),
• server-side refusal fallbacks: betas=["server-side-fallback-2026-07-01"] and fallbacks="default" (both typed parameters of client.beta.messages.create in anthropic 1.8.0).

Custom base_url (e.g. Ollama >= 0.14 or vLLM Anthropic-compatible /v1/messages endpoints): client.messages.create without betas, fallbacks, cache_control or tool_choice (these servers do not all support them; auto is the API default anyway). thinking and effort are sent only when given explicitly. The options actually used are recorded in :meth:AnthropicBackend.info and therefore in the session transcript.

No sampling parameters (temperature, top_p, top_k) are sent; anthropic 1.x removed them from messages.create.

Responses

When a response contains fallback blocks (one per model that declined mid-output), :meth:AnthropicBackend.to_turn returns only the tool_use blocks after the last fallback block as tool calls, and :meth:AnthropicBackend.assistant_message omits the thinking, redacted_thinking and tool_use blocks (and unpaired server-tool blocks or unknown block types) that precede that boundary when the turn is echoed back (:func:echoable_content); a WARNING is logged whenever a block is omitted. Text blocks, paired server-tool blocks, the fallback blocks and everything after the boundary are echoed unchanged. usage.iterations (per-attempt usage) is recorded, and a fallback_message entry marks a response served by a fallback model.

The two sources available when this was written disagree about pre-boundary thinking blocks:

• the claude-api skill (bundled with Claude Code 2.1.282, read 2026-09-27; shared/model-migration.md, "Echoing fallback turns back") says to omit thinking, redacted_thinking and tool_use blocks before the final fallback block; this module follows it;
• the BetaFallbackBlockParam docstring of anthropic 1.8.0 says to echo the assistant turn back verbatim with the fallback block in its original position, and that the server validates thinking runs on both sides of it.

Both agree that the declined model's tool_use blocks are not run and not echoed without results. The rule has not been checked against the live API. The same skill page says that for non-streaming requests (the only kind this backend sends) a mid-output decline omits the declined partial entirely, so the pre-boundary branch is not expected to be reached in practice.

AnthropicBackend

:class:~agribound.agent.backends.base.LLMBackend for the Anthropic Messages API.

Parameters:

Name	Type	Description	Default
model	str or None	Model ID. Default: $AGRIBOUND_AGENT_MODEL or "claude-opus-5".	None
base_url	str or None	Custom endpoint (Anthropic-compatible local server). None uses the SDK's resolution (ANTHROPIC_BASE_URL, profile, then https://api.anthropic.com).	None
api_key	str or None	API key; None lets the SDK resolve credentials. Local servers need a placeholder key (e.g. "ollama").	None
client	object or None	Pre-built client (anything with messages.create and beta.messages.create); used by tests.	None
max_tokens	int	Output token limit per response (default 16000).	DEFAULT_MAX_TOKENS
effort	Any	AUTO (default) enables "high" / {"type": "adaptive"} / "default" / True for the first-party API and disables them for a custom base URL. None (or False) disables; any other value is sent as given. Non-default models may not accept these options (for example claude-haiku-4-5 does not support adaptive thinking).	AUTO
thinking	Any	AUTO (default) enables "high" / {"type": "adaptive"} / "default" / True for the first-party API and disables them for a custom base URL. None (or False) disables; any other value is sent as given. Non-default models may not accept these options (for example claude-haiku-4-5 does not support adaptive thinking).	AUTO
fallbacks	Any	AUTO (default) enables "high" / {"type": "adaptive"} / "default" / True for the first-party API and disables them for a custom base URL. None (or False) disables; any other value is sent as given. Non-default models may not accept these options (for example claude-haiku-4-5 does not support adaptive thinking).	AUTO
cache_system_prompt	Any	AUTO (default) enables "high" / {"type": "adaptive"} / "default" / True for the first-party API and disables them for a custom base URL. None (or False) disables; any other value is sent as given. Non-default models may not accept these options (for example claude-haiku-4-5 does not support adaptive thinking).	AUTO
timeout	float | None	Passed to :class:anthropic.Anthropic when client is not given.	None
max_retries	float | None	Passed to :class:anthropic.Anthropic when client is not given.	None

Source code in agribound/agent/backends/anthropic_backend.py

class AnthropicBackend:
    """:class:`~agribound.agent.backends.base.LLMBackend` for the Anthropic Messages API.

    Parameters
    ----------
    model : str or None
        Model ID. Default: ``$AGRIBOUND_AGENT_MODEL`` or ``"claude-opus-5"``.
    base_url : str or None
        Custom endpoint (Anthropic-compatible local server). *None* uses the
        SDK's resolution (``ANTHROPIC_BASE_URL``, profile, then
        ``https://api.anthropic.com``).
    api_key : str or None
        API key; *None* lets the SDK resolve credentials. Local servers need
        a placeholder key (e.g. ``"ollama"``).
    client : object or None
        Pre-built client (anything with ``messages.create`` and
        ``beta.messages.create``); used by tests.
    max_tokens : int
        Output token limit per response (default 16000).
    effort, thinking, fallbacks, cache_system_prompt
        ``AUTO`` (default) enables ``"high"`` / ``{"type": "adaptive"}`` /
        ``"default"`` / *True* for the first-party API and disables them for
        a custom base URL. *None* (or *False*) disables; any other value is
        sent as given. Non-default models may not accept these options (for
        example ``claude-haiku-4-5`` does not support adaptive thinking).
    timeout, max_retries
        Passed to :class:`anthropic.Anthropic` when *client* is not given.
    """

    name = "anthropic"

    def __init__(
        self,
        model: str | None = None,
        *,
        base_url: str | None = None,
        api_key: str | None = None,
        client: Any = None,
        max_tokens: int = DEFAULT_MAX_TOKENS,
        effort: Any = AUTO,
        thinking: Any = AUTO,
        fallbacks: Any = AUTO,
        cache_system_prompt: Any = AUTO,
        timeout: float | None = None,
        max_retries: int | None = None,
    ) -> None:
        self.model = model or os.environ.get(MODEL_ENV_VAR) or DEFAULT_MODEL
        self.sdk_version: str | None = None
        if client is None:
            try:
                import anthropic
            except ImportError as exc:
                raise AgentDependencyError(
                    "The Anthropic backend needs the 'anthropic' package: "
                    'pip install "agribound[agent]"'
                ) from exc
            kwargs: dict[str, Any] = {}
            if api_key is not None:
                kwargs["api_key"] = api_key
            if base_url is not None:
                kwargs["base_url"] = base_url
            if timeout is not None:
                kwargs["timeout"] = timeout
            if max_retries is not None:
                kwargs["max_retries"] = max_retries
            client = anthropic.Anthropic(**kwargs)
            self.sdk_version = getattr(anthropic, "__version__", None)
        self._client = client
        resolved = getattr(client, "base_url", None) or base_url
        self.base_url = str(resolved) if resolved else None
        self.first_party = is_first_party(self.base_url)
        self.max_tokens = int(max_tokens)

        def pick(value: Any, first_party_default: Any) -> Any:
            if value is AUTO:
                return first_party_default if self.first_party else None
            return value or None

        self.effort = pick(effort, DEFAULT_EFFORT)
        self.thinking = pick(thinking, {"type": "adaptive"})
        self.fallbacks = pick(fallbacks, "default")
        self.cache_system_prompt = bool(pick(cache_system_prompt, True))
        self.tool_choice = {"type": "auto"} if self.first_party else None
        if not self.first_party:
            logger.info(
                "Custom Anthropic-compatible endpoint %s: fallbacks, prompt caching and "
                "tool_choice are not sent%s",
                self.base_url,
                "" if (self.thinking or self.effort) else "; neither are thinking and effort",
            )

    # -- LLMBackend ------------------------------------------------------------

    def info(self) -> dict[str, Any]:
        return {
            "backend": self.name,
            "model": self.model,
            "base_url": self.base_url,
            "first_party": self.first_party,
            "sdk": "anthropic",
            "sdk_version": self.sdk_version,
            "request_options": {
                "endpoint": "beta.messages.create" if self.fallbacks else "messages.create",
                "max_tokens": self.max_tokens,
                "thinking": self.thinking,
                "effort": self.effort,
                "fallbacks": self.fallbacks,
                "betas": self._betas(),
                "tool_choice": self.tool_choice,
                "cache_system_prompt": self.cache_system_prompt,
            },
        }

    def user_message(self, text: str) -> dict[str, Any]:
        return {"role": "user", "content": text}

    def assistant_message(self, turn: ModelTurn) -> dict[str, Any]:
        # All content (thinking, text, tool_use, fallback blocks) is echoed, except the
        # blocks before a mid-output fallback boundary that must not be (echoable_content).
        return {"role": "assistant", "content": echoable_content(turn.raw_content or [])}

    def tool_results_message(self, results: Sequence[ToolResult]) -> dict[str, Any]:
        return {
            "role": "user",
            "content": [
                {
                    "type": "tool_result",
                    "tool_use_id": r.tool_call_id,
                    "content": r.content,
                    "is_error": bool(r.is_error),
                }
                for r in results
            ],
        }

    def _betas(self) -> list[str] | None:
        if not self.fallbacks:
            return None
        return [FALLBACK_BETA_DEFAULT if self.fallbacks == "default" else FALLBACK_BETA_ARRAY]

    def request_kwargs(
        self,
        *,
        system: str,
        tools: Sequence[ToolDefinition],
        messages: Sequence[Any],
    ) -> dict[str, Any]:
        """Keyword arguments of the ``messages.create`` call (exposed for tests)."""
        kwargs: dict[str, Any] = {
            "model": self.model,
            "max_tokens": self.max_tokens,
            "messages": list(messages),
            "tools": [
                {"name": t.name, "description": t.description, "input_schema": t.input_schema}
                for t in tools
            ],
        }
        if self.cache_system_prompt:
            kwargs["system"] = [
                {"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}
            ]
        else:
            kwargs["system"] = system
        if self.tool_choice is not None:
            kwargs["tool_choice"] = dict(self.tool_choice)
        if self.thinking:
            kwargs["thinking"] = dict(self.thinking)
        if self.effort:
            kwargs["output_config"] = {"effort": self.effort}
        if self.fallbacks:
            kwargs["betas"] = self._betas()
            kwargs["fallbacks"] = self.fallbacks
        return kwargs

    def complete(
        self,
        *,
        system: str,
        tools: Sequence[ToolDefinition],
        messages: Sequence[Any],
    ) -> ModelTurn:
        kwargs = self.request_kwargs(system=system, tools=tools, messages=messages)
        if "fallbacks" in kwargs:
            response = self._client.beta.messages.create(**kwargs)
        else:
            response = self._client.messages.create(**kwargs)
        return self.to_turn(response)

    @staticmethod
    def to_turn(response: Any) -> ModelTurn:
        """Convert an SDK ``Message``/``BetaMessage`` into a :class:`ModelTurn`.

        ``text`` joins every text block. ``tool_calls`` holds only the
        ``tool_use`` blocks after the last ``fallback`` block (a declined
        model's tool calls are not run; see the module docstring).
        """
        content = list(getattr(response, "content", None) or [])
        boundary = _last_fallback_index(content)
        texts, calls, fallbacks = [], [], []
        for i, block in enumerate(content):
            kind = getattr(block, "type", None)
            if kind == "text":
                texts.append(block.text)
            elif kind == "tool_use" and i > boundary:
                calls.append(ToolCall(id=block.id, name=block.name, input=block.input))
            elif kind == "fallback":
                trigger = getattr(block, "trigger", None)
                fallbacks.append(
                    {
                        "from": getattr(getattr(block, "from_", None), "model", None),
                        "to": getattr(getattr(block, "to", None), "model", None),
                        "trigger_category": getattr(trigger, "category", None),
                    }
                )
        usage_obj = getattr(response, "usage", None)
        usage = {}
        for key in _USAGE_KEYS:
            value = getattr(usage_obj, key, None)
            if value is not None:
                usage[key] = value
        iterations = []
        for entry in getattr(usage_obj, "iterations", None) or []:
            item: dict[str, Any] = {"type": getattr(entry, "type", None)}
            model = getattr(entry, "model", None)
            if model is not None:
                item["model"] = str(model)
            for key in _USAGE_KEYS:
                value = getattr(entry, key, None)
                if value is not None:
                    item[key] = value
            iterations.append(item)
        return ModelTurn(
            stop_reason=str(getattr(response, "stop_reason", None)),
            text="\n".join(texts),
            tool_calls=calls,
            raw_content=content,
            usage=usage,
            stop_details=_dump(getattr(response, "stop_details", None)),
            model=getattr(response, "model", None),
            request_id=getattr(response, "_request_id", None),
            fallbacks=fallbacks,
            served_by_fallback=any(it["type"] == "fallback_message" for it in iterations),
            iterations=iterations,
        )

request_kwargs

request_kwargs(*, system: str, tools: Sequence[ToolDefinition], messages: Sequence[Any]) -> dict[str, Any]

Keyword arguments of the messages.create call (exposed for tests).

Source code in agribound/agent/backends/anthropic_backend.py

def request_kwargs(
    self,
    *,
    system: str,
    tools: Sequence[ToolDefinition],
    messages: Sequence[Any],
) -> dict[str, Any]:
    """Keyword arguments of the ``messages.create`` call (exposed for tests)."""
    kwargs: dict[str, Any] = {
        "model": self.model,
        "max_tokens": self.max_tokens,
        "messages": list(messages),
        "tools": [
            {"name": t.name, "description": t.description, "input_schema": t.input_schema}
            for t in tools
        ],
    }
    if self.cache_system_prompt:
        kwargs["system"] = [
            {"type": "text", "text": system, "cache_control": {"type": "ephemeral"}}
        ]
    else:
        kwargs["system"] = system
    if self.tool_choice is not None:
        kwargs["tool_choice"] = dict(self.tool_choice)
    if self.thinking:
        kwargs["thinking"] = dict(self.thinking)
    if self.effort:
        kwargs["output_config"] = {"effort": self.effort}
    if self.fallbacks:
        kwargs["betas"] = self._betas()
        kwargs["fallbacks"] = self.fallbacks
    return kwargs

to_turn staticmethod

to_turn(response: Any) -> ModelTurn

Convert an SDK Message/BetaMessage into a :class:ModelTurn.

text joins every text block. tool_calls holds only the tool_use blocks after the last fallback block (a declined model's tool calls are not run; see the module docstring).

Source code in agribound/agent/backends/anthropic_backend.py

@staticmethod
def to_turn(response: Any) -> ModelTurn:
    """Convert an SDK ``Message``/``BetaMessage`` into a :class:`ModelTurn`.

    ``text`` joins every text block. ``tool_calls`` holds only the
    ``tool_use`` blocks after the last ``fallback`` block (a declined
    model's tool calls are not run; see the module docstring).
    """
    content = list(getattr(response, "content", None) or [])
    boundary = _last_fallback_index(content)
    texts, calls, fallbacks = [], [], []
    for i, block in enumerate(content):
        kind = getattr(block, "type", None)
        if kind == "text":
            texts.append(block.text)
        elif kind == "tool_use" and i > boundary:
            calls.append(ToolCall(id=block.id, name=block.name, input=block.input))
        elif kind == "fallback":
            trigger = getattr(block, "trigger", None)
            fallbacks.append(
                {
                    "from": getattr(getattr(block, "from_", None), "model", None),
                    "to": getattr(getattr(block, "to", None), "model", None),
                    "trigger_category": getattr(trigger, "category", None),
                }
            )
    usage_obj = getattr(response, "usage", None)
    usage = {}
    for key in _USAGE_KEYS:
        value = getattr(usage_obj, key, None)
        if value is not None:
            usage[key] = value
    iterations = []
    for entry in getattr(usage_obj, "iterations", None) or []:
        item: dict[str, Any] = {"type": getattr(entry, "type", None)}
        model = getattr(entry, "model", None)
        if model is not None:
            item["model"] = str(model)
        for key in _USAGE_KEYS:
            value = getattr(entry, key, None)
            if value is not None:
                item[key] = value
        iterations.append(item)
    return ModelTurn(
        stop_reason=str(getattr(response, "stop_reason", None)),
        text="\n".join(texts),
        tool_calls=calls,
        raw_content=content,
        usage=usage,
        stop_details=_dump(getattr(response, "stop_details", None)),
        model=getattr(response, "model", None),
        request_id=getattr(response, "_request_id", None),
        fallbacks=fallbacks,
        served_by_fallback=any(it["type"] == "fallback_message" for it in iterations),
        iterations=iterations,
    )