Wire MCP server, SSE transport, and Dash app integration

Author: KoolADE85

dash/_configs.py

@@ -32,6 +32,8 @@ def load_dash_env_vars():
                 "DASH_DISABLE_VERSION_CHECK",
                 "DASH_PRUNE_ERRORS",
                 "DASH_COMPRESS",
+                "DASH_MCP_ENABLED",
+                "DASH_MCP_PATH",
                 "HOST",
                 "PORT",
             )

dash/dash.py

@@ -472,6 +472,8 @@ def __init__(  # pylint: disable=too-many-statements
         on_error: Optional[Callable[[Exception], Any]] = None,
         use_async: Optional[bool] = None,
         health_endpoint: Optional[str] = None,
+        enable_mcp: Optional[bool] = None,
+        mcp_path: Optional[str] = None,
         **obsolete,
     ):
 
@@ -573,6 +575,13 @@ def __init__(  # pylint: disable=too-many-statements
         # keep title as a class property for backwards compatibility
         self.title = title
 
+        # MCP (Model Context Protocol) configuration
+        self._enable_mcp = get_combined_config("mcp_enabled", enable_mcp, True)
+        _mcp_path = get_combined_config("mcp_path", mcp_path, "_mcp")
+        self._mcp_path = (
+            _mcp_path.lstrip("/") if isinstance(_mcp_path, str) else _mcp_path
+        )
+
         # list of dependencies - this one is used by the back end for dispatching
         self.callback_map: dict = {}
         # same deps as a list to catch duplicate outputs, and to send to the front end
@@ -793,6 +802,21 @@ def _setup_routes(self):
                 hook.data["methods"],
             )
 
+        if self._enable_mcp:
+            from .mcp import (  # pylint: disable=import-outside-toplevel
+                enable_mcp_server,
+            )
+
+            try:
+                enable_mcp_server(self, self._mcp_path)
+            except Exception as e:  # pylint: disable=broad-exception-caught
+                self._enable_mcp = False
+                self.logger.warning(
+                    "MCP server could not be started at '%s': %s",
+                    self._mcp_path,
+                    e,
+                )
+
         # catch-all for front-end routes, used by dcc.Location
         self._add_url("<path:path>", self.index)
 
@@ -2526,6 +2550,13 @@ def verify_url_part(served_part, url_part, part_name):
 
             if not jupyter_dash or not jupyter_dash.in_ipython:
                 self.logger.info("Dash is running on %s://%s%s%s\n", *display_url)
+                if self._enable_mcp:
+                    self.logger.info(
+                        " * MCP available at %s://%s%s%s%s\n",
+                        *display_url[:3],
+                        self.config.routes_pathname_prefix,
+                        self._mcp_path,
+                    )
 
         if self.config.extra_hot_reload_paths:
             extra_files = flask_run_options["extra_files"] = []

dash/mcp/__init__.py

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

dash/mcp/_server.py

