Add decorator to make any function mcp-enabled

Author: KoolADE85

dash/dash.py

@@ -611,6 +611,7 @@ def __init__(  # pylint: disable=too-many-statements
         # same deps as a list to catch duplicate outputs, and to send to the front end
         self._callback_list: list = []
         self.callback_api_paths: dict = {}
+        self.mcp_decorated_functions: dict = {}
 
         # list of inline scripts
         self._inline_scripts: list = []

dash/mcp/__init__.py

@@ -1,7 +1,9 @@
 """Dash MCP (Model Context Protocol) server integration."""
 
+from dash.mcp._decorator import mcp_enabled
 from dash.mcp._server import enable_mcp_server
 
 __all__ = [
     "enable_mcp_server",
+    "mcp_enabled",
 ]

dash/mcp/_decorator.py

@@ -0,0 +1,51 @@
+"""Decorator to expose plain Python functions as MCP tools."""
+
+from __future__ import annotations
+
+import functools
+from typing import Any, Callable, Optional
+
+from typing_extensions import TypedDict
+
+
+class MCPToolRegistration(TypedDict):
+    fn: Callable[..., Any]
+    expose_docstring: Optional[bool]
+
+
+MCP_DECORATED_FUNCTIONS: dict[str, MCPToolRegistration] = {}
+
+
+def mcp_enabled(
+    func: Callable[..., Any] | None = None,
+    *,
+    name: str | None = None,
+    expose_docstring: Optional[bool] = None,
+) -> Callable[..., Any]:
+    """Mark a function as an MCP tool.
+
+    Supports both bare and parameterised usage::
+
+        @mcp_enabled
+        def my_tool(x: int) -> str: ...
+
+        @mcp_enabled(name="custom_name", expose_docstring=True)
+        def my_tool(x: int) -> str: ...
+    """
+
+    def _wrap(fn: Callable[..., Any]) -> Callable[..., Any]:
+        tool_name = name if name else fn.__name__
+        MCP_DECORATED_FUNCTIONS[tool_name] = MCPToolRegistration(
+            fn=fn,
+            expose_docstring=expose_docstring,
+        )
+
+        @functools.wraps(fn)
+        def wrapper(*args: Any, **kwargs: Any) -> Any:
+            return fn(*args, **kwargs)
+
+        return wrapper
+
+    if func is not None:
+        return _wrap(func)
+    return _wrap

dash/mcp/_server.py

@@ -25,6 +25,7 @@
 
 from dash import get_app
 from dash._get_app import with_app_context_factory
+from dash.mcp._decorator import MCP_DECORATED_FUNCTIONS
 from dash.mcp.primitives import (
     call_tool,
     list_resource_templates,
@@ -46,6 +47,10 @@
 
 def enable_mcp_server(app: Dash, mcp_path: str) -> None:
     """Add MCP routes to a Dash/Flask app."""
+
+    app.mcp_decorated_functions = dict(MCP_DECORATED_FUNCTIONS)
+    MCP_DECORATED_FUNCTIONS.clear()
+
     # -- Streamable HTTP endpoint --------------------------------------------
 
     def mcp_handler() -> Response:

dash/mcp/primitives/tools/__init__.py

@@ -9,12 +9,14 @@
 from dash.mcp.types import ToolNotFoundError
 
 from .base import MCPToolProvider
+from .tool_decorated_mcp_functions import DecoratedFunctionTools
 from .tool_get_dash_component import GetDashComponentTool
 from .tools_callbacks import CallbackTools
 
 _TOOL_PROVIDERS: list[type[MCPToolProvider]] = [
     CallbackTools,
     GetDashComponentTool,
+    DecoratedFunctionTools,
 ]
 
 

dash/mcp/primitives/tools/tool_decorated_mcp_functions.py

@@ -0,0 +1,148 @@
+"""MCP tools backed by ``@mcp_enabled``-decorated functions."""
+
+from __future__ import annotations
+
+import inspect
+import json
+import typing
+from typing import Any, Callable
+
+from mcp.types import CallToolResult, TextContent, Tool
+
+from dash import get_app
+from dash.mcp._decorator import MCPToolRegistration
+from dash.mcp.primitives.tools.input_schemas import get_input_schema
+from dash.mcp.primitives.tools.input_schemas.schema_callback_type_annotations import (
+    annotation_to_json_schema,
+)
+from dash.mcp.types import MCPInput, is_nullable
+
+from .base import MCPToolProvider
+
+
+def _build_inputs(fn: Callable[..., Any]) -> list[MCPInput]:
+    """Build an ``MCPInput`` from each of the function's arguments."""
+    try:
+        hints = typing.get_type_hints(fn)
+    except Exception:  # pylint: disable=broad-exception-caught
+        hints = getattr(fn, "__annotations__", {})
+
+    sig = inspect.signature(fn)
+    inputs: list[MCPInput] = []
+
+    for name, param in sig.parameters.items():
+        annotation = hints.get(name)
+
+        has_default = param.default is not inspect.Parameter.empty
+        required = not has_default and (
+            annotation is None or not is_nullable(annotation)
+        )
+
+        inputs.append(
+            MCPInput(
+                name=name,
+                id_and_prop="",
+                component_id="",
+                property="",
+                annotation=annotation,
+                component_type=None,
+                component=None,
+                required=required,
+                initial_value=param.default if has_default else None,
+                upstream_output=None,
+            )
+        )
+    return inputs
+
+
+def _build_output_schema(fn: Callable[..., Any]) -> dict[str, Any]:
+    """Build a JSON Schema ``outputSchema`` from the return annotation.
+
+    The schema wraps the return type in ``{"result": <type>}`` to match
+    the object that ``call_tool`` returns as ``structuredContent``.
+    """
+    try:
+        hints = typing.get_type_hints(fn)
+    except Exception:  # pylint: disable=broad-exception-caught
+        hints = getattr(fn, "__annotations__", {})
+
+    ret = hints.get("return")
+    if ret is None:
+        return {}
+
+    inner = annotation_to_json_schema(ret)
+    if inner is None:
+        return {}
+
+    return {
+        "type": "object",
+        "properties": {"result": inner},
+        "required": ["result"],
+    }
+
+
+def _build_tool(tool_name: str, reg: MCPToolRegistration) -> Tool:
+    fn = reg["fn"]
+    inputs = _build_inputs(fn)
+    properties = {p["name"]: get_input_schema(p) for p in inputs}
+    required = [p["name"] for p in inputs if p["required"]]
+
+    input_schema: dict[str, Any] = {"type": "object", "properties": properties}
+    if required:
+        input_schema["required"] = required
+
+    expose_docstring = reg["expose_docstring"]
+    if expose_docstring is None:
+        expose_docstring = get_app().config.get("mcp_expose_docstrings", False)
+
+    description = "MCP tool"
+    if expose_docstring:
+        docstring = getattr(fn, "__doc__", None)
+        if docstring:
+            description = docstring.strip()
+
+    return Tool(
+        name=tool_name,
+        description=description,
+        inputSchema=input_schema,
+        outputSchema=_build_output_schema(fn),
+    )
+
+
+class DecoratedFunctionTools(MCPToolProvider):
+    """Exposes ``@mcp_enabled``-decorated functions as MCP tools."""
+
+    @classmethod
+    def _registry(cls) -> dict[str, MCPToolRegistration]:
+        return get_app().mcp_decorated_functions
+
+    @classmethod
+    def get_tool_names(cls) -> set[str]:
+        return set(cls._registry().keys())
+
+    @classmethod
+    def list_tools(cls) -> list[Tool]:
+        return [_build_tool(name, reg) for name, reg in cls._registry().items()]
+
+    @classmethod
+    def call_tool(cls, tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
+        reg = cls._registry().get(tool_name)
+        if reg is None:
+            return CallToolResult(
+                content=[TextContent(type="text", text=f"Tool not found: {tool_name}")],
+                isError=True,
+            )
+        fn = reg["fn"]
+        try:
+            result = fn(**arguments)
+        except Exception as exc:  # pylint: disable=broad-exception-caught
+            return CallToolResult(
+                content=[TextContent(type="text", text=f"{type(exc).__name__}: {exc}")],
+                isError=True,
+            )
+
+        serialized = json.dumps(result, default=str)
+        return CallToolResult(
+            content=[TextContent(type="text", text=serialized)],
+            structuredContent={"result": result},
+        )