MCP Server Part 7: get_dash_component tool and callback tool execution

dash/mcp/primitives/tools/__init__.py

@@ -0,0 +1,38 @@
+"""MCP tool listing and call handling."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.types import CallToolResult, ListToolsResult
+
+from dash.mcp.types import ToolNotFoundError
+
+from .base import MCPToolProvider
+from .tool_get_dash_component import GetDashComponentTool
+from .tools_callbacks import CallbackTools
+
+_TOOL_PROVIDERS: list[type[MCPToolProvider]] = [
+    CallbackTools,
+    GetDashComponentTool,
+]
+
+
+def list_tools() -> ListToolsResult:
+    """Build the MCP tools/list response."""
+    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:
+    """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."
+        " Please call tools/list to refresh your tool list."
+    )

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

dash/mcp/primitives/tools/callback_adapter_collection.py

@@ -86,8 +86,14 @@ def find_by_tool_name(self, name: str) -> CallbackAdapter | None:
                 return cb
         return None
 
-    def find_by_output(self, id_and_prop: str) -> CallbackAdapter | None:
-        """Find the adapter that outputs to ``id_and_prop`` (``"component_id.property"``)."""
+    @cached_property
+    def outputs_by_prop(self) -> dict[str, list[CallbackAdapter]]:
+        """Index ``id_and_prop`` → callbacks outputting it.
+
+        Mirrors the dash-renderer's ``outputMap`` (see
+        ``dash-renderer/src/actions/dependencies.js``).
+        """
+        idx: dict[str, list[CallbackAdapter]] = {}
         for cb in self._callbacks:
             try:
                 parsed = split_callback_id(cb.output_id)
@@ -96,9 +102,31 @@ def find_by_output(self, id_and_prop: str) -> CallbackAdapter | None:
             if isinstance(parsed, dict):
                 parsed = [parsed]
             for p in parsed:
-                if f"{p['id']}.{clean_property_name(p['property'])}" == id_and_prop:
-                    return cb
-        return None
+                key = f"{p['id']}.{clean_property_name(p['property'])}"
+                idx.setdefault(key, []).append(cb)
+        return idx
+
+    @cached_property
+    def inputs_by_prop(self) -> dict[str, list[CallbackAdapter]]:
+        """Index ``id_and_prop`` → callbacks consuming it as input/state.
+
+        Mirrors the dash-renderer's ``inputMap`` (see
+        ``dash-renderer/src/actions/dependencies.js``).
+        Many callbacks may share a key.
+        """
+        idx: dict[str, list[CallbackAdapter]] = {}
+        for cb in self._callbacks:
+            # pylint: disable-next=protected-access
+            deps = cb._cb_info.get("inputs", []) + cb._cb_info.get("state", [])
+            for dep in deps:
+                key = f"{dep.get('id', 'unknown')}.{dep.get('property', 'unknown')}"
+                idx.setdefault(key, []).append(cb)
+        return idx
+
+    def find_by_output(self, id_and_prop: str) -> CallbackAdapter | None:
+        """Find the adapter that outputs to ``id_and_prop`` (``"component_id.property"``)."""
+        candidates = self.outputs_by_prop.get(id_and_prop, [])
+        return candidates[0] if candidates else None
 
     def get_initial_value(self, id_and_prop: str) -> Any:
         """Return the initial value for ``id_and_prop`` (``"component_id.property"``).

dash/mcp/primitives/tools/tool_get_dash_component.py

@@ -0,0 +1,129 @@
+"""Built-in tool: get_dash_component."""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from mcp.types import CallToolResult, TextContent, Tool
+from pydantic import Field, TypeAdapter
+from typing_extensions import Annotated, NotRequired, TypedDict
+
+from dash import get_app
+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")]
+    property: NotRequired[
+        Annotated[
+            str,
+            Field(
+                description="The property name to read (e.g. 'options', 'value'). Omit to list all defined properties."
+            ),
+        ]
+    ]
+
+
+_INPUT_SCHEMA = TypeAdapter(_ComponentQueryInput).json_schema()
+_OUTPUT_SCHEMA = TypeAdapter(ComponentQueryResult).json_schema()
+
+NAME = "get_dash_component"
+
+
+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,
+            )
+        ]
+
+    @classmethod
+    def call_tool(cls, tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
+        comp_id = arguments.get("component_id", "")
+        if not comp_id:
+            return CallToolResult(
+                content=[TextContent(type="text", text="component_id is required")],
+                isError=True,
+            )
+
+        prop_filter = arguments.get("property", "")
+        component = find_component(comp_id)
+        callback_map = get_app().mcp_callback_map
+
+        if component is None:
+            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,
+            )
+
+        properties: dict[str, ComponentPropertyInfo] = {}
+        for prop_name in getattr(component, "_prop_names", []):
+            if prop_filter and prop_name != prop_filter:
+                continue
+
+            id_and_prop = f"{comp_id}.{prop_name}"
+            value = callback_map.get_initial_value(id_and_prop)
+            if value is None:
+                value = getattr(component, prop_name, None)
+            if value is None:
+                continue
+
+            modified_by = [
+                cb.tool_name for cb in callback_map.outputs_by_prop.get(id_and_prop, [])
+            ]
+            input_to = [
+                cb.tool_name for cb in callback_map.inputs_by_prop.get(id_and_prop, [])
+            ]
+
+            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,
+        )
+
+        return CallToolResult(
+            content=[
+                TextContent(type="text", text=json.dumps(structured, default=str))
+            ],
+            structuredContent=structured,
+        )

dash/mcp/primitives/tools/tools_callbacks.py

@@ -0,0 +1,51 @@
+"""Dynamic callback tools for MCP.
+
+Exposes every server-callable callback as an MCP tool.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.types import CallToolResult, TextContent, Tool
+
+from dash import get_app
+from dash.mcp.types import CallbackExecutionError, ToolNotFoundError
+
+from .base import MCPToolProvider
+from .callback_utils import run_callback
+from .results import format_callback_response
+
+
+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."""
+        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)