tool_agent

Level 2 · Tools and Skills — capability-based tool filtering with ToolRegistry.

A tool can declare the dependencies it needs to function via a requires list. The ToolRegistry then exposes only the tools whose requirements are actually satisfied by the current deps. This example registers two tools with different requirements, wires only one of the two backends, and shows the registry filtering the unsatisfiable tool out of the active set.

• Lines: ~49
• Prerequisites: none
• Key concepts: ToolRegistry, capability filtering, the requires declaration

Run it

Installed via pip? Copy this example into your project with the CLI (no checkout needed):

pip install "symfonic-core[cli]"
symfonic examples add tool_agent
python -m tool_agent

Or, from a source checkout (run from the repo root):

python -m examples.tool_agent

Expected output:

Active: ['doc_read']
All: ['doc_read', 'skill_list']
Excluded: ['skill_list']

Full code

"""Tool agent -- what the deployment owns, and what a turn is offered.

Usage: python -m examples.tool_agent

Public surface used, and nothing else::

    from symfonic import Agent
    from symfonic.capabilities.tools import ToolsCapability, keyword_router, tool_name
    from symfonic.core.tools import symfonic_tool
    from symfonic.platform import GrantEffects

The previous version of this file registered two tools in a ``ToolRegistry``,
asked which were active given the dependencies it had wired, and printed
three lists. It never ran an agent, so "excluded" meant "absent from a list"
rather than "the model could not call it".

The lesson survives and gets sharper. There are two narrowings, not one:

* **What the deployment can serve.** A tool whose backing store is not wired
  is not registered, so it never enters the palette on any turn. That is a
  deployment fact and it is decided once.
* **What this turn is offered.** ``ToolsCapability`` routes the registration
  per turn, so a question about stock does not carry the refund tool.

The second is visible only from what the provider was *bound*, which is why
the provider below records it. A tool registered and never bound is one the
model cannot call, and from the outside that is indistinguishable from a
model that chose not to.

``keyword_router`` refuses at composition a keyword naming a tool outside
``registered`` -- so pointing the router at a tool this deployment cannot
serve is a composition error, not a router that silently never matches.
"""

from __future__ import annotations

import asyncio
from typing import Any

from symfonic import Agent
from symfonic.capabilities.tools import ToolsCapability, keyword_router, tool_name
from symfonic.core.tools import symfonic_tool
from symfonic.platform import GrantEffects

DOCUMENTS = {"handbook": "Proof sourdough 4-6 hours at 24C."}


@symfonic_tool(name="doc_read")
async def doc_read(title: str) -> str:
    """Read a document from the handbook store."""
    return DOCUMENTS.get(title, f"no document called {title!r}")


@symfonic_tool(name="stock_count")
async def stock_count(sku: str) -> str:
    """Count units of a SKU in the inventory store."""
    return f"{sku}: 12 units"


@symfonic_tool(name="skill_list")
async def skill_list() -> str:
    """List the procedural skills this deployment has learned."""
    return "not reachable in this example"


class RecordingProvider:
    """Keeps the palettes it was bound and the prompts it was sent."""

    _symfonic_provider_family = "unknown"
    _symfonic_default_model = None

    def __init__(self, *script: Any) -> None:
        self.bound: list[tuple[str, ...]] = []
        self.prompts: list[str] = []
        self._script = list(script)

    def get_chat_model(self, config: Any) -> Any:  # noqa: ARG002
        from langchain_core.messages import AIMessage
        from langchain_core.outputs import ChatGeneration, ChatResult

        from symfonic.core.testing import MockChatModel

        bound, prompts, script = self.bound, self.prompts, self._script
        rounds = {"n": 0}

        class _Model(MockChatModel):
            def bind_tools(self, tools: Any, **kwargs: Any) -> Any:
                bound.append(tuple(sorted(getattr(t, "name", str(t)) for t in tools)))
                return super().bind_tools(tools, **kwargs)

            def _generate(self, messages, stop=None, run_manager=None, **kwargs):  # noqa: ANN001, ANN202
                prompts.append("\n".join(str(m.content) for m in messages))
                entry = script[min(rounds["n"], len(script) - 1)]
                rounds["n"] += 1
                message = AIMessage(content="" if isinstance(entry, list) else entry)
                if isinstance(entry, list):
                    message.tool_calls = entry  # type: ignore[attr-defined]
                return ChatResult(generations=[ChatGeneration(message=message)])

            async def _agenerate(self, messages, stop=None, run_manager=None, **kw):  # noqa: ANN001, ANN202
                return self._generate(messages, stop, **kw)

        return _Model()

    def supports_thinking(self) -> bool:
        return False

    def supports_streaming(self) -> bool:
        return True

    def supports_forced_tool_choice(self, config: Any) -> bool:  # noqa: ARG002
        return True


