lint

Author: KoolADE85

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

@@ -8,6 +8,7 @@
 
 from typing import TYPE_CHECKING
 
+from ..tool_background_tasks import GET_RESULT_TOOL_NAME
 from .base import ToolDescriptionSource
 
 if TYPE_CHECKING:
@@ -19,9 +20,9 @@ class BackgroundCallbackDescription(ToolDescriptionSource):
 
     @classmethod
     def describe(cls, callback: CallbackAdapter) -> list[str]:
+        # pylint: disable-next=protected-access
         if not callback._cb_info.get("background"):
             return []
-        from ..tool_background_tasks import GET_RESULT_TOOL_NAME
 
         return [
             "",

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, 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,
@@ -61,17 +63,21 @@ def task_result_to_tool_result(create_task_result: CreateTaskResult) -> CallTool
     """
     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."
+        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

@@ -21,10 +21,7 @@
 
 
 def _has_background_callbacks() -> bool:
-    return any(
-        cb_info.get("background")
-        for cb_info in get_app().callback_map.values()
-    )
+    return any(cb_info.get("background") for cb_info in get_app().callback_map.values())
 
 
 class BackgroundTaskTools(MCPToolProvider):

dash/mcp/primitives/tools/tools_callbacks.py

@@ -47,6 +47,7 @@ def call_tool(
                 " Please call tools/list to refresh your tool list."
             )
 
+        # pylint: disable-next=protected-access
         is_background = bool(cb._cb_info.get("background"))
 
         try:

dash/mcp/tasks/tasks.py

@@ -6,7 +6,7 @@
 from datetime import datetime, timezone
 from typing import Any
 
-from mcp.types import CreateTaskResult, GetTaskResult, Task
+from mcp.types import CancelTaskResult, CreateTaskResult, GetTaskResult, Task
 
 from dash import get_app
 from dash.mcp.primitives.tools.results import format_callback_response
@@ -33,6 +33,7 @@ def create_task(dispatch_response: dict[str, Any], callback) -> CreateTaskResult
     cache_key = dispatch_response["cacheKey"]
     job_id = str(dispatch_response["job"])
     task_id = f"{callback.tool_name}:{job_id}:{cache_key}"
+    # pylint: disable-next=protected-access
     interval = callback._cb_info.get("background", {}).get("interval", 1000)
     now = datetime.now(timezone.utc)
     return CreateTaskResult(
@@ -75,14 +76,17 @@ def get_task(task_id: str) -> GetTaskResult:
     adapter = get_app().mcp_callback_map.find_by_tool_name(tool_name)
     interval = None
     if adapter is not None:
+        # pylint: disable-next=protected-access
         interval = adapter._cb_info.get("background", {}).get("interval", 1000)
 
     now = datetime.now(timezone.utc)
     return GetTaskResult(
         taskId=task_id,
         status=status,
         statusMessage=str(progress) if progress else None,
-        createdAt=datetime.fromisoformat(manager.handle.get(f"{cache_key}-created_at") or now.isoformat()),
+        createdAt=datetime.fromisoformat(
+            manager.handle.get(f"{cache_key}-created_at") or now.isoformat()
+        ),
         lastUpdatedAt=now,
         ttl=manager.expire * 1000 if manager.expire else None,
         pollInterval=interval,
@@ -117,7 +121,9 @@ def get_task_result(task_id: str) -> Any:
     response_data = json.loads(response.get_data(as_text=True))
 
     if "response" not in response_data:
-        raise MCPError("Task result not ready. Poll tasks/get until status is 'completed'.")
+        raise MCPError(
+            "Task result not ready. Poll tasks/get until status is 'completed'."
+        )
 
     return format_callback_response(response_data, adapter)
 
@@ -127,9 +133,7 @@ def cancel_task(task_id: str) -> Any:
 
     Same underlying mechanism as the renderer's cancelJob query param.
     """
-    from mcp.types import CancelTaskResult
-
-    tool_name, job_id, cache_key = parse_task_id(task_id)
+    _tool_name, job_id, cache_key = parse_task_id(task_id)
 
     manager = _get_callback_manager()
     if manager is None: