symfonic.agent.backend.tools

tools

Tool normalization and execution behind the kernel's tool port (FAC-6, RES-3).

NormalizedTools dataclass

NormalizedTools(tools: tuple[BaseTool, ...] = (), by_name: dict[str, BaseTool] | None = None)

The construction-time-resolved tool set.

Frozen and resolved once, at construction: FAC-6 fixes the tool set for the life of the agent. There is no per-invocation override and no mutable registry, because the invocation plan is compiled per call and a tool set that can change underneath it is precisely the drift the plan model exists to remove.

ToolAdapter

ToolAdapter(tools: NormalizedTools)

The kernel's ToolPort, bound to this agent's fixed tool set.

It is built from the same normalized set that produced plan group G4, which is why the loop performs no second allowlist check: one manifest, one answer to "may this call run" (IPL-5).

Run-unique call ids are not minted here. The kernel owns them, because uniqueness is a property of the run and this object only ever sees one call at a time.

Source code in src/symfonic/agent/backend/tools.py

def __init__(self, tools: NormalizedTools) -> None:
    self._tools = tools

refuse

refuse(request: ToolRequest, reason: str) -> ToolCall

The outcome of a call a precondition would not let run.

The port builds it rather than the kernel, for the reason the kernel carries outcomes opaquely at all: this is the facade's type, and a kernel that constructed one would be deciding what a tool result looks like. Same shape as an unknown tool and a raising tool -- the error text is what the model is shown as the observation, which is how a tool-using loop is told no and can try something else.

duration_ms is zero because nothing ran. Not "very fast": a reader totalling tool time must not count a call that never happened.

Source code in src/symfonic/agent/backend/tools.py

def refuse(self, request: ToolRequest, reason: str) -> ToolCall:
    """The outcome of a call a precondition would not let run.

    The port builds it rather than the kernel, for the reason the kernel
    carries outcomes opaquely at all: this is the facade's type, and a
    kernel that constructed one would be deciding what a tool result looks
    like. Same shape as an unknown tool and a raising tool -- the error
    text is what the model is shown as the observation, which is how a
    tool-using loop is told no and can try something else.

    ``duration_ms`` is zero because nothing ran. Not "very fast": a reader
    totalling tool time must not count a call that never happened.
    """
    return ToolCall(
        id=request.call_id,
        name=request.name,
        arguments=dict(request.arguments),
        error=reason or "refused by a tool precondition",
        duration_ms=0.0,
    )

execute_tool_call async

execute_tool_call(tools: NormalizedTools, call_id: str, name: str, arguments: dict[str, Any]) -> ToolCall

Run one tool call and return its completed :class:ToolCall record.

A tool that raises is reported, never hidden: the exception text lands on ToolCall.error and is fed back to the model as the tool's observation, which is how a tool-using loop recovers. BaseException (and therefore asyncio.CancelledError) is deliberately not caught — cancellation must propagate unchanged (EVT-9).

Source code in src/symfonic/agent/backend/tools.py

async def execute_tool_call(
    tools: NormalizedTools,
    call_id: str,
    name: str,
    arguments: dict[str, Any],
) -> ToolCall:
    """Run one tool call and return its completed :class:`ToolCall` record.

    A tool that raises is reported, never hidden: the exception text lands on
    ``ToolCall.error`` and is fed back to the model as the tool's observation,
    which is how a tool-using loop recovers. ``BaseException`` (and therefore
    ``asyncio.CancelledError``) is deliberately *not* caught — cancellation
    must propagate unchanged (EVT-9).
    """
    started = time.perf_counter()
    tool = tools.lookup(name)

    if tool is None:
        return ToolCall(
            id=call_id,
            name=name,
            arguments=arguments,
            error=f"unknown tool {name!r}",
            duration_ms=_elapsed_ms(started),
        )

    try:
        output = await _invoke(tool, arguments)
    except Exception as exc:  # noqa: BLE001 — reported on ToolCall.error
        return ToolCall(
            id=call_id,
            name=name,
            arguments=arguments,
            error=f"{type(exc).__name__}: {exc}",
            duration_ms=_elapsed_ms(started),
        )

    return ToolCall(
        id=call_id,
        name=name,
        arguments=arguments,
        result=output if isinstance(output, str) else str(output),
        duration_ms=_elapsed_ms(started),
    )

