MCP Server Part 9: Background callbacks

.github/workflows/testing.yml

@@ -108,7 +108,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip wheel
           python -m pip install "setuptools<80.0.0"
-          find packages -name dash-*.whl -print -exec sh -c 'pip install "{}[dev,ci,testing]"' \;
+          find packages -name dash-*.whl -print -exec sh -c 'pip install "{}[dev,ci,testing,diskcache]"' \;
 
       - name: Install dash-renderer dependencies
         working-directory: dash/dash-renderer
@@ -231,7 +231,7 @@ jobs:
         run: |
           python -m pip install --upgrade pip wheel
           python -m pip install "setuptools<80.0.0"
-          find packages -name dash-*.whl -print -exec sh -c 'pip install "{}[ci,testing,dev]"' \;
+          find packages -name dash-*.whl -print -exec sh -c 'pip install "{}[ci,testing,dev,diskcache]"' \;
 
       - name: Build/Setup test components
         run: npm run setup-tests.py  # TODO build the packages and save them to packages/ in build job

dash/_callback.py

@@ -452,10 +452,13 @@ def _setup_background_callback(
     return to_json(data)
 
 
-def _progress_background_callback(response, callback_manager, background):
+def _progress_background_callback(
+    response, callback_manager, background, cache_key=None
+):
     progress_outputs = background.get("progress")
-    adapter = get_app().backend.request_adapter()
-    cache_key = adapter.args.get("cacheKey")
+    if cache_key is None:
+        adapter = get_app().backend.request_adapter()
+        cache_key = adapter.args.get("cacheKey")
 
     if progress_outputs:
         # Get the progress before the result as it would be erased after the results.
@@ -467,21 +470,38 @@ def _progress_background_callback(response, callback_manager, background):
 
 
 def _update_background_callback(
-    error_handler, callback_ctx, response, kwargs, background, multi
+    error_handler,
+    callback_ctx,
+    response,
+    kwargs,
+    background,
+    multi,
+    cache_key=None,
+    job_id=None,
 ):
     """Set up the background callback and manage jobs."""
     callback_manager = _get_callback_manager(kwargs, background)
 
-    adapter = get_app().backend.request_adapter()
-    cache_key = adapter.args.get("cacheKey") if adapter else None
-    job_id = adapter.args.get("job") if adapter else None
+    if cache_key is None or job_id is None:
+        adapter = get_app().backend.request_adapter()
+        cache_key = cache_key or (adapter.args.get("cacheKey") if adapter else None)
+        job_id = job_id or (adapter.args.get("job") if adapter else None)
 
-    _progress_background_callback(response, callback_manager, background)
+    _progress_background_callback(
+        response, callback_manager, background, cache_key=cache_key
+    )
 
     output_value = callback_manager.get_result(cache_key, job_id)
 
     return _handle_rest_background_callback(
-        output_value, callback_manager, response, error_handler, callback_ctx, multi
+        output_value,
+        callback_manager,
+        response,
+        error_handler,
+        callback_ctx,
+        multi,
+        cache_key=cache_key,
+        job_id=job_id,
     )
 
 
@@ -493,10 +513,13 @@ def _handle_rest_background_callback(
     callback_ctx,
     multi,
     has_update=False,
+    cache_key=None,
+    job_id=None,
 ):
-    adapter = get_app().backend.request_adapter()
-    cache_key = adapter.args.get("cacheKey") if adapter else None
-    job_id = adapter.args.get("job") if adapter else None
+    if cache_key is None or job_id is None:
+        adapter = get_app().backend.request_adapter()
+        cache_key = cache_key or (adapter.args.get("cacheKey") if adapter else None)
+        job_id = job_id or (adapter.args.get("job") if adapter else None)
     # Must get job_running after get_result since get_results terminates it.
     job_running = callback_manager.job_running(job_id)
     if not job_running and output_value is callback_manager.UNDEFINED:

dash/background_callback/managers/diskcache_manager.py

@@ -36,7 +36,7 @@ def __init__(self, cache=None, cache_by=None, expire=None):
             is determined by the default behavior of the ``cache`` instance.
         """
         try:
-            import diskcache  # type: ignore[import-not-found] # pylint: disable=import-outside-toplevel
+            import diskcache  # type: ignore[import-not-found,import-untyped] # pylint: disable=import-outside-toplevel
             import psutil  # type: ignore[import-untyped] # noqa: F401,E402 pylint: disable=import-outside-toplevel,unused-import,unused-variable,import-error
             import multiprocess  # type: ignore[import-untyped] # noqa: F401,E402 pylint: disable=import-outside-toplevel,unused-import,unused-variable
         except ImportError as missing_imports:

dash/mcp/_server.py

@@ -34,6 +34,7 @@
     list_tools,
     read_resource,
 )
+from dash.mcp.tasks import get_task, get_task_result, cancel_task
 from dash.mcp.primitives.tools.callback_adapter_collection import (
     CallbackAdapterCollection,
 )
@@ -251,11 +252,16 @@ def _process_mcp_message(data: dict[str, Any]) -> dict[str, Any] | None:
         "initialize": _handle_initialize,
         "tools/list": list_tools,
         "tools/call": lambda: call_tool(
-            params.get("name", ""), params.get("arguments", {})
+            tool_name=params.get("name", ""),
+            arguments=params.get("arguments", {}),
+            task=params.get("task"),
         ),
         "resources/list": list_resources,
         "resources/templates/list": list_resource_templates,
         "resources/read": lambda: read_resource(params.get("uri", "")),
+        "tasks/get": lambda: get_task(task_id=params.get("taskId", "")),
+        "tasks/result": lambda: get_task_result(task_id=params.get("taskId", "")),
+        "tasks/cancel": lambda: cancel_task(task_id=params.get("taskId", "")),
     }
 
     try:

dash/mcp/primitives/tools/__init__.py

@@ -4,17 +4,19 @@
 
 from typing import Any
 
-from mcp.types import CallToolResult, ListToolsResult
+from mcp.types import CallToolResult, CreateTaskResult, ListToolsResult
 
 from dash.mcp.types import ToolNotFoundError
 
 from .base import MCPToolProvider
+from .tool_background_tasks import BackgroundTaskTools
 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,
+    BackgroundTaskTools,
     GetDashComponentTool,
     DecoratedFunctionTools,
 ]
@@ -28,11 +30,17 @@ def list_tools() -> ListToolsResult:
     return ListToolsResult(tools=tools)
 
 
-def call_tool(tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
-    """Route a tools/call request by tool name."""
+def call_tool(
+    tool_name: str, arguments: dict[str, Any], task: dict | None = None
+) -> CallToolResult | CreateTaskResult:
+    """Route a tools/call request by tool name.
+
+    The optional ``task`` parameter (per MCP Tasks protocol) is passed
+    through to providers that support background callbacks.
+    """
     for provider in _TOOL_PROVIDERS:
         if tool_name in provider.get_tool_names():
-            return provider.call_tool(tool_name, arguments)
+            return provider.call_tool(tool_name, arguments, task=task)
     raise ToolNotFoundError(
         f"Tool not found: {tool_name}."
         " The app's callbacks may have changed."

dash/mcp/primitives/tools/base.py

@@ -4,7 +4,7 @@
 
 from typing import Any
 
-from mcp.types import CallToolResult, Tool
+from mcp.types import CallToolResult, CreateTaskResult, Tool
 
 
 class MCPToolProvider:
@@ -24,5 +24,7 @@ def list_tools(cls) -> list[Tool]:
         raise NotImplementedError
 
     @classmethod
-    def call_tool(cls, tool_name: str, arguments: dict[str, Any]) -> CallToolResult:
+    def call_tool(
+        cls, tool_name: str, arguments: dict[str, Any], task: dict | None = None
+    ) -> CallToolResult | CreateTaskResult:
         raise NotImplementedError

dash/mcp/primitives/tools/descriptions/__init__.py

@@ -13,6 +13,7 @@
 from typing import TYPE_CHECKING
 
 from .base import ToolDescriptionSource
+from .description_background_callbacks import BackgroundCallbackDescription
 from .description_docstring import DocstringDescription
 from .description_outputs import OutputSummaryDescription
 
@@ -22,6 +23,7 @@
 _SOURCES: list[type[ToolDescriptionSource]] = [
     OutputSummaryDescription,
     DocstringDescription,
+    BackgroundCallbackDescription,
 ]
 
 

dash/mcp/primitives/tools/descriptions/description_background_callbacks.py

@@ -0,0 +1,32 @@
+"""Description for background (long-running) callbacks.
+
+Informs the LLM that the tool returns a taskId immediately
+and must be polled via the background task result tool.
+"""
+
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+from ..tool_background_tasks import GET_RESULT_TOOL_NAME
+from .base import ToolDescriptionSource
+
+if TYPE_CHECKING:
+    from dash.mcp.primitives.tools.callback_adapter import CallbackAdapter
+
+
+class BackgroundCallbackDescription(ToolDescriptionSource):
+    """Add async polling instructions for background callbacks."""
+
+    @classmethod
+    def describe(cls, callback: CallbackAdapter) -> list[str]:
+        # pylint: disable-next=protected-access
+        if not callback._cb_info.get("background"):
+            return []
+
+        return [
+            "",
+            "This is a long-running background operation. "
+            "It returns a taskId immediately. "
+            f"Call tool `{GET_RESULT_TOOL_NAME}` with the taskId to poll for the result.",
+        ]

dash/mcp/primitives/tools/results/__init__.py

@@ -7,17 +7,19 @@
 from __future__ import annotations
 
 import json
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
-from mcp.types import CallToolResult, TextContent
+from mcp.types import CallToolResult, CreateTaskResult, TextContent
 
 from dash.types import CallbackExecutionResponse
-from dash.mcp.primitives.tools.callback_adapter import CallbackAdapter
 
 from .base import ResultFormatter
 from .result_dataframe import DataFrameResult
 from .result_plotly_figure import PlotlyFigureResult
 
+if TYPE_CHECKING:
+    from dash.mcp.primitives.tools.callback_adapter import CallbackAdapter
+
 _RESULT_FORMATTERS: list[type[ResultFormatter]] = [
     PlotlyFigureResult,
     DataFrameResult,
@@ -50,3 +52,32 @@ def format_callback_response(
         content=content,
         structuredContent=dict(response),
     )
+
+
+def task_result_to_tool_result(create_task_result: CreateTaskResult) -> CallToolResult:
+    """Wrap a CreateTaskResult as a CallToolResult with polling instructions.
+
+    MCP Tasks are not yet supported by LLM clients, so this converts the
+    task metadata into a tool response that guides the LLM to poll via
+    the get_background_task_result tool.
+    """
+    task = create_task_result.task
+    return CallToolResult(
+        content=[
+            TextContent(
+                type="text",
+                text=json.dumps(
+                    {
+                        "taskId": task.taskId,
+                        "status": task.status,
+                        "pollInterval": task.pollInterval,
+                        "message": (
+                            "This is a long-running background callback. "
+                            "Call the get_background_task_result tool with this taskId "
+                            "to poll for the result."
+                        ),
+                    }
+                ),
+            )
+        ],
+    )

dash/mcp/primitives/tools/tool_background_tasks.py

@@ -0,0 +1,102 @@
+"""Built-in tools for background callback task lifecycle.
+
+Thin wrappers around the spec-aligned core in dash.mcp.tasks.
+Only registered when the app has background callbacks.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from mcp.types import CallToolResult, TextContent, Tool
+
+from dash import get_app
+from dash.mcp.tasks import get_task, get_task_result, cancel_task
+
+from .base import MCPToolProvider
+
+
+GET_RESULT_TOOL_NAME = "get_background_task_result"
+CANCEL_TOOL_NAME = "cancel_background_task"
+
+
+def _has_background_callbacks() -> bool:
+    return any(cb_info.get("background") for cb_info in get_app().callback_map.values())
+
+
+class BackgroundTaskTools(MCPToolProvider):
+    """Built-in tools for polling and cancelling background callback tasks.
+
+    Only registered when the app has background callbacks.
+    """
+
+    @classmethod
+    def get_tool_names(cls) -> set[str]:
+        if not _has_background_callbacks():
+            return set()
+        return {GET_RESULT_TOOL_NAME, CANCEL_TOOL_NAME}
+
+    @classmethod
+    def list_tools(cls) -> list[Tool]:
+        if not _has_background_callbacks():
+            return []
+        return [
+            Tool(
+                name=GET_RESULT_TOOL_NAME,
+                description=(
+                    "Poll for the result of a long-running background callback. "
+                    "Pass the taskId returned by the original tool call. "
+                    "If the task is still running, call this tool again. "
+                    "If complete, returns the callback result."
+                ),
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "taskId": {
+                            "type": "string",
+                            "description": "The taskId returned by the background callback tool.",
+                        },
+                    },
+                    "required": ["taskId"],
+                },
+            ),
+            Tool(
+                name=CANCEL_TOOL_NAME,
+                description="Cancel a running background callback.",
+                inputSchema={
+                    "type": "object",
+                    "properties": {
+                        "taskId": {
+                            "type": "string",
+                            "description": "The taskId of the background task to cancel.",
+                        },
+                    },
+                    "required": ["taskId"],
+                },
+            ),
+        ]
+
+    @classmethod
+    def call_tool(
+        cls,
+        tool_name: str,
+        arguments: dict[str, Any],
+        task: dict | None = None,
+    ) -> CallToolResult:
+        task_id = arguments.get("taskId", "")
+
+        if tool_name == GET_RESULT_TOOL_NAME:
+            task_status = get_task(task_id)
+            if task_status.status == "completed":
+                return get_task_result(task_id)
+            return CallToolResult(
+                content=[TextContent(type="text", text=task_status.model_dump_json())],
+            )
+
+        if tool_name == CANCEL_TOOL_NAME:
+            result = cancel_task(task_id)
+            return CallToolResult(
+                content=[TextContent(type="text", text=result.model_dump_json())],
+            )
+
+        raise ValueError(f"Unknown tool: {tool_name}")