@@ -0,0 +1,277 @@
+"""Flask route setup, Streamable HTTP transport, and MCP message handling."""
+
+from __future__ import annotations
+
+import atexit
+import json
+import logging
+import uuid
+from typing import TYPE_CHECKING, Any
+
+from flask import Response, request
+
+from dash.mcp.types import MCPError
+
+if TYPE_CHECKING:
+    from dash import Dash
+
+from dash import get_app
+
+from mcp.types import (
+    LATEST_PROTOCOL_VERSION,
+    ErrorData,
+    Implementation,
+    InitializeResult,
+    JSONRPCError,
+    JSONRPCResponse,
+    ResourcesCapability,
+    ServerCapabilities,
+    ToolsCapability,
+)
+
+from dash.version import __version__
+from dash.mcp._sse import (
+    close_sse_stream,
+    create_sse_stream,
+    shutdown_all_streams,
+)
+from dash.mcp.primitives import (
+    call_tool,
+    list_resource_templates,
+    list_resources,
+    list_tools,
+    read_resource,
+)
+from dash.mcp.primitives.tools.callback_adapter_collection import (
+    CallbackAdapterCollection,
+)
+
+logger = logging.getLogger(__name__)
+
+
+def enable_mcp_server(app: Dash, mcp_path: str) -> None:
+    """
+    Add MCP routes to a Dash/Flask app.
+
+    Registers a single Streamable HTTP endpoint for the MCP protocol.
+    Uses ``app._add_url()`` so that ``routes_pathname_prefix`` is applied
+    automatically.
+
+    Args:
+        app: The Dash application instance.
+        mcp_path: Route prefix for MCP endpoints.
+    """
+    # Session storage: session_id -> metadata
+    sessions: dict[str, dict[str, Any]] = {}
+
+    def _create_session() -> str:
+        sid = str(uuid.uuid4())
+        sessions[sid] = {}
+        return sid
+
+    # -- Streamable HTTP endpoint --------------------------------------------
+
+    def mcp_handler() -> Response:
+        if request.method == "POST":
+            return _handle_post()
+        if request.method == "GET":
+            return _handle_get()
+        if request.method == "DELETE":
+            return _handle_delete()
+        return Response(
+            json.dumps({"error": "Method not allowed"}),
+            content_type="application/json",
+            status=405,
+        )
+
+    def _handle_get() -> Response:
+        session_id = request.headers.get("mcp-session-id")
+        if not session_id or session_id not in sessions:
+            return Response(
+                json.dumps({"error": "Session not found"}),
+                content_type="application/json",
+                status=404,
+            )
+        return create_sse_stream(sessions, session_id)
+
+    def _handle_post() -> Response:
+        content_type = request.content_type or ""
+        if "application/json" not in content_type:
+            return Response(
+                json.dumps({"error": "Content-Type must be application/json"}),
+                content_type="application/json",
+                status=415,
+            )
+
+        try:
+            data = request.get_json()
+        except Exception:
+            return Response(
+                json.dumps({"error": "Invalid JSON"}),
+                content_type="application/json",
+                status=400,
+            )
+
+        method = data.get("method", "")
+        request_id = data.get("id")
+        session_id = request.headers.get("mcp-session-id")
+
+        stale_session = False
+        if method == "initialize":
+            session_id = _create_session()
+        elif session_id and session_id not in sessions:
+            stale_session = True
+            sessions[session_id] = {}
+        elif not session_id:
+            session_id = _create_session()
+
+        response_data = _process_mcp_message(data)
+
+        if response_data is None:
+            return Response("", status=202)
+
+        if stale_session:
+            _inject_warning(response_data, _STALE_SESSION_WARNING)
+
+        return Response(
+            json.dumps(response_data),
+            content_type="application/json",
+            status=200,
+            headers={"mcp-session-id": session_id},
+        )
+
+    def _handle_delete() -> Response:
+        session_id = request.headers.get("mcp-session-id")
+        if not session_id or session_id not in sessions:
+            return Response(
+                json.dumps({"error": "Session not found"}),
+                content_type="application/json",
+                status=404,
+            )
+        close_sse_stream(sessions[session_id])
+        del sessions[session_id]
+        logger.info("MCP session terminated: %s", session_id)
+        return Response("", status=204)
+
+    # -- Register routes -----------------------------------------------------
+
+    from dash._get_app import with_app_context_factory
+
+    app._add_url(
+        mcp_path, with_app_context_factory(mcp_handler, app), ["GET", "POST", "DELETE"]
+    )
+
+    # Close all SSE streams on server shutdown so MCP clients see a
+    # clean stream end and can reconnect promptly.
+    atexit.register(shutdown_all_streams, sessions)
+
+    logger.info(
+        "MCP routes registered at %s%s",
+        app.config.routes_pathname_prefix,
+        mcp_path,
+    )
+
+
+_STALE_SESSION_WARNING = (
+    "[Warning: your session was not recognised"
+    " — the app may have restarted."
+    " Please call tools/list to refresh your tool list."
+    " Please ask the user to reconnect to the MCP server.]"
+)
+
+
+def _inject_warning(response_data: dict[str, Any], warning: str) -> None:
+    """Append a warning to a JSON-RPC response dict.
+
+    For successful ``tools/call`` responses the warning is added as an
+    extra text content block so the agent sees it alongside the result.
+    For error responses the warning is appended to the error message.
+    Other responses (tools/list, resources/*) are left unchanged — the
+    JSON-RPC spec forbids extra top-level keys.
+    """
+    # tools/call success: result has a "content" list
+    result = response_data.get("result")
+    if isinstance(result, dict) and isinstance(result.get("content"), list):
+        result["content"].append({"type": "text", "text": warning})
+        return
+
+    # Error response
+    error = response_data.get("error")
+    if isinstance(error, dict) and "message" in error:
+        error["message"] += " " + warning
+
+
+def _handle_initialize() -> InitializeResult:
+    return InitializeResult(
+        protocolVersion=LATEST_PROTOCOL_VERSION,
+        capabilities=ServerCapabilities(
+            tools=ToolsCapability(listChanged=True),
+            resources=ResourcesCapability(),
+        ),
+        serverInfo=Implementation(name="Plotly Dash", version=__version__),
+        instructions=(
+            "This is a Dash web application. "
+            "Dash apps are stateless: calling a tool executes "
+            "a callback and returns its result to you, but does "
+            "NOT update the user's browser. "
+            "Use tool results to answer questions about what "
+            "the app would produce for given inputs."
+        ),
+    )
+
+
+def _process_mcp_message(data: dict[str, Any]) -> dict[str, Any] | None:
+    """
+    Process an MCP JSON-RPC message and return the response dict.
+
+    Returns ``None`` for notifications (no ``id`` field).
+    """
+    method = data.get("method", "")
+    params = data.get("params", {}) or {}
+    request_id = data.get("id")
+
+    app = get_app()
+    if not hasattr(app, "mcp_callback_map"):
+        app.mcp_callback_map = CallbackAdapterCollection(app)
+
+    mcp_methods = {
+        "initialize": _handle_initialize,
+        "tools/list": lambda: list_tools(),
+        "tools/call": lambda: call_tool(
+            params.get("name", ""), params.get("arguments", {})
+        ),
+        "resources/list": lambda: list_resources(),
+        "resources/templates/list": lambda: list_resource_templates(),
+        "resources/read": lambda: read_resource(params.get("uri", "")),
+    }
+
+    try:
+        handler = mcp_methods.get(method)
+        if handler is None:
+            if method.startswith("notifications/"):
+                return None
+            raise ValueError(f"Unknown method: {method}")
+
+        result = handler()
+
+        response = JSONRPCResponse(
+            jsonrpc="2.0",
+            id=request_id,
+            result=result.model_dump(exclude_none=True, mode="json"),
+        )
+        return response.model_dump(exclude_none=True, mode="json")
+
+    except MCPError as e:
+        logger.error("MCP error: %s", e)
+        return JSONRPCError(
+            jsonrpc="2.0",
+            id=request_id,
+            error=ErrorData(code=e.code, message=str(e)),
+        ).model_dump(exclude_none=True)
+    except Exception as e:
+        logger.error("MCP error: %s", e, exc_info=True)
+        return JSONRPCError(
+            jsonrpc="2.0",
+            id=request_id,
+            error=ErrorData(code=-32603, message=f"{type(e).__name__}: {e}"),
+        ).model_dump(exclude_none=True)