Refactor tools to use a base class (just as resources do)

KoolADE85 · KoolADE85 · commit 4ea79dea6c32 · 2026-04-16T17:20:05.000-06:00
diff --git a/dash/mcp/primitives/tools/__init__.py b/dash/mcp/primitives/tools/__init__.py
@@ -1,12 +1,4 @@
-"""MCP tool listing and call handling.
-
-Each tool module exports:
-- ``get_tool_names() -> set[str]``
-- ``get_tools() -> list[Tool]``
-- ``call_tool(tool_name, arguments) -> CallToolResult``
-
-The __init__ assembles the list and dispatches calls by name.
-"""
+"""MCP tool listing and call handling."""
 
 from __future__ import annotations
 
@@ -16,26 +8,29 @@
 
 from dash.mcp.types import ToolNotFoundError
 
-from . import tool_get_dash_component as _get_component
-from . import tools_callbacks as _callbacks
+from .base import MCPToolProvider
+from .tool_get_dash_component import GetDashComponentTool
+from .tools_callbacks import CallbackTools
 
-_TOOL_MODULES = [_callbacks, _get_component]
+_TOOL_PROVIDERS: list[type[MCPToolProvider]] = [
+    CallbackTools,
+    GetDashComponentTool,
+]
 
 
 def list_tools() -> ListToolsResult:
     """Build the MCP tools/list response."""
     tools = []
