Skip to content

symfonic.core.tools

tools

symfonic.core.tools — Tool registry, metadata, and the @symfonic_tool decorator.

SafetyLevel

Bases: str, Enum

Safety classification for tools.

SymfonicToolMetadata dataclass

SymfonicToolMetadata(
    scope: ScopeLiteral = "public",
    routing_mode: RoutingMode = "observe",
    visibility: Visibility = "visible",
)

Symfonic-specific tool metadata.

Attached to the decorated tool as _symfonic_metadata. The registry reads this attribute, raises SymfonicAgentError if the experimental flag is off, and otherwise lifts the metadata into its own :class:~symfonic.core.tools.registry.ToolRegistration fields (visibility="hidden" -> visible_to_agents=False).

Future Symfonic versions may add fields here; consumers should treat the object as read-only and prefer :func:get_symfonic_metadata over poking the attribute directly.

ToolCategory

Bases: str, Enum

Categories for tool classification.

ToolRegistration dataclass

ToolRegistration(
    tool: BaseTool,
    category: ToolCategory = ToolCategory.CUSTOM,
    safety_level: SafetyLevel = SafetyLevel.RESTRICTED,
    max_calls_per_invocation: int | None = None,
    visible_to_agents: bool = True,
    cost_usd: float | None = None,
    requires: type | None = None,
)

Metadata for a registered tool.

ToolRegistry

ToolRegistry()

Tool registration, filtering, cost estimation, and freeze.

Frozen automatically by AgentGraph.compile(). After freeze, register() raises RuntimeError.

Source code in src/symfonic/core/tools/registry.py
def __init__(self) -> None:
    self._registrations: dict[str, ToolRegistration] = {}
    self._frozen: bool = False

all_tools

all_tools() -> list[BaseTool]

Return all registered tools.

Source code in src/symfonic/core/tools/registry.py
def all_tools(self) -> list[BaseTool]:
    """Return all registered tools."""
    return [r.tool for r in self._registrations.values()]

cost_usd_for_invocation

cost_usd_for_invocation(
    tool_names: list[str],
) -> float | None

Calculate total cost for a set of tool invocations.

Returns None if any tool has unknown cost.

Source code in src/symfonic/core/tools/registry.py
def cost_usd_for_invocation(self, tool_names: list[str]) -> float | None:
    """Calculate total cost for a set of tool invocations.

    Returns None if any tool has unknown cost.
    """
    total = 0.0
    for name in tool_names:
        reg = self._registrations.get(name)
        if reg is None or reg.cost_usd is None:
            return None
        total += reg.cost_usd
    return total

filter

filter(
    allowed_names: list[str] | None = None,
    categories: list[ToolCategory] | None = None,
    max_safety_level: SafetyLevel = SafetyLevel.DANGEROUS,
    agent_visible_only: bool = False,
    available_capabilities: set[type] | None = None,
    debug: bool = False,
) -> list[BaseTool]

Filter tools by criteria.

When debug=True, each excluded tool emits a logger.debug() message.

Source code in src/symfonic/core/tools/registry.py
def filter(
    self,
    allowed_names: list[str] | None = None,
    categories: list[ToolCategory] | None = None,
    max_safety_level: SafetyLevel = SafetyLevel.DANGEROUS,
    agent_visible_only: bool = False,
    available_capabilities: set[type] | None = None,
    debug: bool = False,
) -> list[BaseTool]:
    """Filter tools by criteria.

    When debug=True, each excluded tool emits a logger.debug() message.
    """
    result: list[BaseTool] = []
    max_order = _SAFETY_ORDER[max_safety_level]

    for name, reg in self._registrations.items():
        # Name filter
        if allowed_names is not None and name not in allowed_names:
            if debug:
                logger.debug("Tool '%s' excluded — not in allowed_names", name)
            continue

        # Category filter
        if categories is not None and reg.category not in categories:
            if debug:
                logger.debug(
                    "Tool '%s' excluded — category %s not in filter",
                    name, reg.category,
                )
            continue

        # Safety filter
        if _SAFETY_ORDER[reg.safety_level] > max_order:
            if debug:
                logger.debug(
                    "Tool '%s' excluded — safety level %s exceeds max",
                    name, reg.safety_level,
                )
            continue

        # Agent visibility filter
        if agent_visible_only and not reg.visible_to_agents:
            if debug:
                logger.debug("Tool '%s' excluded — not visible to agents", name)
            continue

        # Capability filter
        if (
            reg.requires is not None
            and available_capabilities is not None
            and reg.requires not in available_capabilities
        ):
            if debug:
                logger.debug(
                    "Tool '%s' excluded — missing capability %s",
                    name,
                    reg.requires.__name__,
                )
            continue

        result.append(reg.tool)

    return result

freeze

freeze() -> None

Freeze the registry. Called automatically by AgentGraph.compile().

Source code in src/symfonic/core/tools/registry.py
def freeze(self) -> None:
    """Freeze the registry. Called automatically by AgentGraph.compile()."""
    self._frozen = True

