feat(mcp): add ordinary durable task driver

README.md

@@ -410,7 +410,7 @@ See the [Sandbox Configuration Guide](backend/docs/CONFIGURATION.md#sandbox) to
 
 DeerFlow supports configurable MCP servers and skills to extend its capabilities.
 For HTTP/SSE MCP servers, OAuth token flows are supported (`client_credentials`, `refresh_token`).
-For stdio MCP servers, per-tool call timeouts can be configured with `tool_call_timeout`.
+For stdio MCP servers, per-tool call timeouts can be configured with `tool_call_timeout`; durable background-task calls honor the same setting for HTTP/SSE servers as well.
 MCP tool names are prefixed with `<server_name>_` by default to prevent collisions across servers. If a server already namespaces its own tools, set `tool_name_prefix: false` on that server in `extensions_config.json` to keep the original names. Disable the prefix only when the resulting names remain unique across all enabled servers.
 Settings > Tools updates one MCP server at a time: an invalid stdio command on one server no longer blocks toggling another, while enabling that invalid server remains protected by the command allowlist and surfaces the backend validation message in the UI.
 Targeted updates accept both DeerFlow's `type` field and the MCP-spec `transport` field for SSE/HTTP servers.
@@ -425,7 +425,8 @@ explicit, model-selected MCP tool path can run alongside the separate automatic
 OpenViking memory backend; it does not replace automatic turn capture or recall. See the
 [OpenViking MCP tools configuration](backend/docs/MCP_SERVER.md#openviking-mcp-tools).
 
-The Gateway also includes a disabled-by-default, protocol-neutral foundation for durable long-running MCP tasks. It stores remote task handles outside model context, polls them under cross-worker leases, rejects results returned after their lease expires, schedules the next attempt from the time a remote status call finishes, isolates unexpected failures between claimed tasks, cancels in-flight polling during Gateway shutdown, and makes expired claims recoverable after restart. If remote submission succeeds but the handle cannot be persisted, the runtime makes a best-effort cancellation so an untracked task is not silently left running. The exact scoped duplicate-handle conflict is surfaced without cancellation because an existing durable row already owns that remote task. Durable recovery requires a SQL database backend (`sqlite` or `postgres`); the in-memory backend does not initialize this task repository. This foundation does not make existing MCP tools asynchronous by itself: `mcp_tasks.enabled` should remain `false` until a compatible task driver is configured. Ordinary `submit/status/cancel` tools and the future MCP Tasks extension can share the same runtime without making the model remember remote task IDs.
+The Gateway can adapt an MCP server's ordinary `submit` / `status` / `cancel` tools into durable background tasks. The Agent sees only the configured submit tool and a DeerFlow-local task ID; remote IDs are persisted before the submit call returns, while status and cancel stay internal to the runtime. Polling uses cross-worker leases, exponential retry backoff, scoped MCP sessions, bounded result storage, and restart recovery. Current-thread tasks are available through `GET /api/threads/{thread_id}/mcp-tasks` and its detail endpoint. Enable `mcp_tasks` in `config.yaml`, configure `task_toolsets` with exact raw tool names in `extensions_config.json`, and use a SQL database backend (`sqlite` or `postgres`). This phase does not yet wake the Agent when a task completes or add a frontend task panel.
+
 See the [MCP Server Guide](backend/docs/MCP_SERVER.md) for detailed instructions.
 
 Security: pass per-request MCP credentials only through `config.context.secrets`;

backend/app/gateway/app.py

@@ -27,6 +27,7 @@
     input_polish,
     integrations,
     mcp,
+    mcp_tasks,
     memory,
     models,
     runs,
@@ -329,25 +330,54 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
         except Exception:
             logger.exception("Failed to initialize scheduled task service")
 
-        try:
-            from app.mcp_tasks import McpTaskService
-            from deerflow.mcp.tasks import McpTaskDriverRegistry
-
-            if getattr(app.state, "mcp_task_repo", None) is not None:
-                mcp_task_drivers = McpTaskDriverRegistry()
-                mcp_task_service = McpTaskService(
-                    repository=app.state.mcp_task_repo,
-                    drivers=mcp_task_drivers,
-                    poll_interval_seconds=startup_config.mcp_tasks.poll_interval_seconds,
-                    lease_seconds=startup_config.mcp_tasks.lease_seconds,
-                    max_concurrent_polls=startup_config.mcp_tasks.max_concurrent_polls,
+        from app.mcp_tasks import McpTaskService
+        from deerflow.config.extensions_config import ExtensionsConfig
+        from deerflow.config.mcp_tasks_config import McpTasksConfig
+        from deerflow.mcp.task_tool_caller import McpTaskToolCaller
+        from deerflow.mcp.tasks import (
+            ORDINARY_MCP_TASK_DRIVER,
+            McpTaskDriverRegistry,
+            OrdinaryMcpTaskDriver,
+        )
+        from deerflow.mcp.tasks.runtime import (
+            configured_task_toolset_count,
+            set_mcp_task_submitter,
+            validate_mcp_task_runtime_configuration,
+        )
+
+        task_extensions_config = ExtensionsConfig.from_file()
+        mcp_tasks_config = getattr(startup_config, "mcp_tasks", McpTasksConfig())
+        mcp_task_repo = getattr(app.state, "mcp_task_repo", None)
+        set_mcp_task_submitter(None)
+        validate_mcp_task_runtime_configuration(
+            mcp_tasks_config=mcp_tasks_config,
+            extensions_config=task_extensions_config,
+            repository_available=mcp_task_repo is not None,
+        )
+        if mcp_task_repo is not None:
+            mcp_task_drivers = McpTaskDriverRegistry()
+            if configured_task_toolset_count(task_extensions_config):
+                mcp_task_drivers.register(
+                    ORDINARY_MCP_TASK_DRIVER,
+                    OrdinaryMcpTaskDriver(McpTaskToolCaller(task_extensions_config)),
                 )
-                app.state.mcp_task_drivers = mcp_task_drivers
-                app.state.mcp_task_service = mcp_task_service
-                if startup_config.mcp_tasks.enabled:
-                    await mcp_task_service.start()
-        except Exception:
-            logger.exception("Failed to initialize MCP task service")
+            mcp_task_service = McpTaskService(
+                repository=mcp_task_repo,
+                drivers=mcp_task_drivers,
+                poll_interval_seconds=mcp_tasks_config.poll_interval_seconds,
+                lease_seconds=mcp_tasks_config.lease_seconds,
+                max_concurrent_polls=mcp_tasks_config.max_concurrent_polls,
+                max_poll_backoff_seconds=mcp_tasks_config.max_poll_backoff_seconds,
+                input_required_poll_interval_seconds=mcp_tasks_config.input_required_poll_interval_seconds,
+                tracking_degraded_after_errors=mcp_tasks_config.tracking_degraded_after_errors,
+                max_result_bytes=mcp_tasks_config.max_result_bytes,
+                result_preview_max_chars=mcp_tasks_config.result_preview_max_chars,
+            )
+            app.state.mcp_task_drivers = mcp_task_drivers
+            app.state.mcp_task_service = mcp_task_service
+            if mcp_tasks_config.enabled:
+                await mcp_task_service.start()
+                set_mcp_task_submitter(mcp_task_service)
 
         yield
 
@@ -383,6 +413,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
                 await app.state.mcp_task_service.stop()
             except Exception:
                 logger.exception("Failed to stop MCP task service")
+            finally:
+                from deerflow.mcp.tasks.runtime import set_mcp_task_submitter
+
+                set_mcp_task_submitter(None)
 
         try:
             from deerflow.community.browser_automation import get_browser_session_manager
@@ -657,6 +691,9 @@ def create_app() -> FastAPI:
     # MCP API is mounted at /api/mcp
     app.include_router(mcp.router)
 
+    # Durable MCP tasks are scoped to their owning thread.
+    app.include_router(mcp_tasks.router)
+
     # Memory API is mounted at /api/memory
     app.include_router(memory.router)
 

backend/app/gateway/routers/mcp.py

@@ -13,6 +13,7 @@
 from deerflow.config.extensions_config import (
     ExtensionsConfig,
     McpRoutingConfig,
+    McpTaskToolsetConfig,
     McpToolOverride,
     atomic_write_extensions_config,
     extensions_config_write_lock,
@@ -379,11 +380,21 @@ class McpServerConfigResponse(BaseModel):
     routing: McpRoutingConfig = Field(default_factory=McpRoutingConfig, description="Soft routing hints for tools from this MCP server")
     tools: dict[str, McpToolOverride] = Field(default_factory=dict, description="Per-original-tool MCP configuration overrides")
     tool_name_prefix: bool = Field(default=True, description="Whether to prefix discovered tool names with the MCP server name")
-    tool_call_timeout: float | None = Field(default=None, description="Timeout in seconds for individual stdio MCP tool calls")
+    tool_call_timeout: float | None = Field(
+        default=None,
+        description="Timeout in seconds for individual stdio MCP calls and durable-task calls on every transport",
+    )
     # Default matches McpServerConfig: this model's defaults feed model_dump()
     # into the persisted extensions config on PUT, so an API-created server that
     # omits the field must get the same bring-up timeout as a file-created one.
-    session_init_timeout: float | None = Field(default=DEFAULT_MCP_SESSION_INIT_TIMEOUT, description="Timeout in seconds for MCP server bring-up (tool discovery and persistent stdio session initialization); null means no timeout")
+    session_init_timeout: float | None = Field(
+        default=DEFAULT_MCP_SESSION_INIT_TIMEOUT,
+        description="Timeout in seconds for MCP server bring-up and durable HTTP/SSE task-session initialization; null means no timeout",
+    )
+    task_toolsets: list[McpTaskToolsetConfig] = Field(
+        default_factory=list,
+        description="Raw submit/status/cancel tool groups managed as durable background tasks",
+    )
     model_config = ConfigDict(extra="allow")
 
     @model_validator(mode="before")

backend/app/gateway/routers/mcp_tasks.py

@@ -0,0 +1,95 @@
+"""Thread-scoped read API for durable MCP background tasks."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from fastapi import APIRouter, HTTPException, Query, Request
+
+from app.gateway.authz import require_permission
+from app.gateway.deps import get_current_user, get_mcp_task_repo, get_mcp_task_service
+from deerflow.utils.thread_id import ThreadId
+
+router = APIRouter(prefix="/api/threads/{thread_id}/mcp-tasks", tags=["mcp-tasks"])
+
+_MAX_PUBLIC_ERROR_CHARS = 500
+
+
+def _short_error(value: Any) -> str | None:
+    if value is None:
+        return None
+    return str(value)[:_MAX_PUBLIC_ERROR_CHARS]
+
+
+def _tracking_degraded(record: dict[str, Any], *, threshold: int) -> bool:
+    return int(record.get("consecutive_poll_error_count") or 0) >= threshold
+
+
+def _list_item(record: dict[str, Any], *, threshold: int) -> dict[str, Any]:
+    return {
+        "task_id": record["id"],
+        "task_name": record["task_name"],
+        "status": record["status"],
+        "created_at": record["created_at"],
+        "updated_at": record["updated_at"],
+        "error": _short_error(record.get("error")),
+        "tracking_degraded": _tracking_degraded(record, threshold=threshold),
+    }
+
+
+def _detail(record: dict[str, Any], *, threshold: int) -> dict[str, Any]:
+    return {
+        **_list_item(record, threshold=threshold),
+        "last_polled_at": record.get("last_polled_at"),
+        "last_poll_error": _short_error(record.get("last_poll_error")),
+        "result": record.get("result"),
+        "result_preview": record.get("result_preview"),
+        "result_truncated": bool(record.get("result_truncated")),
+        "result_artifact": record.get("result_artifact"),
+        "input_required": record.get("input_required"),
+    }
+
+
+async def _current_user_id(request: Request) -> str:
+    user_id = await get_current_user(request)
+    if user_id is None:
+        raise HTTPException(status_code=401, detail="Authentication required")
+    return user_id
+
+
+@router.get("")
+@require_permission("threads", "read", owner_check=True)
+async def list_mcp_tasks(
+    thread_id: ThreadId,
+    request: Request,
+    limit: int = Query(default=50, ge=1, le=100),
+) -> list[dict[str, Any]]:
+    repository = get_mcp_task_repo(request)
+    service = get_mcp_task_service(request)
+    user_id = await _current_user_id(request)
+    records = await repository.list_by_thread(
+        thread_id,
+        user_id=user_id,
+        limit=limit,
+    )
+    threshold = service.tracking_degraded_after_errors
+    return [_list_item(record, threshold=threshold) for record in records]
+
+
+@router.get("/{task_id}")
+@require_permission("threads", "read", owner_check=True)
+async def get_mcp_task(
+    thread_id: ThreadId,
+    task_id: str,
+    request: Request,
+) -> dict[str, Any]:
+    repository = get_mcp_task_repo(request)
+    service = get_mcp_task_service(request)
+    user_id = await _current_user_id(request)
+    record = await repository.get(task_id, user_id=user_id)
+    if record is None or record["thread_id"] != thread_id:
+        raise HTTPException(status_code=404, detail="MCP task not found")
+    return _detail(
+        record,
+        threshold=service.tracking_degraded_after_errors,
+    )