`symfonic.capabilities.tools`¶

tools ¶

Tool catalogue and selection capability (T3.1.2).

Two things live here:

The catalogue — one reading of the tool registry, and the only place tool metadata and the agent-facing manifest are derived. Whatever wants to know a tool's name, summary, visibility, cost, safety level, or required capability asks the catalogue; nothing re-derives it.

The selection stages — intent routing, lazy (procedural) routing, operator allowlists, role palettes, forced choice, and registry policy, each an object with the same signature, composed by an explicit pipeline that records what every stage decided. The order is a list you can read, not an emergent property of where the code happened to live.

The capability imports nothing but its own package (LAY-ADR §2): tools, registries, resolvers, and verdicts are all read structurally.

AllowlistStage ¶

AllowlistStage(*, always_include: Sequence[str] = (), catalog: ToolCatalog | None = None)

Re-add mandated tools that an earlier stage dropped.

A mandated name that is not in the catalogue is ignored rather than fabricated: the framework constrains the choice set, it never authors tools.

Source code in src/symfonic/capabilities/tools/stages/allowlist.py

def __init__(
    self,
    *,
    always_include: Sequence[str] = (),
    catalog: ToolCatalog | None = None,
) -> None:
    self._always_include = tuple(name for name in always_include if name)
    self._catalog = catalog

ForcedChoiceStage ¶

ForcedChoiceStage(*, resolver: Any = None, state_key: str = 'forced_tool_choice', is_tool_message: Callable[[Any], bool] = _default_is_tool_message)

Resolve the forced tool_choice for this iteration.

Precedence, unchanged from v7.10:

an explicit state["forced_tool_choice"] stamp — tests and adopters pre-stamp directly;
the registered resolver, which re-runs every iteration so force and release are symmetric by construction;
None — let the model choose, the safe default.

The release check runs on both paths: once a ToolMessage naming the forced tool is in history the force is spent. A resolver that raises releases too, because a stalled loop is worse than an unforced turn.

Source code in src/symfonic/capabilities/tools/stages/forced.py

def __init__(
    self,
    *,
    resolver: Any = None,
    state_key: str = "forced_tool_choice",
    is_tool_message: Callable[[Any], bool] = _default_is_tool_message,
) -> None:
    self._resolver = resolver
    self._state_key = state_key
    self._is_tool_message = is_tool_message

ForcedToolUnavailable ¶

Bases: ContractViolationError

A forced tool that cannot be offered. Raised before the model is called.

A ContractViolationError because that is what it is: two parts of a composition disagreeing, or a composition naming a tool that does not exist. A caller catching the framework's contract error catches this without knowing the name.

IntentRoutingStage ¶

IntentRoutingStage(*, verdict: Any = None, trigger_keywords: Mapping[str, Sequence[str]] | None = None, always_include: Sequence[str] = DEFAULT_ALWAYS_INCLUDE)

Narrow the turn's tools from an IntentVerdict-shaped object.

Decision matrix (unchanged from v7.0.1):

no verdict, or ambiguous — abstain. A noisy signal must not cost recall.
knowledge — keep only the always-include names.
action — keep the verdict's matched tools, plus every tool the domain declares no trigger keywords for (pre-v6.1.8 semantics: an absent entry means "always include"), plus always-include.

Source code in src/symfonic/capabilities/tools/stages/intent.py

def __init__(
    self,
    *,
    verdict: Any = None,
    trigger_keywords: Mapping[str, Sequence[str]] | None = None,
    always_include: Sequence[str] = DEFAULT_ALWAYS_INCLUDE,
) -> None:
    self._verdict = verdict
    self._trigger_keywords = dict(trigger_keywords or {})
    self._always_include = tuple(name for name in always_include if name)

decide ¶

decide(tools: Sequence[Any]) -> StageOutcome

The decision matrix, synchronously.

Split out because the v7.0.1 entry point (narrow_tools_for_intent) is synchronous and is called from inside a running event loop. One implementation, two callers — the alternative was asyncio.run from a coroutine, which raises, or a second copy of the matrix, which drifts.

Source code in src/symfonic/capabilities/tools/stages/intent.py

def decide(self, tools: Sequence[Any]) -> StageOutcome:
    """The decision matrix, synchronously.

    Split out because the v7.0.1 entry point
    (``narrow_tools_for_intent``) is synchronous and is called from
    inside a running event loop. One implementation, two callers —
    the alternative was ``asyncio.run`` from a coroutine, which
    raises, or a second copy of the matrix, which drifts.
    """
    tools = tuple(tools)
    if not tools:
        return StageOutcome.abstain("no candidate tools")

    verdict = self._verdict
    label = getattr(verdict, "label", None)
    if verdict is None or label == "ambiguous":
        return StageOutcome.abstain("no actionable verdict")

    always = set(self._always_include)

    if label == "knowledge":
        # Unnamed tools survive here exactly as they do on the action
        # path: they cannot be gated by a name-keyed policy, and the
        # v7.0.1 helper kept them.
        return StageOutcome.select(
            filter_by_names(tools, always),
            "knowledge turn: always-include only",
        )

    keep = set(always)
    keep.update(getattr(verdict, "matched_tool_keywords", None) or [])
    for tool in tools:
        name = tool_name(tool)
        if name is None:
            continue
        if not self._trigger_keywords.get(name):
            keep.add(name)

    return StageOutcome.select(
        filter_by_names(tools, keep), f"{label} turn: keyword intersection",
    )

LazyRoutingStage ¶

LazyRoutingStage(*, allowed_names: Sequence[str] | None = None)

Narrow to the tool names procedural routing resolved.

Source code in src/symfonic/capabilities/tools/stages/lazy.py