get_registration

get_registration(tool_name: str) -> ToolRegistration | None

Get registration metadata for a tool by name.

Source code in src/symfonic/core/tools/registry.py
def get_registration(self, tool_name: str) -> ToolRegistration | None:
    """Get registration metadata for a tool by name."""
    return self._registrations.get(tool_name)

register

register(tool: BaseTool, **metadata: Any) -> ToolRegistry

Register a single tool with optional metadata. Returns self for chaining.

Raises SymfonicAgentError(code='bad_request') if tool was produced by @symfonic_tool and the experimental flag is off.

Source code in src/symfonic/core/tools/registry.py
def register(self, tool: BaseTool, **metadata: Any) -> ToolRegistry:
    """Register a single tool with optional metadata. Returns self for chaining.

    Raises ``SymfonicAgentError(code='bad_request')`` if ``tool`` was
    produced by ``@symfonic_tool`` and the experimental flag is off.
    """
    self._assert_mutable()
    lifted = self._apply_symfonic_metadata(tool, metadata)
    reg = ToolRegistration(tool=tool, **lifted)
    self._registrations[tool.name] = reg
    return self

register_many

register_many(tools: list[BaseTool]) -> ToolRegistry

Register multiple tools with default metadata. Returns self for chaining.

Source code in src/symfonic/core/tools/registry.py
def register_many(self, tools: list[BaseTool]) -> ToolRegistry:
    """Register multiple tools with default metadata. Returns self for chaining."""
    for tool in tools:
        self.register(tool)
    return self

set_experimental_tool_decorator

set_experimental_tool_decorator(enabled: bool) -> None

Deprecated no-op. @symfonic_tool is GA and needs no gate.

Retained so existing engine wiring and adopter code that toggled the former experimental gate keep working. The enabled value is ignored — decorator-produced tools always register.

Source code in src/symfonic/core/tools/registry.py
def set_experimental_tool_decorator(self, enabled: bool) -> None:
    """Deprecated no-op. ``@symfonic_tool`` is GA and needs no gate.

    Retained so existing engine wiring and adopter code that toggled
    the former experimental gate keep working. The ``enabled`` value
    is ignored — decorator-produced tools always register.
    """
    return None

tools_within_budget

tools_within_budget(
    budget_usd: float, include_unknown_cost: bool = True
) -> list[BaseTool]

Return tools whose cost is within budget.

If include_unknown_cost is True, tools with no cost info are included.

Source code in src/symfonic/core/tools/registry.py
def tools_within_budget(
    self, budget_usd: float, include_unknown_cost: bool = True
) -> list[BaseTool]:
    """Return tools whose cost is within budget.

    If include_unknown_cost is True, tools with no cost info are included.
    """
    result: list[BaseTool] = []
    for reg in self._registrations.values():
        if reg.cost_usd is None:
            if include_unknown_cost:
                result.append(reg.tool)
        elif reg.cost_usd <= budget_usd:
            result.append(reg.tool)
    return result

get_symfonic_metadata

get_symfonic_metadata(
    obj: Any,
) -> SymfonicToolMetadata | None

Return the :class:SymfonicToolMetadata attached to obj, or None.

Source code in src/symfonic/core/tools/decorator.py
def get_symfonic_metadata(obj: Any) -> SymfonicToolMetadata | None:
    """Return the :class:`SymfonicToolMetadata` attached to ``obj``, or None."""
    meta = getattr(obj, _METADATA_ATTR, None)
    return meta if isinstance(meta, SymfonicToolMetadata) else None

is_symfonic_tool

is_symfonic_tool(obj: Any) -> bool

Return True if obj was produced by :func:symfonic_tool.

Source code in src/symfonic/core/tools/decorator.py
def is_symfonic_tool(obj: Any) -> bool:
    """Return ``True`` if ``obj`` was produced by :func:`symfonic_tool`."""
    return isinstance(getattr(obj, _METADATA_ATTR, None), SymfonicToolMetadata)

symfonic_tool

symfonic_tool(
    *,
    scope: ScopeLiteral = "public",
    routing_mode: RoutingMode = "observe",
    visibility: Visibility = "visible",
    name: str | None = None,
    description: str | None = None,
) -> Callable[[Callable[..., Any]], StructuredTool]

Decorate a Python callable into a symfonic-compatible tool.

The resulting object is a :class:~langchain_core.tools.StructuredTool accepted by :meth:~symfonic.core.tools.registry.ToolRegistry.register without any further plumbing.

Parameters

scope: Logical visibility scope. "public" (default), "internal", or "admin". Symfonic does not yet enforce scope at the invocation tier; the field is captured today so policy code can consume it without a future signature change. routing_mode: Per-tool override for the framework's dynamic tool-routing verdict. Defaults to "observe"; "enforce" causes the router to actually narrow the manifest, "off" opts out of routing entirely. visibility: "visible" (default) keeps the tool in the agent-visible manifest; "hidden" flips the registry's visible_to_agents field to False so the tool is bound to the LangGraph runtime but never offered to the LLM in the manifest. Useful for system-level tools (telemetry emitters, feature-flag readers) that should be callable by orchestration code but never selected by the model. name: Override the tool name. Defaults to fn.__name__. description: Override the description. Defaults to the function's docstring.