-    for mod in _TOOL_MODULES:
-        tools.extend(mod.get_tools())
+    for provider in _TOOL_PROVIDERS:
+        tools.extend(provider.list_tools())
     return ListToolsResult(tools=tools)
 
 
 def call_tool(tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
-    """Dispatch a tools/call request by tool name."""
-    for mod in _TOOL_MODULES:
-        if tool_name in mod.get_tool_names():
-            result = mod.call_tool(tool_name, arguments)
-            return result
+    """Route a tools/call request by tool name."""
+    for provider in _TOOL_PROVIDERS:
+        if tool_name in provider.get_tool_names():
+            return provider.call_tool(tool_name, arguments)
     raise ToolNotFoundError(
         f"Tool not found: {tool_name}."
         " The app's callbacks may have changed."
diff --git a/dash/mcp/primitives/tools/base.py b/dash/mcp/primitives/tools/base.py
@@ -0,0 +1,28 @@
+"""Base class for MCP tool providers."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.types import CallToolResult, Tool
+
+
+class MCPToolProvider:
+    """A provider of one or more MCP tools.
+
+    Subclasses implement ``list_tools`` to return the tools they provide,
+    ``get_tool_names`` to advertise those names for routing, and
+    ``call_tool`` to execute a tool by name.
+    """
+
+    @classmethod
+    def get_tool_names(cls) -> set[str]:
+        raise NotImplementedError
+
+    @classmethod
+    def list_tools(cls) -> list[Tool]:
+        raise NotImplementedError
+
+    @classmethod
+    def call_tool(cls, tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
+        raise NotImplementedError
diff --git a/dash/mcp/primitives/tools/tool_get_dash_component.py b/dash/mcp/primitives/tools/tool_get_dash_component.py
@@ -13,6 +13,8 @@
 from dash._layout_utils import find_component
 from dash.mcp.types import ComponentPropertyInfo, ComponentQueryResult
 
+from .base import MCPToolProvider
+
 
 class _ComponentQueryInput(TypedDict):
     component_id: Annotated[str, Field(description="The component ID to query")]
@@ -32,92 +34,94 @@ class _ComponentQueryInput(TypedDict):
 NAME = "get_dash_component"
 
 
-def get_tool_names() -> set[str]:
-    return {NAME}
-
-
-def get_tools() -> list[Tool]:
-    return [_build_tool()]
-
-
-def _build_tool() -> Tool:
-    return Tool(
-        name=NAME,
-        description=(
-            "Get a component's properties, values, and tool relationships. "
-            "If property is omitted, returns all defined properties. "
-            "If property is specified, returns only that property. "
-            "See the dash://components resource for available component IDs."
-        ),
-        inputSchema=_INPUT_SCHEMA,
-        outputSchema=_OUTPUT_SCHEMA,
-    )
-
-
-def call_tool(tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
-    comp_id = arguments.get("component_id", "")
-    if not comp_id:
-        raise ValueError("component_id is required")
+class GetDashComponentTool(MCPToolProvider):
+    """Inspects a component's properties and its tool relationships."""
+
+    @classmethod
+    def get_tool_names(cls) -> set[str]:
+        return {NAME}
+
+    @classmethod
+    def list_tools(cls) -> list[Tool]:
+        return [
+            Tool(
+                name=NAME,
+                description=(
+                    "Get a component's properties, values, and tool relationships. "
+                    "If property is omitted, returns all defined properties. "
+                    "If property is specified, returns only that property. "
+                    "See the dash://components resource for available component IDs."
+                ),
+                inputSchema=_INPUT_SCHEMA,
+                outputSchema=_OUTPUT_SCHEMA,
+            )
+        ]
 
-    prop_filter = arguments.get("property", "")
-    component = find_component(comp_id)
+    @classmethod
+    def call_tool(cls, tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
+        comp_id = arguments.get("component_id", "")
+        if not comp_id:
+            raise ValueError("component_id is required")
+
+        prop_filter = arguments.get("property", "")
+        component = find_component(comp_id)
+
+        if component is None:
+            callback_map = get_app().mcp_callback_map
+            rendering_tools = [
+                cb.tool_name
+                for cb in callback_map
+                if any(out["component_id"] == comp_id for out in cb.outputs)
+            ]
+            msg = f"Component '{comp_id}' not found in static layout."
+            if rendering_tools:
+                msg += f" However, the following tools would modify it: {rendering_tools}."
+            msg += " Use the dash://components resource to see statically available component IDs."
+            return CallToolResult(
+                content=[TextContent(type="text", text=msg)],
+                isError=True,
+            )
 
-    if component is None:
         callback_map = get_app().mcp_callback_map
-        rendering_tools = [
-            cb.tool_name
-            for cb in callback_map
-            if any(out["component_id"] == comp_id for out in cb.outputs)
-        ]
-        msg = f"Component '{comp_id}' not found in static layout."
-        if rendering_tools:
-            msg += f" However, the following tools would modify it: {rendering_tools}."
-        msg += " Use the dash://components resource to see statically available component IDs."
-        return CallToolResult(
-            content=[TextContent(type="text", text=msg)],
-            isError=True,
-        )
 
-    callback_map = get_app().mcp_callback_map
-
-    properties: dict[str, ComponentPropertyInfo] = {}
-    for prop_name in getattr(component, "_prop_names", []):
-        if prop_filter and prop_name != prop_filter:
-            continue
-
-        value = callback_map.get_initial_value(f"{comp_id}.{prop_name}")
-        if value is None:
-            value = getattr(component, prop_name, None)
-        if value is None:
-            continue
-
-        modified_by: list[str] = []
-        input_to: list[str] = []
-        id_and_prop = f"{comp_id}.{prop_name}"
-        for cb in callback_map:
-            for out in cb.outputs:
-                if out["id_and_prop"] == id_and_prop:
-                    modified_by.append(cb.tool_name)
-            for inp in cb.inputs:
-                if inp["id_and_prop"] == id_and_prop:
-                    input_to.append(cb.tool_name)
-
-        properties[prop_name] = ComponentPropertyInfo(
-            initial_value=value,
-            modified_by_tool=modified_by,
-            input_to_tool=input_to,
+        properties: dict[str, ComponentPropertyInfo] = {}
+        for prop_name in getattr(component, "_prop_names", []):
+            if prop_filter and prop_name != prop_filter:
+                continue
+
+            value = callback_map.get_initial_value(f"{comp_id}.{prop_name}")
+            if value is None:
+                value = getattr(component, prop_name, None)
+            if value is None:
+                continue
+
+            modified_by: list[str] = []
+            input_to: list[str] = []
+            id_and_prop = f"{comp_id}.{prop_name}"
+            for cb in callback_map:
+                for out in cb.outputs:
+                    if out["id_and_prop"] == id_and_prop:
+                        modified_by.append(cb.tool_name)
+                for inp in cb.inputs:
+                    if inp["id_and_prop"] == id_and_prop:
+                        input_to.append(cb.tool_name)
+
+            properties[prop_name] = ComponentPropertyInfo(
+                initial_value=value,
+                modified_by_tool=modified_by,
+                input_to_tool=input_to,
+            )
+
+        labels = callback_map.component_label_map.get(comp_id, [])
+
+        structured: ComponentQueryResult = ComponentQueryResult(
+            component_id=comp_id,
+            component_type=type(component).__name__,
+            label=labels if labels else None,
+            properties=properties,
         )
 
-    labels = callback_map.component_label_map.get(comp_id, [])
-
-    structured: ComponentQueryResult = ComponentQueryResult(
-        component_id=comp_id,
-        component_type=type(component).__name__,
-        label=labels if labels else None,
-        properties=properties,
-    )
-
-    return CallToolResult(
-        content=[TextContent(type="text", text=json.dumps(structured, default=str))],
-        structuredContent=structured,
-    )
+        return CallToolResult(
+            content=[TextContent(type="text", text=json.dumps(structured, default=str))],
+            structuredContent=structured,
+        )
diff --git a/dash/mcp/primitives/tools/tools_callbacks.py b/dash/mcp/primitives/tools/tools_callbacks.py
@@ -1,6 +1,6 @@
 """Dynamic callback tools for MCP.
 
-Handles listing, naming, and executing callback-based tools.
+Exposes every server-callable callback as an MCP tool.
 """
 
 from __future__ import annotations
@@ -12,36 +12,41 @@
 from dash import get_app
 from dash.mcp.types import CallbackExecutionError, ToolNotFoundError
 
+from .base import MCPToolProvider
 from .results import format_callback_response
 
 
-def get_tool_names() -> set[str]:
-    return get_app().mcp_callback_map.tool_names
-
-
-def get_tools() -> list[Tool]:
-    """Return one Tool per server-callable callback."""
-    return get_app().mcp_callback_map.as_mcp_tools()
-
-
-def call_tool(tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
-    """Execute a callback tool by name."""
-    from .callback_utils import run_callback
-
-    callback_map = get_app().mcp_callback_map
-    cb = callback_map.find_by_tool_name(tool_name)
-    if cb is None:
-        raise ToolNotFoundError(
-            f"Tool not found: {tool_name}."
-            " The app's callbacks may have changed."
-            " Please call tools/list to refresh your tool list."
-        )
-
-    try:
-        dispatch_response = run_callback(cb, arguments)
-    except CallbackExecutionError as e:
-        return CallToolResult(
-            content=[TextContent(type="text", text=str(e))],
-            isError=True,
-        )
-    return format_callback_response(dispatch_response, cb)
+class CallbackTools(MCPToolProvider):
+    """Exposes every server-callable callback as an MCP tool."""
+
+    @classmethod
+    def get_tool_names(cls) -> set[str]:
+        return get_app().mcp_callback_map.tool_names
+
+    @classmethod
+    def list_tools(cls) -> list[Tool]:
+        """Return one Tool per server-callable callback."""
+        return get_app().mcp_callback_map.as_mcp_tools()
+
+    @classmethod
+    def call_tool(cls, tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
+        """Execute a callback tool by name."""
+        from .callback_utils import run_callback
+
+        callback_map = get_app().mcp_callback_map
+        cb = callback_map.find_by_tool_name(tool_name)
+        if cb is None:
+            raise ToolNotFoundError(
+                f"Tool not found: {tool_name}."
+                " The app's callbacks may have changed."
+                " Please call tools/list to refresh your tool list."
+            )
+
+        try:
+            callback_response = run_callback(cb, arguments)
+        except CallbackExecutionError as e:
+            return CallToolResult(
+                content=[TextContent(type="text", text=str(e))],
+                isError=True,
+            )
+        return format_callback_response(callback_response, cb)