def __init__(self, *, allowed_names: Sequence[str] | None = None) -> None:
    self._allowed = None if allowed_names is None else tuple(allowed_names)

PolicyStage ¶

PolicyStage(*, catalog: ToolCatalog, max_safety_level: Any = None, budget_usd: float | None = None, include_unknown_cost: bool = True, available_capabilities: set[type] | None = None, agent_visible_only: bool = False, categories: Sequence[Any] | None = None)

Drop tools the configured policy forbids.

A tool absent from the catalogue is kept. Its metadata is unknown, and denying it would make "the registry has never heard of this tool" indistinguishable from "policy forbids this tool" — the first is a wiring bug that has to stay visible.

Source code in src/symfonic/capabilities/tools/stages/policy.py

def __init__(
    self,
    *,
    catalog: ToolCatalog,
    max_safety_level: Any = None,
    budget_usd: float | None = None,
    include_unknown_cost: bool = True,
    available_capabilities: set[type] | None = None,
    agent_visible_only: bool = False,
    categories: Sequence[Any] | None = None,
) -> None:
    self._catalog = catalog
    self._max_safety_rank = _safety_rank(max_safety_level)
    self._budget_usd = budget_usd
    self._include_unknown_cost = include_unknown_cost
    self._available_capabilities = available_capabilities
    self._agent_visible_only = agent_visible_only
    self._categories = (
        None if categories is None else {_category_value(c) for c in categories}
    )

RolePaletteStage ¶

RolePaletteStage(*, resolver: Any = None, role: str = '')

Narrow to the palette a role resolver returns.

Three answers, three meanings:

None — no policy. Abstain (the safe default for an empty role_tools map or an unmapped role).
a non-empty list — the palette. It can only narrow; a resolver returning tools that were not offered is not honoured, because the framework constrains the choice set and never authors it.
[] — abstain with a warning. The Protocol has always documented that the consumer "will detect this and abstain to all_tools with a WARN"; before T3.1.2 nothing did, and an empty palette reached _bind_tools, which binds nothing for an empty list — silently dropping every tool AND any forced tool_choice.

Source code in src/symfonic/capabilities/tools/stages/role.py

def __init__(self, *, resolver: Any = None, role: str = "") -> None:
    self._resolver = resolver
    self._role = role

RoutedTool `dataclass` ¶

RoutedTool(content: str, metadata: Mapping[str, Any] = dict())

One routing row, in the shape :mod:..routing reads.

content and metadata['action_tool'] carry the same name because routing prefers the second and falls back to the first; filling both means this row survives either rule rather than depending on which one is current.

SelectionContext `dataclass` ¶

SelectionContext(state: MutableMapping[str, Any] = dict(), messages: Sequence[Any] = (), role: str = '', query: str = '')

What the stages know about the turn being selected for.

state is passed through to adopter-supplied resolvers by identity, not by copy: the pre-T3.1.2 react node handed them the live LangGraph state and a resolver that reads a tenant key out of it must keep working.

forced_choice is the one mutable field. The pipeline writes it when a stage resolves a force, and later stages (and the pipeline's own protection rule) read it. It is deliberately not an input a caller pre-seeds — the forced-choice stage owns that question, and a second writer would recreate the "two answers to one question" bug class the plan model exists to remove.

resolved_messages ¶

resolved_messages() -> Sequence[Any]

messages if supplied, else the conversation on state.

Source code in src/symfonic/capabilities/tools/context.py

def resolved_messages(self) -> Sequence[Any]:
    """``messages`` if supplied, else the conversation on ``state``."""
    if self.messages:
        return self.messages
    return self.state.get("messages") or ()

SelectionResult `dataclass` ¶

SelectionResult(tools: tuple[Any, ...] = (), forced_choice: str | None = None, trace: tuple[StageRecord, ...] = ())

The tools to bind, the choice to force, and why.

SelectionStage ¶

Bases: Protocol

One narrowing (or widening) lever in the selection pipeline.

name labels the stage in the decision trace.

protects_forced declares whether the pipeline reinstates a forced tool this stage dropped. It is True for stages that run after force resolution and express a preference (a role palette, a policy ceiling); a stage that must be able to drop a forced tool — a hard security filter, say — sets it False and says so.

apply never raises for a caller's benefit: the pipeline contains failures either way, but a stage that can degrade meaningfully should return StageOutcome.abstain(degraded=True) and log why.

StageOutcome `dataclass` ¶

StageOutcome(tools: tuple[Any, ...] | None = None, forced_choice: str | None = None, sets_forced_choice: bool = False, degraded: bool = False, detail: str = '')

One stage's answer.

tools is None means abstain — "no policy here, leave the set alone" — and is distinct from an empty tuple, which means "policy applied and nothing survived". Collapsing the two is how a resolver that could not decide ends up handing the model zero tools.

force `classmethod` ¶

force(choice: str | None, detail: str = '', *, degraded: bool = False) -> StageOutcome

Record a forced tool choice without touching the candidate set.

Source code in src/symfonic/capabilities/tools/values.py

@classmethod
def force(
    cls, choice: str | None, detail: str = "", *, degraded: bool = False,
) -> StageOutcome:
    """Record a forced tool choice without touching the candidate set."""
    return cls(
        tools=None,
        forced_choice=choice,
        sets_forced_choice=True,
        degraded=degraded,
        detail=detail,
    )

StageRecord `dataclass` ¶

StageRecord(stage: str, action: str, kept: tuple[str, ...] = (), dropped: tuple[str, ...] = (), added: tuple[str, ...] = (), reinstated: tuple[str, ...] = (), forced_choice: str | None = None, detail: str = '')

What one stage did, in names, for the decision trace.