feat(mcp): add ordinary durable task driver - #4690
Conversation
poll_attempt_count grows on every claim (successful polls included), so it cannot drive a failure backoff without misjudging normal long tasks. Add consecutive_poll_error_count: incremented when a claim is released after a poll error, reset to zero by any applied snapshot. The backoff/terminal policy that consumes it lands with the first concrete driver.
github-actions
Bot
added
area:backend
There was a problem hiding this comment.
Pull request overview
Adds the first concrete MCP durable-task “driver” by adapting explicitly configured ordinary submit / status / cancel MCP tools into a Gateway-managed, SQL-persisted background task lifecycle. This integrates with the existing durable task foundation (from the dependent PR) so long-running MCP work can be submitted without keeping remote task IDs in model-visible context, while task polling happens outside the Agent run and can recover after restarts.
Changes:
- Introduces
task_toolsetsMCP server configuration and tool wrapping: submit stays Agent-visible but becomes a durable background submit; status/cancel are hidden and called only by the background poller. - Implements the ordinary-tools task driver + background caller + poller service, including bounded result storage, exponential backoff, and lease-based multi-worker recovery.
- Adds DB migrations, thread-scoped task query endpoints, and supporting docs/examples/tests.
Reviewed changes
Copilot reviewed 51 out of 51 changed files in this pull request and generated 2 comments.
Show a summary per file
| File | Description |
|---|---|
| README.md | Documents durable ordinary MCP background tasks and the new thread-scoped query endpoints. |
| extensions_config.example.json | Adds an example MCP server entry demonstrating task_toolsets. |
| deploy/helm/deer-flow/values.yaml | Bumps Helm-embedded config_version to 34. |
| deploy/helm/deer-flow/README.md | Updates Helm README config snippet (but currently shows config_version: 33, which is inconsistent with 34). |
| config.example.yaml | Adds mcp_tasks configuration block and bumps config_version to 34. |
| backend/tests/test_persistence_bootstrap.py | Updates expected Alembic HEAD to 0012_mcp_task_results. |
| backend/tests/test_persistence_bootstrap_regression.py | Updates expected Alembic HEAD to 0012_mcp_task_results. |
| backend/tests/test_persistence_bootstrap_concurrency.py | Updates expected Alembic HEAD to 0012_mcp_task_results. |
| backend/tests/test_migration_0007_scheduled_run_active_dedupe.py | Updates expected Alembic HEAD to 0012_mcp_task_results. |
| backend/tests/test_migration_0004_run_ownership_dedupe.py | Updates expected Alembic HEAD to 0012_mcp_task_results. |
| backend/tests/test_mcp_tool_name_prefix.py | Updates MCP interceptor wiring patch target to build_mcp_tool_interceptors. |
| backend/tests/test_mcp_tasks_router.py | Adds coverage for the new thread-scoped MCP task list/detail routes and non-leakage of remote IDs. |
| backend/tests/test_mcp_task_toolset_config.py | Adds validation coverage for task_toolsets uniqueness and raw-name preservation. |
| backend/tests/test_mcp_task_tool_wrapping.py | Adds coverage for submit wrapping + status/cancel hiding behavior. |
| backend/tests/test_mcp_task_tool_caller.py | Adds coverage for background MCP tool calling (stdio scope reuse, HTTP auth init, reconnect eviction). |
| backend/tests/test_mcp_task_service.py | Adds coverage for durable submit/poll lifecycle, backoff, protocol failures, result bounding, and shutdown cancellation (contains a likely flaky timing assertion). |
| backend/tests/test_mcp_task_runtime_config.py | Adds startup validation coverage ensuring task toolsets don’t silently run synchronously. |
| backend/tests/test_mcp_task_repository.py | Adds coverage for SQL repository semantics (leases, terminalization, uniqueness, error counters). |
| backend/tests/test_mcp_task_ordinary_e2e.py | Adds e2e-ish coverage for ordinary driver submit → poll → restart recovery using SQLite persistence. |
| backend/tests/test_mcp_task_ordinary_driver.py | Adds coverage for ordinary submit/status/cancel contract parsing and status mapping. |
| backend/tests/test_mcp_task_models.py | Adds coverage for task model normalization and registry invariants. |
| backend/tests/test_mcp_task_config.py | Adds coverage for new McpTasksConfig defaults/bounds and startup-only registration. |
| backend/tests/test_mcp_session_pool.py | Adds coverage for close_session evicting only the targeted (server, scope) pair. |
| backend/packages/harness/deerflow/persistence/models/init.py | Exposes McpTaskRow in the persistence models package. |
| backend/packages/harness/deerflow/persistence/migrations/versions/0012_mcp_task_results.py | Adds bounded result preview/truncation/artifact columns for mcp_tasks. |
| backend/packages/harness/deerflow/persistence/migrations/versions/0011_mcp_tasks.py | Adds the mcp_tasks table and indexes/constraints. |
| backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py | Implements SQL repository (create/get/list/claim/apply/release) with lease and uniqueness semantics. |
| backend/packages/harness/deerflow/persistence/mcp_tasks/model.py | Adds McpTaskRow ORM model, including uniqueness constraint and indexes. |
| backend/packages/harness/deerflow/persistence/mcp_tasks/init.py | Exposes repository and error types for MCP tasks persistence. |
| backend/packages/harness/deerflow/mcp/tools.py | Adds task toolset wrapping (background submit + hiding status/cancel) and centralizes interceptor construction. |
| backend/packages/harness/deerflow/mcp/tasks/runtime.py | Adds process-local submitter bridge + startup validation for task runtime readiness. |
| backend/packages/harness/deerflow/mcp/tasks/ordinary.py | Implements the ordinary submit/status/cancel driver (structuredContent-only parsing, strict contract validation). |
| backend/packages/harness/deerflow/mcp/tasks/models.py | Adds protocol-neutral task models (statuses, snapshot, reference, submission, submit request). |
| backend/packages/harness/deerflow/mcp/tasks/driver.py | Adds driver protocol + registry. |
| backend/packages/harness/deerflow/mcp/tasks/init.py | Exports the task runtime public surface (driver, models, ordinary driver constants). |
| backend/packages/harness/deerflow/mcp/task_tool_caller.py | Adds background MCP tool caller with stdio session scope reuse and HTTP/SSE ephemeral sessions + OAuth/interceptors. |
| backend/packages/harness/deerflow/mcp/session_pool.py | Adds close_session helper for scoped stdio reconnection after failures. |
| backend/packages/harness/deerflow/mcp/oauth.py | Allows injecting an OAuthTokenManager into the OAuth tool interceptor builder. |
| backend/packages/harness/deerflow/mcp/interceptors.py | Centralizes building OAuth + custom MCP interceptors. |
| backend/packages/harness/deerflow/config/reload_boundary.py | Registers mcp_tasks as restart-required (startup-only) config. |
| backend/packages/harness/deerflow/config/mcp_tasks_config.py | Adds McpTasksConfig with bounded defaults for polling/backoff/result limits. |
| backend/packages/harness/deerflow/config/extensions_config.py | Adds McpTaskToolsetConfig and validates uniqueness of raw tool bindings across roles/groups. |
| backend/packages/harness/deerflow/config/app_config.py | Adds mcp_tasks field to AppConfig. |
| backend/docs/MCP_SERVER.md | Adds operator-facing documentation for ordinary durable background tasks and their contract. |
| backend/app/mcp_tasks/service.py | Implements McpTaskService for durable submit + background poll loop with backoff and bounded results. |
| backend/app/mcp_tasks/init.py | Exposes McpTaskService. |
| backend/app/gateway/routers/mcp.py | Includes task_toolsets in the MCP config response model. |
| backend/app/gateway/routers/mcp_tasks.py | Adds thread-scoped read APIs for MCP tasks (list/detail) with bounded fields and no remote ID leakage. |
| backend/app/gateway/deps.py | Wires McpTaskRepository and provides get_mcp_task_repo/get_mcp_task_service deps. |
| backend/app/gateway/app.py | Initializes task runtime at lifespan startup, registers ordinary driver, starts poller when enabled, and mounts task routes. |
| backend/AGENTS.md | Documents the new durable MCP task architecture and adds mcp_tasks to startup-only field list. |
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 50 out of 50 changed files in this pull request and generated no new comments.
Suppressed comments (2)
backend/tests/test_mcp_task_service.py:391
- This test uses
datetime.now(UTC)in the assertions after the poller runs, which can make the bounds flaky under load/slow CI. Capturing a singlenowand using it both forrun_onceand the comparisons makes the timing assertions deterministic.
backend/packages/harness/deerflow/mcp/tasks/ordinary.py:91 - When the MCP tool returns
isError=True, the raisedRuntimeErrordrops any server-provided error detail (typically incontent), solast_poll_errorand logs lose actionable context and retries become harder to debug.
def _structured_content(call_result: Any, *, tool_name: str) -> Any:
if bool(getattr(call_result, "isError", False)):
raise RuntimeError(f"MCP task tool {tool_name!r} returned an error")
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 51 out of 51 changed files in this pull request and generated no new comments.
Suppressed comments (1)
backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py:248
release_claim()clears the lease and pushesnext_poll_atforward even if the worker’s lease expired while the status call was in-flight. That means a slow/hung poll that eventually errors can still “win” after its lease expiry and delay recovery by other workers (and contradictsapply_snapshot()which discards results afterlease_expires_at). Add an expiry guard here too so stale poll errors don’t mutate scheduling/state after the lease has expired.
update(McpTaskRow)
.where(
McpTaskRow.id == task_id,
McpTaskRow.lease_owner == lease_owner,
)
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
Solid durable-task foundation: lease-based claiming with the lease_expires_at >= polled_at guard in apply_snapshot, strict structuredContent-only contract validation that maps task_not_found/malformed output to permanent FAILED, bounded result storage that stores a text preview rather than ever persisting truncated JSON, duplicate (user, server, remote) detection that refuses to cancel an already-tracked handle, and ownership-enforced read APIs that strip remote IDs/driver config. One suggestion below on the agent-facing submit wrapper.
|
|
||
| return StructuredTool( | ||
| name=tool.name, | ||
| description=(f"Start {task_name!r} as a durable background task. Returns a DeerFlow task ID immediately; status polling is handled automatically."), |
There was a problem hiding this comment.
Suggestion: this hard-coded description replaces the MCP server's original tool.description, so the agent loses the server-provided explanation of what this submit tool actually does (e.g. "generate a quarterly financial report with these parameters"). When multiple task_toolsets are configured on the same Gateway, the agent is left with only the local task_name plus the args-schema field hints to pick the right submit tool, which makes tool selection harder and weakens the case for configuring more than one toolset per server. Consider preserving or appending tool.description (e.g. f"{tool.description}\n\nSubmitted as durable background task '{task_name}'; returns a DeerFlow task ID immediately and status polling is handled automatically.") so the agent keeps the business context alongside the new durable-task return contract. The args_schema is already reused, so only the prose semantics are lost today.
There was a problem hiding this comment.
Good point—fixed in 52e96e9. The durable submit wrapper now preserves the original MCP tool description verbatim and appends the DeerFlow background-task contract, while continuing to reuse the original args schema and metadata. I added a regression test that failed before the implementation change (1 failed, 5 passed) and now passes (6 passed); the focused suite passes 513 tests, and format/lint also pass.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 51 out of 51 changed files in this pull request and generated no new comments.
Suppressed comments (2)
backend/packages/harness/deerflow/mcp/tasks/ordinary.py:91
- When an MCP tool call returns isError=true, _structured_content raises a generic RuntimeError without including any server-provided error detail. This makes poll failures hard to debug and results in unhelpful persisted last_poll_error strings. Consider appending a bounded snippet of returned text content (when available) to the exception message.
def _structured_content(call_result: Any, *, tool_name: str) -> Any:
if bool(getattr(call_result, "isError", False)):
raise RuntimeError(f"MCP task tool {tool_name!r} returned an error")
value = getattr(call_result, "structuredContent", None)
if value is None:
backend/packages/harness/deerflow/mcp/interceptors.py:56
- The interceptor load failure warning omits the exception message (it only shows a stack trace via exc_info). Including the exception text in the log message makes it easier to diagnose misconfigured mcpInterceptors in environments where stack traces are suppressed or truncated.
except Exception:
target_logger.warning(
f"Failed to load MCP interceptor {interceptor_path}",
exc_info=True,
)
willem-bd
left a comment
willem-bd
left a comment
There was a problem hiding this comment.
One new finding: the HTTP/SSE background poll path has no call-level timeout, so a single unresponsive remote server can stall the entire poll loop indefinitely. Stdio calls are bounded by tool_call_timeout/session_init_timeout, but the HTTP/SSE path explicitly drops tool_call_timeout and passes timeout_seconds=None into _invoke, and asyncio.gather(..., return_exceptions=True) in run_once still waits for every coroutine to finish before returning.
| async with create_session(effective_connection) as remote_session: | ||
| await remote_session.initialize() | ||
| try: | ||
| call_result = await remote_session.call_tool(request.name, request.args) |
There was a problem hiding this comment.
The non-persistent (HTTP/SSE) session path calls remote_session.initialize() and remote_session.call_tool() with no timeout. timeout_seconds is passed as None here (line 143), and tool_call_timeout is explicitly ignored for non-stdio transports (lines 126-131), so unlike the stdio branch there is no read_timeout_seconds bounding the call.
This has an operational consequence in McpTaskService.run_once (service.py:138): asyncio.gather(..., return_exceptions=True) still waits for every child coroutine to complete before returning - return_exceptions=True suppresses raised exceptions as return values, it does not abandon a coroutine that is still running. So one HTTP/SSE status poll that hangs on an unresponsive server stalls the entire scan, and _run_loop never reaches its asyncio.wait_for(self._stop.wait(), ...) interval sleep to schedule the next batch. The lease will eventually expire so another worker can reclaim the task, but this worker's poller is effectively dead until stop() cancels it.
Consider wrapping the HTTP/SSE initialize()/call_tool() calls in asyncio.wait_for with a configurable deadline (or reusing tool_call_timeout as a default rather than ignoring it), so a single unresponsive server cannot stall the background poller. The same gap affects the agent-facing submit path through call_tool, where an HTTP/SSE submit tool with no timeout would hang the Agent's tool call indefinitely.
There was a problem hiding this comment.
Fixed in e8a5ea8. HTTP/SSE durable-task calls now apply session_init_timeout to ephemeral-session initialization and tool_call_timeout to both the SDK read deadline and a caller-level deadline. This covers submit/status/cancel, so a hung remote call raises TimeoutError, lets the existing service error/backoff path run, and no longer blocks the whole poll batch indefinitely.
Added regression coverage for hung initialization and hung calls on both HTTP and SSE; all four tests failed before the fix and now pass. Validation: task caller tests 9 passed, all MCP tests 372 passed, format/lint passed, and the full backend suite reached 11,119 passed / 71 skipped with the same 15 environment-baseline browser/web-fetch failures already documented on this PR.
There was a problem hiding this comment.
Pull request overview
Copilot reviewed 51 out of 51 changed files in this pull request and generated no new comments.
Suppressed comments (1)
backend/packages/harness/deerflow/persistence/mcp_tasks/sql.py:248
release_claim()updates the row based only onlease_owner, so a worker that spent longer thanlease_secondsin a failing status call can still clear the lease and reschedulenext_poll_ateven though the lease has already expired. That violates the same “discard results after lease expiry” invariant thatapply_snapshot()enforces vialease_expires_at >= polled_at, and can delay recovery by another worker after a timeout/hang.
Gate release_claim() on an unexpired lease (and use a single captured now for updated_at) so stale workers can’t reschedule after expiry.
update(McpTaskRow)
.where(
McpTaskRow.id == task_id,
McpTaskRow.lease_owner == lease_owner,
)
Part of #4652
Depends on #4665.
Why
PR #4665 establishes the protocol-neutral persistence and polling foundation for long-running MCP work, but it deliberately does not connect any real MCP tools. Existing MCP calls therefore remain synchronous, and the durable task runtime is not yet directly usable.
This PR adds the first concrete driver: an explicitly configured ordinary
submit/status/cancelcontract. DeerFlow persists the remote handle before returning, exposes only a local task ID to the Agent, and polls status outside the Agent run. This makes long-running MCP work usable without asking the model to remember remote IDs or spend context on polling loops.This is the second implementation step. It adds backend task queries, but it does not yet wake the Agent on completion, expose natural-language cancellation, or add the frontend task panel.
What changed
task_toolsetsto each MCP server configuration. Bindings use exact raw tool names, support multiple groups per server, and reject a raw tool reused across roles or groups.structuredContent, strictly validates the fixed submit/status/cancel contract, maps remote lifecycle states, and treatserror_code: task_not_foundor malformed structured output as permanent failures.(server_name, user_id:thread_id)stdio session scope, exact raw tool names, the configured session-initialization timeout, scoped reconnect after a broken session, ephemeral HTTP/SSE sessions, shared interceptors, and server-level OAuth for both session initialization and calls.input_required, derivedtracking_degradedreporting, strict JSON result validation, bounded result previews, and preserved external result artifacts.result_preview,result_truncated, andresult_artifact, plus(user_id, server_name, remote_task_id)duplicate handling that never cancels an already tracked remote task./api/threads/{thread_id}/mcp-tasks. Responses enforce user/thread ownership and never expose remote task IDs or driver configuration.Not in this PR: Agent completion wake-up, natural-language or frontend cancellation, ThreadState projection,
input_requiredresume, a frontend task panel, or a SEP-2663 driver.Surface area
frontend/langgraph.json, or prompt changedocker/or sandboxed executionskills/mcp_tasks.enabledandtask_toolsetsconfigurationScreenshots / Recording
Not applicable; this PR has no frontend change.
Validation
cd backend && make format— passedcd backend && make lint— passedbash scripts/check_config_version.sh— passed (config_version=34in both the root example and Helm values)helm lint deploy/helm/deer-flow— 1 chart linted, 0 failedmainand documented in feat(mcp): add durable task runtime foundation #4665; no PR2 code path failed.AI assistance
Tool(s) used: Codex
How you used it: Codex traced the existing MCP discovery, session pooling, OAuth, Gateway lifecycle, persistence, migration, and authorization boundaries; implemented the approved PR 2 scope; added regression and fake-server recovery coverage; ran focused, full-suite, Helm, formatting, and lint validation; performed duplicate checks; and prepared this PR description. I directed the scope and reviewed the architecture and validation results.