Returns

Callable A decorator that, applied to a callable, returns a :class:StructuredTool with the symfonic metadata attached.

Notes
  • The decorated callable's async / sync / async-generator semantics are preserved on the underlying tool's func / coroutine slots. inspect.isasyncgenfunction continues to return True for generator tools so Phase 3's :func:~symfonic.core.tools.streaming.wrap_async_gen_tool works unchanged.
  • Registration requires no flag (GA). The resulting tool registers through ToolRegistry.register like any LangChain tool; its scope / routing_mode / visibility metadata is lifted into the ToolRegistration automatically.
Source code in src/symfonic/core/tools/decorator.py
def symfonic_tool(
    *,
    scope: ScopeLiteral = "public",
    routing_mode: RoutingMode = "observe",
    visibility: Visibility = "visible",
    name: str | None = None,
    description: str | None = None,
) -> Callable[[Callable[..., Any]], StructuredTool]:
    """Decorate a Python callable into a symfonic-compatible tool.

    The resulting object is a :class:`~langchain_core.tools.StructuredTool`
    accepted by :meth:`~symfonic.core.tools.registry.ToolRegistry.register`
    without any further plumbing.

    Parameters
    ----------
    scope:
        Logical visibility scope.  ``"public"`` (default), ``"internal"``,
        or ``"admin"``.  Symfonic does not yet enforce ``scope`` at the
        invocation tier; the field is captured today so policy code can
        consume it without a future signature change.
    routing_mode:
        Per-tool override for the framework's dynamic tool-routing
        verdict.  Defaults to ``"observe"``; ``"enforce"`` causes the
        router to actually narrow the manifest, ``"off"`` opts out of
        routing entirely.
    visibility:
        ``"visible"`` (default) keeps the tool in the agent-visible
        manifest; ``"hidden"`` flips the registry's
        ``visible_to_agents`` field to ``False`` so the tool is bound
        to the LangGraph runtime but never offered to the LLM in the
        manifest.  Useful for system-level tools (telemetry emitters,
        feature-flag readers) that should be callable by orchestration
        code but never selected by the model.
    name:
        Override the tool name.  Defaults to ``fn.__name__``.
    description:
        Override the description.  Defaults to the function's docstring.

    Returns
    -------
    Callable
        A decorator that, applied to a callable, returns a
        :class:`StructuredTool` with the symfonic metadata attached.

    Notes
    -----
    * The decorated callable's async / sync / async-generator
      semantics are preserved on the underlying tool's ``func`` /
      ``coroutine`` slots.  ``inspect.isasyncgenfunction`` continues
      to return ``True`` for generator tools so Phase 3's
      :func:`~symfonic.core.tools.streaming.wrap_async_gen_tool` works
      unchanged.
    * Registration requires no flag (GA). The resulting tool registers
      through ``ToolRegistry.register`` like any LangChain tool; its
      ``scope`` / ``routing_mode`` / ``visibility`` metadata is lifted
      into the ``ToolRegistration`` automatically.
    """

    metadata = SymfonicToolMetadata(
        scope=scope,
        routing_mode=routing_mode,
        visibility=visibility,
    )

    def decorator(fn: Callable[..., Any]) -> StructuredTool:
        if not callable(fn):
            raise TypeError(
                "@symfonic_tool can only decorate callables, got "
                f"{type(fn).__name__!r}"
            )

        tool_name = name or fn.__name__
        tool_desc = description if description is not None else (
            inspect.getdoc(fn) or f"Symfonic tool '{tool_name}'."
        )

        # parse_docstring=True so ``Args:`` blocks surface as field
        # descriptions; error_on_invalid_docstring=False keeps the
        # decorator forgiving for callers who do not follow Google-style
        # docstrings.
        builder_kwargs: dict[str, Any] = {
            "name": tool_name,
            "description": tool_desc,
            "parse_docstring": True,
            "error_on_invalid_docstring": False,
        }

        if inspect.iscoroutinefunction(fn) or inspect.isasyncgenfunction(fn):
            # StructuredTool stores async impls on .coroutine.  Pass
            # ``func=None`` so the builder does not try to use the
            # async function as a sync one.
            built = StructuredTool.from_function(
                func=None,
                coroutine=fn,
                **builder_kwargs,
            )
        else:
            built = StructuredTool.from_function(func=fn, **builder_kwargs)

        # Attach metadata.  StructuredTool is a Pydantic model with
        # extra="forbid" on its model_config, so we use object.__setattr__
        # to bypass the validation and stash the attribute on the
        # instance (mirrors how LangChain itself attaches private state).
        object.__setattr__(built, _METADATA_ATTR, metadata)

        return built

    return decorator