merge_capability_tools

merge_capability_tools(adopter: Sequence[Any], contributed: Sequence[tuple[str, Any]]) -> NormalizedTools

Normalise the adopter's tools and the capabilities' into one set.

One set, because the tool path is read in four places — bind_tools, the G4 manifest, ToolAdapter, and the tool_call grant — and a second collection that only some of them consult is exactly how a tool becomes bindable and not callable.

contributed is (capability, tool) pairs, which is why this exists rather than a bare normalize_tools(list(a) + list(b)). Every refusal below names the capability. A collision reported as "tools[4] resolves to the name 'run_agent', which is already registered by an earlier tool" is true and tells the adopter nothing about which capability to configure — and unlike tools=[...], the adopter did not write the offending list and cannot see it.

The adopter's tools go first and keep the plain index in their messages, so an error in a hand-written list reads exactly as it did before capabilities could contribute anything.

Raises:

Type	Description
ConfigurationError	for a non-tool object, or for any name collision — capability against adopter, or capability against capability.

Source code in src/symfonic/agent/backend/tools.py

def merge_capability_tools(
    adopter: Sequence[Any],
    contributed: Sequence[tuple[str, Any]],
) -> NormalizedTools:
    """Normalise the adopter's tools and the capabilities' into one set.

    One set, because the tool path is read in four places — ``bind_tools``, the
    G4 manifest, ``ToolAdapter``, and the ``tool_call`` grant — and a second
    collection that only some of them consult is exactly how a tool becomes
    bindable and not callable.

    ``contributed`` is ``(capability, tool)`` pairs, which is why this exists
    rather than a bare ``normalize_tools(list(a) + list(b))``. Every refusal
    below names the capability. A collision reported as "tools[4] resolves to
    the name 'run_agent', which is already registered by an earlier tool" is
    true and tells the adopter nothing about which capability to configure —
    and unlike ``tools=[...]``, the adopter did not write the offending list
    and cannot see it.

    The adopter's tools go first and keep the plain index in their messages, so
    an error in a hand-written list reads exactly as it did before capabilities
    could contribute anything.

    Raises:
        ConfigurationError: for a non-tool object, or for any name collision —
            capability against adopter, or capability against capability.
    """
    merged = normalize_tools(adopter)
    if not contributed:
        return merged

    # Local: ``agent.cutover`` imports this module's package, so a top-level
    # import closes a cycle. It is also the cheaper placement -- an agent with
    # no contributed tools never reaches this line.
    from symfonic.agent.cutover.delegation import bind_contributed_tool

    resolved = list(merged.tools)
    by_name = dict(merged.by_name or {})
    owner: dict[str, str] = {}

    for capability, candidate in contributed:
        try:
            # A capability contributes a *description* -- a name, a
            # description, a coroutine -- rather than a runtime tool object,
            # so that a capability package does not carry the tool library on
            # its import path. Wrapping it in the type the runtime binds is
            # the composition root's job, and for a kernel agent this is that
            # step. Anything already bound passes through untouched.
            tool = adapt_tool(bind_contributed_tool(candidate), len(resolved))
        except ConfigurationError as exc:
            # Not ``_adapt``'s message with a prefix. That message is written
            # for a hand-written list -- it says ``tools[4]``, an index into a
            # merged sequence the adopter never wrote and cannot inspect. What
            # the adopter can act on is the capability's name and the name the
            # thing claimed to have.
            offered = getattr(candidate, "name", None)
            raise ConfigurationError(
                f"capability {capability!r} contributed "
                + (f"{offered!r}, which is " if offered else "")
                + f"not a tool: a {type(candidate).__name__} is neither a "
                "@symfonic_tool object, a LangChain BaseTool, nor a plain "
                "annotated callable. A capability offering a tool has to offer "
                "something executable: a name and a description reach the "
                "manifest and the model, and then the executor has nothing to "
                "call."
            ) from exc
        if tool.name in by_name:
            held = owner.get(tool.name)
            raise ConfigurationError(
                f"capability {capability!r} contributed a tool named "
                f"{tool.name!r}, which is already registered by "
                + (f"capability {held!r}." if held else "a tool passed to Agent(tools=[...]).")
                + " Two tools cannot share a name: the manifest is keyed by it "
                "and the executor looks calls up by it, so one of them would be "
                "bound and never reachable."
            )
        by_name[tool.name] = tool
        owner[tool.name] = capability
        resolved.append(tool)

    return NormalizedTools(tools=tuple(resolved), by_name=by_name)