#: This deployment wired a document store and no skill store, so it can serve
#: one of the two tools it knows how to write. The other is not registered:
#: a tool the deployment cannot back is not a tool it can offer.
SERVABLE = [doc_read, stock_count]
KEYWORDS = {
    "doc_read": ["document", "handbook", "proof", "read"],
    "stock_count": ["stock", "units", "inventory", "sku"],
}


def build(provider: RecordingProvider) -> Agent:
    names = tuple(tool_name(tool) for tool in SERVABLE)
    return Agent(
        provider,
        instructions="Answer from the handbook.",
        tools=SERVABLE,
        capabilities=[
            # The routing stage reads the turn to pick a palette, so it
            # declares a memory-read effect -- and a capability may not grant
            # itself one. The authorization comes from here.
            GrantEffects("memory-read"),
            ToolsCapability(
                entries_for=keyword_router(KEYWORDS, registered=names),
                registered=names,
            ),
        ],
    )


async def main() -> None:
    provider = RecordingProvider(
        [{"name": "doc_read", "args": {"title": "handbook"}, "id": "c1"}],
        "Four to six hours at 24C.",
    )
    result = await build(provider).run("What does the handbook say about proofing?")

    print("== what this deployment can serve ==")
    print("written    : doc_read, stock_count, skill_list")
    print(f"registered : {', '.join(tool_name(t) for t in SERVABLE)}")
    print("note       : skill_list has no store behind it, so it is not")
    print("             registered. Not offered-and-refused -- absent.")

    print()
    print("== what this turn was offered ==")
    print(f"at compile : {', '.join(provider.bound[0])}")
    print(f"this turn  : {', '.join(provider.bound[-1])}")

    call = next(c for c in (result.tool_calls or []) if c.name == "doc_read")
    print()
    print("== and it ran ==")
    print(f"called     : {call.name}({call.arguments})")
    print(f"returned   : {call.result}")
    print(f"the model read it: {'4-6 hours' in provider.prompts[1]}")
    print(f"answered   : {result.text}")

    print()
    print("== pointing the router at a tool nobody serves ==")
    try:
        keyword_router({"skill_list": ["skills"]}, registered=("doc_read",))
    except ValueError as refused:
        print(f"refused    : {refused}")
    print("note       : at composition, not at the turn. A keyword naming an")
    print("             unregistered tool is a router that would never match,")
    print("             which reads as a policy and behaves like a typo.")


if __name__ == "__main__":
    asyncio.run(main())

Step by step

1. Declare tools with their requirements

class DocTool:
    name = "doc_read"
    description = "Read documents"
    requires: list[type] = [DocumentStore]
    async def __call__(self, state: Any, **kw: Any) -> str:
        return "doc"

A tool is any object exposing name, description, an async __call__, and an optional requires list of the protocols it depends on. DocTool declares it needs a DocumentStore; SkillTool declares it needs a SkillStore. The requires list is the contract the registry uses to decide availability.

This example uses the plain class form to make the requires contract explicit. In application code you would more often use the @symfonic_tool decorator, which derives the schema from your type hints and docstring.

2. Register the tools

registry = ToolRegistry()
registry.register(DocTool())
registry.register(SkillTool())

ToolRegistry is the container. Registering a tool makes it known — not necessarily active. Whether it becomes active depends on step 4.

3. Wire only some dependencies

deps = BaseAgentDeps(ModelProvider=MockModelProvider())
deps.register(DocumentStore, InMemoryDocumentStore())
# note: no SkillStore is registered

We deliberately provide a DocumentStore but not a SkillStore. This is the whole point of the demo: the environment can satisfy DocTool but not SkillTool.

4. Let the registry filter

active = registry.get_active_tools(deps)

get_active_tools(deps) returns only the tools whose every requires entry is present in deps. DocTool passes (its DocumentStore is registered); SkillTool is filtered out (no SkillStore). The model therefore only ever sees tools it can actually call — no runtime "capability missing" surprises.

5. Inspect the split

print(f"Active: {[t.name for t in active]}")           # ['doc_read']
print(f"All: {[t.name for t in registry.all_tools()]}") # ['doc_read', 'skill_list']
excluded = [t for t in registry.all_tools() if t not in active]
print(f"Excluded: {[t.name for t in excluded]}")        # ['skill_list']

all_tools() is everything registered; active is the satisfiable subset. The difference is what got filtered and why.

Why this matters

Capability filtering keeps the model's tool surface honest: a tool that cannot run in the current deployment never appears in the prompt, so the model cannot hallucinate a call to it. Register the same tool set everywhere, and each environment (dev, staging, a specific tenant) automatically exposes only what it can back.

What to try next

• Write tools with the decorator → Writing Tools
• Guard or rewrite tool calls at runtime → Tool-Call Steering
• Manage reusable skills → skill_agent

See also

• Capabilities
• Tool-Call Policy