normalize_tools

normalize_tools(tools: Sequence[Any]) -> NormalizedTools

Accept the three documented input forms; reject everything else.

Accepted (FAC-6): an object produced by @symfonic_tool(...), any LangChain BaseTool, or a plain Python callable with annotated parameters and a docstring — adapted through symfonic_tool() with its documented defaults.

Raises:

Type	Description
ConfigurationError	for a non-tool object (naming it and its index), or for two tools resolving to the same name. Last-one-wins would silently drop a tool the adopter believes is registered, so the collision is a hard error rather than a warning.

Source code in src/symfonic/agent/backend/tools.py

def normalize_tools(tools: Sequence[Any]) -> NormalizedTools:
    """Accept the three documented input forms; reject everything else.

    Accepted (FAC-6): an object produced by ``@symfonic_tool(...)``, any
    LangChain ``BaseTool``, or a plain Python callable with annotated
    parameters and a docstring — adapted through ``symfonic_tool()`` with its
    documented defaults.

    Raises:
        ConfigurationError: for a non-tool object (naming it and its index),
            or for two tools resolving to the same name. Last-one-wins would
            silently drop a tool the adopter believes is registered, so the
            collision is a hard error rather than a warning.
    """
    resolved: list[BaseTool] = []
    by_name: dict[str, BaseTool] = {}

    for index, candidate in enumerate(tools):
        tool = adapt_tool(candidate, index)
        if tool.name in by_name:
            raise ConfigurationError(
                f"tools[{index}] resolves to the name {tool.name!r}, which is "
                "already registered by an earlier tool. Two tools cannot share "
                "a name; rename one with @symfonic_tool(name=...)."
            )
        by_name[tool.name] = tool
        resolved.append(tool)

    return NormalizedTools(tools=tuple(resolved), by_name=by_name)

remap_tool_call_ids

remap_tool_call_ids(message: Any, call_ids: Sequence[str]) -> Any

Return an assistant wire message whose tool ids match facade ids.

Source code in src/symfonic/agent/backend/tool_ids.py

def remap_tool_call_ids(message: Any, call_ids: Sequence[str]) -> Any:
    """Return an assistant wire message whose tool ids match facade ids."""
    tool_calls = list(getattr(message, "tool_calls", None) or [])
    if [call.get("id") for call in tool_calls] == list(call_ids):
        return message

    updates: dict[str, Any] = {
        "tool_calls": [
            {**call, "id": call_id} for call, call_id in zip(tool_calls, call_ids, strict=True)
        ]
    }

    content = getattr(message, "content", None)
    if isinstance(content, list):
        remapped_content = []
        tool_index = 0
        for block in content:
            if isinstance(block, dict) and block.get("type") == "tool_use":
                block = {**block, "id": call_ids[tool_index]}
                tool_index += 1
            remapped_content.append(block)
        updates["content"] = remapped_content

    chunks = getattr(message, "tool_call_chunks", None)
    if chunks:
        updates["tool_call_chunks"] = [
            {**chunk, "id": call_ids[index]} for index, chunk in enumerate(chunks)
        ]

    return message.model_copy(update=updates)