diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 9e4a75a2f4..dc691794e2 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -21,7 +21,7 @@ and can be removed by LRU pressure or explicit cleanup. This cannot close the transport after its loop is gone. Normal `asyncio.run()` drains its owners. - **Persistent stdio session capacity**: `MCPSessionPool.MAX_SESSIONS` is a hard cap on the live LRU registry. Capacity is enforced both before creating a session and when an in-flight session is promoted, because different keys can finish initialization concurrently after all observing spare capacity. Promotion-time victims are signalled and drained by separately tracked teardown tasks on their owning loops; the newly promoted owner must never await a victim, so a blocked victim exit cannot prevent its own closure or disconnect recovery. -- **Cache invalidation**: Detects extensions-config changes by comparing the resolved config path and a `(mtime, size, sha256)` content signature against the values recorded at initialization, not a strict mtime `>` comparison. This catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. If `initialize_mcp_tools()` itself observes a config-signature change between the pre-load and post-load snapshots, that discard branch must reset the tool cache through the same session-pool retirement path as normal stale invalidation before waiters retry; otherwise a stale load can leave `(server_name, scope_key)` sessions from the abandoned connection available to the next wrapper build. +- **Cache invalidation**: Staleness is detected in two stages. First, the resolved config path and a `(mtime, size, sha256)` content signature are compared against the values recorded when the cache was published, not a strict mtime `>` comparison — this catches same-second edits, mtime that stays put or moves backward (`git checkout`, `cp -p` / backup restore, `tar` / `rsync`, object-store / network mounts), and a switch to a different config file with an equal-or-older mtime. The signature helper (`config/file_signature.py::get_config_signature`) is shared with `config/app_config.py::get_app_config()` for the sibling runtime-editable config file, rather than each maintaining its own copy. Second, because the file also carries `skills` and middleware settings, a changed signature alone is not enough: `mcp/cache.py::_effective_mcp_config_snapshot()` serializes the effective MCP slice (enabled servers' parsed config in declaration order plus `mcpInterceptors`; the raw `transport` alias is dropped so equivalent `type`/`transport` spellings compare equal) and the cache is retired only when that slice changed. A skills-only or middleware-only edit therefore adopts the new signature and keeps the cached tools and pooled sessions, while any real MCP change keeps the existing process-wide reset through `_reset_mcp_tools_cache_state_and_retire_pool_locked()`. The snapshot may embed resolved credentials, so it stays in process memory and is never logged or persisted. `_is_cache_stale()` adopts a new signature when the MCP slice is unchanged, so it is not a read-only predicate; production callers run it under `_init_condition`. `ExtensionsConfig.resolve_config_path()` raises `FileNotFoundError` for an explicit `config_path`/`DEER_FLOW_EXTENSIONS_CONFIG_PATH` that points at a missing file — an operator-asserted path going missing is a real misconfiguration, so this is intentionally loud for callers that load the config for actual use (e.g. `from_file()` via `get_mcp_tools()`); only the fallback search mode returns `None`. The MCP cache's own path resolution (`mcp/cache.py::_resolve_config_path`) is narrower: it catches that specific `FileNotFoundError` locally and treats it the same as "unconfigured", so this staleness check degrades to "not stale" instead of propagating an exception when a previously-valid explicit/env-var config disappears mid-run. Two no-config states are distinguished: `_initialized_without_config` makes a cache published while no config existed stale as soon as one appears, while a config deleted after a successful load keeps the last-known-good fail-soft behavior and stays "not stale", while a file that is present but whose content cannot be hashed is treated as stale instead, because the cache cannot prove the revision it published is still the current one. `initialize_mcp_tools()` reads one `ExtensionsConfig` instance, hands that exact object to `get_mcp_tools(extensions_config=...)`, and publishes only when the effective MCP snapshot of the current config still equals the snapshot of the revision that produced the tools; otherwise it resets the tool cache through the same session-pool retirement path as normal stale invalidation before waiters retry. This closes the discovery window — tools built from a superseded MCP revision, or from a resolved path whose signature cannot be re-read, are never published — while a skills-only edit during discovery still publishes because the MCP slice is unchanged. - **Transports**: stdio (command-based), SSE, HTTP - **Per-server tool-name prefixing**: `mcpServers..tool_name_prefix` defaults to `true`, preserving the collision-safe `_` prefix. Servers whose tools already carry a stable namespace may set it to `false`; discovery then calls `langchain_mcp_adapters.tools.load_mcp_tools` with that server's flag. Source routing and stdio session-pool wrapping are based on the producing server and transport, never on whether the visible tool name starts with the server prefix. - **OAuth (HTTP/SSE)**: Supports token endpoint flows (`client_credentials`, `refresh_token`) with automatic token refresh + Authorization header injection. The rendered ` ` passes `mcp/headers.py::illegal_header_value_reason` inside `OAuthTokenManager.get_authorization_header` — the one boundary the tool interceptor, the initial discovery headers and the durable task path all read their value from — so a token endpoint returning something the transport would refuse fails closed instead of letting h11 echo the token into a model-visible tool error. The rendered value is what gets checked, not the two fields separately, because that is what the transport sees. The operator's static `headers` get the same check in `mcp/client.py::build_server_params`, where `build_servers_config` already drops just that server and logs the reason. @@ -41,7 +41,7 @@ - **Configured stdio working directory**: A non-empty `McpServerConfig.cwd` is forwarded by `build_server_params` for discovery, pooled Agent calls, and durable task calls. Omit the connection key when the setting is `None` or empty (including unresolved environment references), so the runtime's thread-workspace default still applies; HTTP/SSE connections must not receive it. `tests/test_mcp_cwd.py` exercises the JSON-config-to-subprocess path with relative entrypoints, relative tool inputs, and discovery/pool defaults for omitted, null, empty, and unresolved values. - **Stdio disconnect recovery**: Ordinary Agent tool calls and durable task submit/status/cancel calls that receive the MCP SDK's explicit `Connection closed` error or an AnyIO closed-stream error evict only that `(server_name, incarnation scope)` session when the registered entry is still the same `ClientSession` that failed. A late error from an old concurrent call cannot evict its replacement or a new in-flight creation. The failing call still surfaces its original error and is never replayed automatically; a later retry creates a fresh subprocess/session. Protocol timeouts, normal `isError=true` tool results, and interceptor failures do not evict a healthy stateful session. - **Stdio path translation**: MCP-returned local file references are not copied. If a `ResourceLink` or conservative free-text path resolves to an existing file inside the thread's mounted user-data tree, it is translated deterministically to `/mnt/user-data/...`; paths outside that tree remain unchanged. -- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it; either endpoint's reset clears the cache only in its own worker). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. +- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it; either endpoint's reset clears the cache only in its own worker). MCP, skill, and embedded-client writers hold the process-local `extensions_config_write_lock` plus the sidecar advisory `extensions_config_file_lock` for the complete read-modify-write/reload cycle, then share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file. Runtime updates are picked up either by the writer's explicit `reset_mcp_tools_cache()` or by another worker's two-stage staleness check above; a change that leaves the effective MCP slice unchanged (skills-only, middleware-only) no longer retires cached tools or pooled sessions, and when no MCP server is enabled the tool-assembly path calls `refresh_mcp_cache_if_active()`, which retires an already-initialized cache and pool (and voids an in-flight initialization) without performing new tool discovery or config hashing on a never-initialized process. The Gateway admin reset (`POST /api/mcp/cache/reset`) is not part of that comparison: calling `reset_mcp_tools_cache()` always retires the cache and pool regardless of whether the effective MCP slice changed. - **Stdio launch policy at the HTTP boundary** (`routers/mcp.py::_validate_mcp_update_request`, shared by `PUT` and the enable branch of `PATCH`): a config file may express anything, but the API is untrusted input, so an API-registered stdio server must (a) name a bare executable from the allowlist — `_DEFAULT_MCP_STDIO_COMMAND_ALLOWLIST` = `{npx, uvx}`, extended by `DEER_FLOW_MCP_STDIO_COMMAND_ALLOWLIST`, with path separators, whitespace, and shell metacharacters rejected in `command`; (b) carry no `args` flag in `_ARBITRARY_EXEC_ARGS`; and (c) set no `env` name in `_CODE_INJECTING_ENV_VARS`. Checks (b) and (c) exist because the command check alone names a binary without constraining what that binary runs. The `env` denylist applies to **every** allowlisted command, and both denylists match `--flag=value` as well as `--flag value`. The `args` denylist's **scope depends on the command**, because where a launcher stops parsing its own flags is what decides whether a token is an exec flag at all: - For a **package launcher** in `_PACKAGE_LAUNCHERS` (`{npx, uvx}`) only the launcher's own **option region** is screened. `npx`/`uvx` stop parsing their flags at the package name and hand every later token to the spawned server's argv, where `-c` is routinely "config" and `-e` "env" — screening those rejected ordinary third-party servers while covering nothing. A bare `--` ends the region too: only the *first* token after it is the package name. Finding that boundary needs each launcher's option **arity**, since a value is not a positional — `npx -p -c ''` **runs** the command (`-p` is `npm exec`'s `--package`, so `` is its value and npm keeps parsing), so ending the region at the first non-flag token would walk straight past it. `_NPX_BOOLEAN_ARGS` is generated from `@npmcli/config`'s definitions (npm 10.9.4) minus the `-p` exec override; `_UVX_VALUE_ARGS` comes from `uvx --help` (uv 0.11.1). Regenerate these against a newer launcher rather than hand-editing. The unknown-option default is deliberately **opposite** per launcher, following the exec set rather than symmetry: npx owns real exec flags (`-c`/`--call`), so an unknown option consumes a value and keeps the region open (npm errors on options it does not define, so this cannot reject a working invocation); uvx owns no string-eval flag at all, so its screen is a tripwire, an unknown option consumes nothing, and uv's large boolean surface cannot over-block. uvx's exec set also drops the short spellings, because `-c` is uv's `--constraints` and `-p` its `--python`. diff --git a/backend/packages/harness/deerflow/mcp/cache.py b/backend/packages/harness/deerflow/mcp/cache.py index b02b385b7b..d05b532e68 100644 --- a/backend/packages/harness/deerflow/mcp/cache.py +++ b/backend/packages/harness/deerflow/mcp/cache.py @@ -1,6 +1,7 @@ """Cache for MCP tools to avoid repeated loading.""" import asyncio +import json import logging import threading from pathlib import Path @@ -9,6 +10,7 @@ from langchain_core.tools import BaseTool from deerflow.config.file_signature import ConfigSignature as _ConfigSignature from deerflow.config.file_signature import get_config_signature as _get_config_signature +from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, normalize_mcp_server_config logger = logging.getLogger(__name__) @@ -31,6 +33,17 @@ _cache_generation = 0 _config_path: Path | None = None # Resolved extensions config path at init time _config_signature: _ConfigSignature | None = None # (mtime, size, sha256) at init time +# JSON snapshot of the effective MCP slice (enabled servers in declaration +# order + mcpInterceptors) that the currently published tools were built from. +# May contain resolved credentials: never log or persist it. +_mcp_config_snapshot: str | None = None + +# True when the published cache came from an initialization with no resolvable +# extensions config. Distinguishes "never configured" (a later config must be +# picked up) from "config deleted after a successful load" (fail-soft: keep +# serving the last-known-good tools). +_initialized_without_config = False + def _resolve_config_path() -> Path | None: """Resolve the extensions config file path, or ``None`` when unconfigured. @@ -76,18 +89,99 @@ def _current_config_state() -> tuple[Path | None, _ConfigSignature | None]: return config_path, _get_config_signature(config_path) +def _effective_mcp_config_snapshot(config) -> str: + """Serialize the MCP-only slice of an extensions config. + + ``extensions_config.json`` also carries skills and middleware settings, so + the whole-file signature cannot distinguish an MCP change from a skill + toggle. The enabled-server list preserves declaration order (it can affect + tool ordering) while ``sort_keys`` only normalizes each server's field + order. Parsed models are compared through + ``config_normalization.normalize_mcp_server_config`` (equivalent + ``type``/``transport`` spellings) and the custom interceptor list through + ``config_normalization.normalize_mcp_interceptor_paths`` (a bare string and + its single-element list, or a missing key and an empty list), so equivalent + spellings do not cause a needless rebuild. + + The result may embed resolved credentials. It stays in process memory and + is never logged or written back to disk. + """ + relevant = { + "enabled_servers": [(name, normalize_mcp_server_config(server)) for name, server in config.get_enabled_mcp_servers().items()], + "mcpInterceptors": normalize_mcp_interceptor_paths((config.model_extra or {}).get("mcpInterceptors")), + } + return json.dumps(relevant, sort_keys=True, ensure_ascii=False) + + +def _signature_is_verifiable(signature: _ConfigSignature | None) -> bool: + """True when a config signature carries a content digest. + + ``get_config_signature`` returns ``(mtime, size, None)`` when the file could + be stat-ed but not read. Such a signature cannot prove the content is + unchanged, so the MCP cache must neither publish nor adopt it. + """ + return signature is not None and signature[2] is not None + + +def _read_stable_mcp_snapshot(config_path: Path, expected_signature: _ConfigSignature) -> str | None: + """Parse the config and return its MCP snapshot only if the file was stable. + + ``expected_signature`` is the signature observed by the caller. A signature + without a content digest is treated as unverifiable, and the file is + re-hashed after parsing: a mismatch means the config changed while it was + being read, so no snapshot can be attributed to a single revision and the + caller must treat the cache as stale. Parse failures also return ``None`` + (conservative: never reuse tools on an unreadable config). + """ + if not _signature_is_verifiable(expected_signature): + logger.info("Extensions config signature has no content digest; treating the MCP cache as stale") + return None + + from deerflow.config.extensions_config import ExtensionsConfig + + try: + config = ExtensionsConfig.from_file(str(config_path)) + except Exception as exc: + # Do NOT pass exc_info/message: ExtensionsConfig.from_file resolves + # ``$VAR`` values before validation, so a ValidationError can embed + # resolved credentials in its input. Only the exception type is safe. + logger.warning( + "Could not parse extensions config while checking MCP cache staleness (%s); treating the cache as stale", + type(exc).__name__, + ) + return None + + current_signature = _get_config_signature(config_path) + if not _signature_is_verifiable(current_signature) or current_signature != expected_signature: + logger.info("Extensions config changed while it was being read; treating the MCP cache as stale") + return None + + return _effective_mcp_config_snapshot(config) + + def _is_cache_stale() -> bool: """Check if the cache is stale due to config file changes. - The cache is stale when the resolved extensions config path changed, or when - the ``(mtime, size, sha256)`` content signature differs from the one recorded - at initialization. Using content equality (``!=``) instead of a strict mtime - ``>`` comparison detects same-second edits and backward mtime moves, and - tracking the resolved path detects a switch to a different config file. + The cache is stale when the effective MCP slice of the resolved config + differs from the one recorded at initialization. The resolved path and the + ``(mtime, size, sha256)`` content signature are the change signals that + trigger that comparison, not invalidation reasons on their own. Using + content equality (``!=``) instead of a strict mtime ``>`` comparison detects + same-second edits and backward mtime moves. A path switch to a file with the + same MCP configuration, or an edit that only touches skills or middleware + settings, therefore keeps the cached tools and sessions and adopts the new + path and signature. + + When the file signature changes but the effective MCP configuration does + not, this function adopts the new signature into ``_config_signature`` and + returns False. It is therefore not a read-only predicate; production callers + invoke it under ``_init_condition``, the same lock the mutating paths use. Returns: True if the cache should be invalidated, False otherwise. """ + global _config_path, _config_signature + if not _cache_initialized: return False # Not initialized yet, not stale @@ -103,18 +197,38 @@ def _is_cache_stale() -> bool: # longer be stat-ed). Treat this as a deliberate fail-soft choice, not an # oversight — a future change that wants "config deleted" to tear down # MCP tools needs its own explicit signal here, not an inferred one. - if _config_signature is None or current_signature is None: + if _config_signature is None: + # A config that appears after an unconfigured initialization must be + # picked up; a config deleted after a successful load keeps the + # last-known-good fail-soft contract below. + if _initialized_without_config and current_signature is not None: + logger.info("Extensions config appeared after an unconfigured MCP cache; cache is stale") + return True return False + if current_signature is None: + return False + + if current_path == _config_path and current_signature == _config_signature and _signature_is_verifiable(current_signature): + return False + + # The resolved path and/or the file content changed. Both are only signals to + # re-read the effective MCP slice: the file also carries skills and middleware + # settings, and a path switch can land on a file with the same MCP + # configuration, so neither is an invalidation reason on its own. if current_path != _config_path: - logger.info("MCP config path changed (%s -> %s), cache is stale", _config_path, current_path) - return True + logger.info("MCP config path changed (%s -> %s); re-checking the effective MCP configuration", _config_path, current_path) - if current_signature != _config_signature: - logger.info("MCP config content changed (signature %s -> %s), cache is stale", _config_signature, current_signature) - return True + if current_path is not None and _mcp_config_snapshot is not None: + current_snapshot = _read_stable_mcp_snapshot(current_path, current_signature) + if current_snapshot is not None and current_snapshot == _mcp_config_snapshot: + logger.info("Extensions config changed but the effective MCP configuration did not; keeping cached MCP tools and sessions") + _config_path = current_path + _config_signature = current_signature + return False - return False + logger.info("MCP config content changed (signature %s -> %s), cache is stale", _config_signature, current_signature) + return True def _wait_for_initialization(generation: int | None) -> None: @@ -132,7 +246,7 @@ async def initialize_mcp_tools() -> list[BaseTool]: List of LangChain tools from all enabled MCP servers. """ global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature - global _initializing_generation, _cache_generation + global _initializing_generation, _cache_generation, _mcp_config_snapshot, _initialized_without_config while True: with _init_condition: @@ -141,7 +255,6 @@ async def initialize_mcp_tools() -> list[BaseTool]: return _mcp_tools_cache or [] if _initializing_generation is None: - pre_path, pre_sig = _current_config_state() claim_generation = _cache_generation _initializing_generation = claim_generation break @@ -150,16 +263,56 @@ async def initialize_mcp_tools() -> list[BaseTool]: await asyncio.to_thread(_wait_for_initialization, waiting_generation) + from deerflow.config.extensions_config import ExtensionsConfig from deerflow.mcp.tools import get_mcp_tools loaded_tools = None + loaded_snapshot = None post_path = None post_sig = None + post_snapshot = None init_succeeded = False try: logger.info("Initializing MCP tools...") - loaded_tools = await get_mcp_tools() + # Read the exact revision we hand to discovery. Comparing pre/post file + # snapshots alone cannot prove which revision produced the tools, because + # get_mcp_tools() would otherwise read the file itself. + try: + loaded_config = ExtensionsConfig.from_file() + except Exception as exc: + # Never let a resolved-credential ValidationError reach a caller's + # logger: from_file() resolves $VAR values before validation, so the + # exception message can embed secrets. Re-raise a sanitized error; + # other MCP failures keep their original traceback. + logger.warning( + "Could not load extensions config before MCP tool discovery (%s); aborting initialization", + type(exc).__name__, + ) + raise RuntimeError("Extensions config could not be loaded for MCP tool discovery") from None + loaded_snapshot = _effective_mcp_config_snapshot(loaded_config) + loaded_tools = await get_mcp_tools(extensions_config=loaded_config) post_path, post_sig = _current_config_state() + if post_path is not None and post_sig is not None: + post_snapshot = _read_stable_mcp_snapshot(post_path, post_sig) + elif post_path is not None: + # The path resolved but its signature could not be read. Publishing + # here would record an unpinned cache that later checks could never + # invalidate, so discard instead. + post_snapshot = None + else: + # No resolvable config now. Re-resolve after the fallback read so a + # config that appeared mid-flight is not published as an unpinned + # cache; the snapshot comparison still rejects a different revision. + try: + fallback_snapshot = _effective_mcp_config_snapshot(ExtensionsConfig.from_file()) + except Exception as exc: + logger.warning( + "Could not load extensions config after MCP tool discovery (%s); discarding result", + type(exc).__name__, + ) + fallback_snapshot = None + recheck_path, recheck_sig = _current_config_state() + post_snapshot = fallback_snapshot if recheck_path is None and recheck_sig is None else None init_succeeded = True finally: if not init_succeeded: @@ -175,13 +328,16 @@ async def initialize_mcp_tools() -> list[BaseTool]: logger.info("MCP cache was reset during initialization; discarding stale result") return [] - if (pre_path, pre_sig) != (post_path, post_sig): + publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot + if not publish: logger.warning("MCP config changed during initialization; discarding stale result") retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked() else: _mcp_tools_cache = loaded_tools _cache_initialized = True _config_path, _config_signature = post_path, post_sig + _mcp_config_snapshot = post_snapshot + _initialized_without_config = post_path is None logger.info("MCP tools initialized: %d tool(s) loaded (config path: %s)", len(_mcp_tools_cache), _config_path) return _mcp_tools_cache finally: @@ -250,15 +406,51 @@ def get_cached_mcp_tools() -> list[BaseTool]: return _mcp_tools_cache or [] +def refresh_mcp_cache_if_active() -> bool: + """Retire stale MCP cache state without lazily initializing tools. + + Tool assembly skips ``get_cached_mcp_tools()`` when no MCP server is + enabled, so a config change that disables the last server would otherwise + leave the previous pool and its persistent sessions alive. This entry point + performs only the staleness check: + + * it returns immediately when no MCP state was ever initialized and no + initialization is in flight, so a process that never initialized MCP + tools pays no config-hashing cost. A process that did publish a cache + (including an empty one) still pays one stat+sha256 of the config file + per call to check that cache for staleness; + * it invalidates an in-flight initialization (bumping the cache generation) + so tools discovered under the superseded config cannot publish; + * it retires the pool outside the lock, matching ``get_cached_mcp_tools``. + + Returns: + True when existing cache state or an in-flight initialization was + retired. + """ + retired_pool = None + retired = False + with _init_condition: + if not _cache_initialized and _initializing_generation is None: + return False + if _initializing_generation is not None or _is_cache_stale(): + retired_pool = _reset_mcp_tools_cache_state_and_retire_pool_locked() + retired = True + if retired_pool is not None: + retired_pool.close_all_sync() + return retired + + def _reset_mcp_tools_cache_state() -> None: """Reset cache state under ``_init_condition`` / ``_init_lock``.""" global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature - global _cache_generation + global _cache_generation, _mcp_config_snapshot, _initialized_without_config _mcp_tools_cache = None _cache_initialized = False _config_path = None _config_signature = None + _mcp_config_snapshot = None + _initialized_without_config = False _cache_generation += 1 _init_condition.notify_all() diff --git a/backend/packages/harness/deerflow/mcp/config_normalization.py b/backend/packages/harness/deerflow/mcp/config_normalization.py new file mode 100644 index 0000000000..95241cc382 --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/config_normalization.py @@ -0,0 +1,46 @@ +"""Shared normalization of MCP configuration values. + +The MCP tools cache, the custom-interceptor builder and the durable-task +configuration snapshot must all agree on when two configs are equivalent. +Keeping the rules here stops those three call sites from drifting apart. +""" + +from __future__ import annotations + +from typing import Any + +# Server fields that only affect how tools are presented or routed to the +# model, never how a durable-task server is contacted or driven. +TASK_PRESENTATION_FIELDS = ("description", "routing", "tools", "tool_name_prefix") + + +def normalize_mcp_interceptor_paths(raw: Any) -> list[Any]: + """Return the custom interceptor paths a raw ``mcpInterceptors`` value selects. + + Mirrors what ``build_mcp_tool_interceptors`` consumes: a bare string is one + path, a list is used verbatim (order and duplicates preserved), and any + other value — including a missing key and ``None`` — selects no custom + interceptors. + """ + if isinstance(raw, str): + return [raw] + if isinstance(raw, list): + return list(raw) + return [] + + +def normalize_mcp_server_config(server: Any, *, task_runtime: bool = False) -> dict[str, Any]: + """Return a parsed server config with alias-only fields normalized away. + + ``transport`` is an MCP-spec alias for ``type``: the model validator copies + it onto ``type`` but keeps the raw extra key (``extra="allow"``), so dropping + it makes equivalent ``type``/``transport`` spellings compare equal. With + ``task_runtime=True`` the presentation-only fields are dropped too, matching + what the durable-task runtime actually depends on. + """ + dumped = server.model_dump(mode="json") + dumped.pop("transport", None) + if task_runtime: + for field in TASK_PRESENTATION_FIELDS: + dumped.pop(field, None) + return dumped diff --git a/backend/packages/harness/deerflow/mcp/interceptors.py b/backend/packages/harness/deerflow/mcp/interceptors.py index 05a52fa11f..3863c04fe9 100644 --- a/backend/packages/harness/deerflow/mcp/interceptors.py +++ b/backend/packages/harness/deerflow/mcp/interceptors.py @@ -6,6 +6,7 @@ import logging from typing import Any from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths from deerflow.mcp.context_headers import build_context_headers_interceptor from deerflow.mcp.oauth import build_oauth_tool_interceptor from deerflow.mcp.user_scoped_auth import build_user_scoped_auth_interceptor @@ -44,16 +45,13 @@ def build_mcp_tool_interceptors( if context_headers_interceptor is not None: interceptors.append(context_headers_interceptor) - raw_paths = (extensions_config.model_extra or {}).get("mcpInterceptors") - if isinstance(raw_paths, str): - raw_paths = [raw_paths] - elif not isinstance(raw_paths, list): - if raw_paths is not None: - target_logger.warning( - "mcpInterceptors must be a list of strings, got %s; skipping", - type(raw_paths).__name__, - ) - raw_paths = [] + raw_value = (extensions_config.model_extra or {}).get("mcpInterceptors") + if raw_value is not None and not isinstance(raw_value, (str, list)): + target_logger.warning( + "mcpInterceptors must be a list of strings, got %s; skipping", + type(raw_value).__name__, + ) + raw_paths = normalize_mcp_interceptor_paths(raw_value) for interceptor_path in raw_paths: try: diff --git a/backend/packages/harness/deerflow/mcp/tasks/runtime.py b/backend/packages/harness/deerflow/mcp/tasks/runtime.py index d118fe70d9..1fde83b883 100644 --- a/backend/packages/harness/deerflow/mcp/tasks/runtime.py +++ b/backend/packages/harness/deerflow/mcp/tasks/runtime.py @@ -5,6 +5,7 @@ from __future__ import annotations from typing import Any, Protocol from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp.config_normalization import normalize_mcp_interceptor_paths, normalize_mcp_server_config from deerflow.mcp.tasks.models import TaskSubmitRequest @@ -51,11 +52,9 @@ def _task_server_configs(extensions_config: ExtensionsConfig) -> _TaskServerConf for server_name, server in extensions_config.get_enabled_mcp_servers().items(): if not server.task_toolsets: continue - runtime_config = server.model_dump(mode="json") - for presentation_field in ("description", "routing", "tools", "tool_name_prefix"): - runtime_config.pop(presentation_field, None) - servers[server_name] = runtime_config - interceptors = (extensions_config.model_extra or {}).get("mcpInterceptors") if servers else None + servers[server_name] = normalize_mcp_server_config(server, task_runtime=True) + raw_interceptors = (extensions_config.model_extra or {}).get("mcpInterceptors") + interceptors = normalize_mcp_interceptor_paths(raw_interceptors) if servers else None return servers, interceptors diff --git a/backend/packages/harness/deerflow/mcp/tools.py b/backend/packages/harness/deerflow/mcp/tools.py index 78a4d02f81..f6b2d9603c 100644 --- a/backend/packages/harness/deerflow/mcp/tools.py +++ b/backend/packages/harness/deerflow/mcp/tools.py @@ -780,7 +780,7 @@ def _configure_task_tools_for_server( return configured -async def get_mcp_tools() -> list[BaseTool]: +async def get_mcp_tools(extensions_config: ExtensionsConfig | None = None) -> list[BaseTool]: """Get all tools from enabled MCP servers. Tools using stdio transport are wrapped with persistent-session logic so @@ -788,6 +788,11 @@ async def get_mcp_tools() -> list[BaseTool]: HTTP/SSE tools are returned unwrapped to avoid cross-task TaskGroup cleanup errors. + Args: + extensions_config: Optional pre-loaded extensions config. Callers that + must prove which config revision produced these tools pass the exact + instance they snapshotted; ``None`` loads the latest config from disk. + Returns: List of LangChain tools from all enabled MCP servers. """ @@ -798,11 +803,13 @@ async def get_mcp_tools() -> list[BaseTool]: logger.warning("langchain-mcp-adapters not installed. Install it to enable MCP tools: pip install langchain-mcp-adapters") return [] - # NOTE: We use ExtensionsConfig.from_file() instead of get_extensions_config() - # to always read the latest configuration from disk. This ensures that changes - # made through the Gateway API (which runs in a separate process) are immediately - # reflected when initializing MCP tools. - extensions_config = ExtensionsConfig.from_file() + if extensions_config is None: + # NOTE: We use ExtensionsConfig.from_file() instead of get_extensions_config() + # to always read the latest configuration from disk. This ensures that changes + # made through the Gateway API (which runs in a separate process) are immediately + # reflected when initializing MCP tools. Callers that need to prove which + # revision produced these tools pass the instance they snapshotted instead. + extensions_config = ExtensionsConfig.from_file() validate_mcp_task_config_snapshot(extensions_config) servers_config = build_servers_config(extensions_config) diff --git a/backend/packages/harness/deerflow/tools/tools.py b/backend/packages/harness/deerflow/tools/tools.py index 551171f513..d894a05d56 100644 --- a/backend/packages/harness/deerflow/tools/tools.py +++ b/backend/packages/harness/deerflow/tools/tools.py @@ -233,29 +233,45 @@ def get_available_tools( if include_mcp: try: from deerflow.config.extensions_config import ExtensionsConfig - from deerflow.mcp.cache import get_cached_mcp_tools + from deerflow.mcp.cache import get_cached_mcp_tools, refresh_mcp_cache_if_active - extensions_config = ExtensionsConfig.from_file() - if extensions_config.get_enabled_mcp_servers(): - mcp_tools = get_cached_mcp_tools() - if mcp_tools: - logger.info(f"Using {len(mcp_tools)} cached MCP tool(s)") + try: + extensions_config = ExtensionsConfig.from_file() + except Exception as exc: + # Only this call carries the resolved-credential risk: + # from_file() resolves $VAR values before validation, so a + # ValidationError message can embed secrets. Log the type only. + logger.error("Failed to load MCP extensions config (%s)", type(exc).__name__) + else: + if extensions_config.get_enabled_mcp_servers(): + mcp_tools = get_cached_mcp_tools() + if mcp_tools: + logger.info(f"Using {len(mcp_tools)} cached MCP tool(s)") - # Tag MCP-sourced tools so deferred-tool assembly at each - # agent construction site can identify them. Lead agents - # assemble their full configured MCP catalog and apply active - # skill policy at runtime; subagents may pass an already - # policy-filtered list because their skills load at startup. - for t in mcp_tools: - tag_mcp_tool(t) - if mcp_plugins is not None: - from deerflow.capabilities.runtime import filter_mcp_plugins + # Tag MCP-sourced tools so deferred-tool assembly at each + # agent construction site can identify them. Lead agents + # assemble their full configured MCP catalog and apply active + # skill policy at runtime; subagents may pass an already + # policy-filtered list because their skills load at startup. + for t in mcp_tools: + tag_mcp_tool(t) + else: + # A change that disables the last MCP server must still retire + # the previously initialized cache and its pooled sessions. + # This never initializes tools: a process that never initialized + # MCP pays no config-hashing or discovery cost, while one that + # did still checks the existing cache for staleness. + refresh_mcp_cache_if_active() + if mcp_plugins is not None: + from deerflow.capabilities.runtime import filter_mcp_plugins - mcp_tools = filter_mcp_plugins(mcp_tools, mcp_plugins, extensions_config) + mcp_tools = filter_mcp_plugins(mcp_tools, mcp_plugins, extensions_config) except ImportError: logger.warning("MCP module not available. Install 'langchain-mcp-adapters' package to enable MCP tools.") - except Exception as e: - logger.error(f"Failed to get cached MCP tools: {e}") + except Exception: + # Tool caching, pool cleanup and plugin filtering raise ordinary + # exceptions whose messages are safe and needed for debugging. + logger.exception("Failed to get cached MCP tools") # Add invoke_acp_agent tool if any ACP agents are configured acp_tools: list[BaseTool] = [] diff --git a/backend/tests/test_mcp_cache.py b/backend/tests/test_mcp_cache.py index 63d25080f1..a6580af01e 100644 --- a/backend/tests/test_mcp_cache.py +++ b/backend/tests/test_mcp_cache.py @@ -22,6 +22,7 @@ from __future__ import annotations import asyncio import json +import logging import os import threading from pathlib import Path @@ -30,6 +31,7 @@ import pytest import deerflow.mcp.cache as cache_module from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.config.file_signature import get_config_signature _MISSING = object() @@ -47,11 +49,22 @@ _TRACKED_GLOBALS = ( "_init_condition", "_initializing_generation", "_cache_generation", + "_mcp_config_snapshot", + "_initialized_without_config", ) -def _write_extensions_config(path: Path, servers: dict) -> None: - path.write_text(json.dumps({"mcpServers": servers, "skills": {}}), encoding="utf-8") +def _write_extensions_config( + path: Path, + servers: dict, + *, + skills: dict | None = None, + interceptors: list | str | None = None, +) -> None: + payload: dict = {"mcpServers": servers, "skills": skills or {}} + if interceptors is not None: + payload["mcpInterceptors"] = interceptors + path.write_text(json.dumps(payload), encoding="utf-8") def _server(command: str = "npx") -> dict: @@ -65,7 +78,7 @@ def cache_globals(): cache_module._mcp_tools_cache = None cache_module._cache_initialized = False - for name in ("_config_path", "_config_signature", "_config_mtime"): + for name in ("_config_path", "_config_signature", "_config_mtime", "_mcp_config_snapshot", "_initialized_without_config"): if hasattr(cache_module, name): setattr(cache_module, name, None) # threading.Lock is safe across threads and does not bind to event loops, @@ -95,7 +108,7 @@ def _initialize_against(monkeypatch, config_path: Path) -> None: """ monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path)) - async def _fake_get_mcp_tools(): + async def _fake_get_mcp_tools(**_kwargs): return [] monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) @@ -316,7 +329,7 @@ class TestCrossLoopReinitialization: _write_extensions_config(cfg, {"srv1": _server()}) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) - async def _fake_tools(): + async def _fake_tools(**_kwargs): return [] monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) @@ -357,7 +370,7 @@ class TestCrossLoopReinitialization: calls = 0 - async def _fake_tools(): + async def _fake_tools(**_kwargs): nonlocal calls calls += 1 await asyncio.sleep(0.01) @@ -404,7 +417,7 @@ class TestCrossLoopReinitialization: _write_extensions_config(cfg, {"srv1": _server()}) monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) - async def _fake_tools(): + async def _fake_tools(**_kwargs): return [] monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) @@ -436,7 +449,7 @@ def test_config_change_during_initialization_discards_stale_tools(cache_globals, calls = 0 - async def _fake_tools(): + async def _fake_tools(**_kwargs): nonlocal calls calls += 1 if calls == 1: @@ -501,7 +514,7 @@ def test_config_change_during_initialization_retires_pool_for_same_server_connec loaded_pools = [] loaded_sessions = [] - async def _fake_tools(): + async def _fake_tools(**_kwargs): nonlocal calls calls += 1 server = json.loads(cfg.read_text())["mcpServers"]["same"] @@ -541,7 +554,7 @@ def test_reset_mcp_tools_cache_does_not_wait_for_in_flight_initialization(cache_ started = threading.Event() finish = threading.Event() - async def _fake_tools(): + async def _fake_tools(**_kwargs): started.set() await asyncio.to_thread(finish.wait) return [] @@ -593,7 +606,7 @@ def test_automatic_stale_invalidation_retires_session_pool_before_reinitializing _write_extensions_config(cfg, {"new": _server("uvx")}) loaded_pools = [] - async def _fake_tools(): + async def _fake_tools(**_kwargs): loaded_pools.append(session_pool_module.get_session_pool()) return ["new-tools"] @@ -623,7 +636,7 @@ def test_reset_mcp_tools_cache_retires_session_pool_before_releasing_initializer race_results = [] loaded_pools = [] - async def _fake_tools(): + async def _fake_tools(**_kwargs): loaded_pools.append(session_pool_module.get_session_pool()) return ["race-tools"] @@ -662,7 +675,7 @@ def test_cancelled_initializer_releases_generation_claim(cache_globals, monkeypa release = asyncio.Event() calls = 0 - async def _fake_tools(): + async def _fake_tools(**_kwargs): nonlocal calls calls += 1 started.set() @@ -690,3 +703,506 @@ def test_cancelled_initializer_releases_generation_claim(cache_globals, monkeypa assert cache_module._cache_initialized is True assert cache_module._initializing_generation is None assert calls == 2 + + +# --------------------------------------------------------------------------- +# Effective-MCP-config invalidation +# +# ``extensions_config.json`` carries both ``mcpServers`` and ``skills``, and the +# cache signature covers the whole file. Before this fix any skill edit closed +# every pooled MCP session even though no MCP server changed. These tests pin +# the narrowed contract: only the effective MCP slice may retire MCP state. +# --------------------------------------------------------------------------- + + +def test_skills_only_change_keeps_tools_and_pool(cache_globals, monkeypatch, tmp_path): + """Toggling a skill must not rebuild tools, replace the pool, or close sessions.""" + from deerflow.mcp.session_pool import get_session_pool, reset_session_pool + + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": True}}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + sentinel_tool = object() + + async def _fake_get_mcp_tools(**_kwargs): + return [sentinel_tool] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + asyncio.run(cache_module.initialize_mcp_tools()) + + old_pool = get_session_pool() + closed: list[bool] = [] + monkeypatch.setattr(old_pool, "close_all_sync", lambda: closed.append(True)) + + try: + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": False}}) + + assert cache_module.get_cached_mcp_tools() == [sentinel_tool] + assert cache_module._mcp_tools_cache == [sentinel_tool] + assert get_session_pool() is old_pool + assert closed == [] + assert cache_module._config_signature == get_config_signature(cfg) + finally: + reset_session_pool() + + +def test_same_mtime_same_size_skills_only_change_is_not_stale(cache_globals, monkeypatch, tmp_path): + """A skills-only edit that hides behind mtime+size still updates the checked signature.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": True}}) + _initialize_against(monkeypatch, cfg) + recorded_mtime = cfg.stat().st_mtime + recorded_size = cfg.stat().st_size + + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-b": {"enabled": True}}) + os.utime(cfg, (recorded_mtime, recorded_mtime)) + assert cfg.stat().st_mtime == recorded_mtime + assert cfg.stat().st_size == recorded_size + + assert cache_module._is_cache_stale() is False + assert cache_module._config_signature == get_config_signature(cfg) + + +def test_mcp_connection_change_is_still_stale(cache_globals, monkeypatch, tmp_path): + """A real MCP change must keep retiring the cache (no over-narrowing).""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + _initialize_against(monkeypatch, cfg) + + _write_extensions_config(cfg, {"srv1": _server("uvx")}) + + assert cache_module._is_cache_stale() is True + + +def test_mcp_interceptors_change_is_still_stale(cache_globals, monkeypatch, tmp_path): + """A global mcpInterceptors change must keep retiring the cache.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, interceptors=["pkg.a:build"]) + _initialize_against(monkeypatch, cfg) + + _write_extensions_config(cfg, {"srv1": _server()}, interceptors=["pkg.b:build"]) + + assert cache_module._is_cache_stale() is True + + +def test_explicit_reset_still_clears_after_skills_only_change(cache_globals, monkeypatch, tmp_path): + """The semantic comparison must never suppress an explicit global reset.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": True}}) + _initialize_against(monkeypatch, cfg) + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": False}}) + assert cache_module.get_cached_mcp_tools() == [] + + cache_module.reset_mcp_tools_cache() + + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + assert cache_module._mcp_config_snapshot is None + + +def test_malformed_config_is_still_stale(cache_globals, monkeypatch, tmp_path): + """An unreadable config must never be mistaken for a skills-only no-op.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + _initialize_against(monkeypatch, cfg) + + cfg.write_text("{not json", encoding="utf-8") + + assert cache_module._is_cache_stale() is True + + +def test_refresh_is_noop_before_initialization(cache_globals, monkeypatch): + """Deployments that never initialized MCP pay no config-hashing cost.""" + calls: list[str] = [] + monkeypatch.setattr( + cache_module, + "_current_config_state", + lambda: calls.append("checked") or (None, None), + ) + + assert cache_module.refresh_mcp_cache_if_active() is False + assert calls == [] + + +def test_refresh_retires_cache_when_last_server_is_disabled(cache_globals, monkeypatch, tmp_path): + """Disabling the last server still converges an already-initialized cache.""" + from deerflow.mcp.session_pool import get_session_pool, reset_session_pool + + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + _initialize_against(monkeypatch, cfg) + + old_pool = get_session_pool() + closed: list[bool] = [] + monkeypatch.setattr(old_pool, "close_all_sync", lambda: closed.append(True)) + _write_extensions_config(cfg, {}) + + try: + assert cache_module.refresh_mcp_cache_if_active() is True + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + assert cache_module._mcp_config_snapshot is None + assert get_session_pool() is not old_pool + assert closed == [True] + finally: + reset_session_pool() + + +def test_refresh_keeps_cache_when_only_skills_changed(cache_globals, monkeypatch, tmp_path): + """The lightweight refresh path honours the effective-MCP comparison.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": True}}) + _initialize_against(monkeypatch, cfg) + + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": False}}) + + assert cache_module.refresh_mcp_cache_if_active() is False + assert cache_module._cache_initialized is True + + +def test_skills_only_change_during_initialization_publishes_loaded_tools(cache_globals, monkeypatch, tmp_path): + """A skills-only edit during tool discovery must not force a needless rebuild.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": True}}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + calls = 0 + + async def _fake_tools(**_kwargs): + nonlocal calls + calls += 1 + _write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": False}}) + return ["loaded-tools"] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + + result = asyncio.run(cache_module.initialize_mcp_tools()) + + assert result == ["loaded-tools"] + assert cache_module._cache_initialized is True + assert cache_module._mcp_tools_cache == ["loaded-tools"] + assert calls == 1 + + +def test_refresh_during_initialization_discards_stale_publish(cache_globals, monkeypatch, tmp_path): + """Disabling the last server while discovery is in flight must void the result.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + started = threading.Event() + release = threading.Event() + + async def _fake_tools(**_kwargs): + started.set() + while not release.is_set(): + await asyncio.sleep(0.01) + return ["stale-tools"] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + + async def _run(): + owner = asyncio.create_task(cache_module.initialize_mcp_tools()) + while not started.is_set(): + await asyncio.sleep(0.01) + _write_extensions_config(cfg, {}) + assert cache_module.refresh_mcp_cache_if_active() is True + release.set() + return await asyncio.wait_for(owner, timeout=2) + + assert asyncio.run(_run()) == [] + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + + +def test_initialization_without_a_readable_snapshot_discards_result(cache_globals, monkeypatch, tmp_path): + """Never publish tools that cannot be tied to a parsed MCP snapshot.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + async def _fake_get_mcp_tools(**_kwargs): + return ["loaded-tools"] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + monkeypatch.setattr(cache_module, "_read_stable_mcp_snapshot", lambda path, signature: None) + + result = asyncio.run(cache_module.initialize_mcp_tools()) + + assert result == [] + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + + +def test_parse_failure_log_does_not_leak_resolved_credentials(cache_globals, monkeypatch, tmp_path, caplog): + """A malformed config must not echo resolved $VAR secrets into the log.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + _initialize_against(monkeypatch, cfg) + + monkeypatch.setenv("MCP_SECRET", "TOPSECRET123") + cfg.write_text( + json.dumps( + { + "mcpServers": { + "srv1": {"enabled": True, "type": "stdio", "command": "npx", "headers": "$MCP_SECRET"}, + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + + with caplog.at_level(logging.WARNING): + assert cache_module._is_cache_stale() is True + + assert "TOPSECRET123" not in caplog.text + + +def test_equivalent_transport_alias_spellings_are_not_stale(cache_globals, monkeypatch, tmp_path): + """``type=stdio`` and ``transport=stdio`` describe the same connection.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": {"enabled": True, "type": "stdio", "command": "npx"}}) + _initialize_against(monkeypatch, cfg) + + _write_extensions_config(cfg, {"srv1": {"enabled": True, "transport": "stdio", "command": "npx"}}) + + assert cache_module._is_cache_stale() is False + + +def test_initialization_hands_the_snapshotted_config_to_discovery(cache_globals, monkeypatch, tmp_path): + """Discovery must use the exact revision the cache snapshotted, not re-read.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server("npx")}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + seen: dict[str, str] = {} + + async def _fake_tools(*, extensions_config): + seen["command"] = extensions_config.mcp_servers["srv1"].command + # A transient flip to Y and back to X must not invalidate X-built tools. + _write_extensions_config(cfg, {"srv1": _server("uvx")}) + _write_extensions_config(cfg, {"srv1": _server("npx")}) + return ["loaded-tools"] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools) + + result = asyncio.run(cache_module.initialize_mcp_tools()) + + assert seen["command"] == "npx" + assert result == ["loaded-tools"] + assert cache_module._cache_initialized is True + assert cache_module._mcp_config_snapshot is not None + + +def test_config_appearing_after_unconfigured_init_is_stale(cache_globals, monkeypatch, tmp_path): + """A config that appears after an unconfigured init must be picked up.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + + state: dict[str, object] = {"path": None} + monkeypatch.setattr( + ExtensionsConfig, + "resolve_config_path", + classmethod(lambda cls, config_path=None: state["path"]), + ) + + async def _fake_get_mcp_tools(**_kwargs): + return [] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + asyncio.run(cache_module.initialize_mcp_tools()) + + assert cache_module._cache_initialized is True + assert cache_module._config_signature is None + assert cache_module._initialized_without_config is True + + state["path"] = cfg + + assert cache_module._is_cache_stale() is True + + +def test_initialization_with_unreadable_post_signature_discards_result(cache_globals, monkeypatch, tmp_path): + """A resolved path with an unreadable signature must not publish an unpinned cache.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + async def _fake_get_mcp_tools(**_kwargs): + return ["loaded-tools"] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + monkeypatch.setattr(cache_module, "_current_config_state", lambda: (cfg, None)) + + result = asyncio.run(cache_module.initialize_mcp_tools()) + + assert result == [] + assert cache_module._cache_initialized is False + assert cache_module._mcp_tools_cache is None + + +def test_lazy_init_does_not_leak_credentials_on_malformed_config(cache_globals, monkeypatch, tmp_path, caplog): + """Lazy init must not echo resolved secrets when the config is malformed.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + monkeypatch.setenv("MCP_SECRET", "TOPSECRET123") + + cfg.write_text( + json.dumps( + { + "mcpServers": { + "srv1": {"enabled": True, "type": "stdio", "command": "npx", "headers": "$MCP_SECRET"}, + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + + with caplog.at_level(logging.WARNING): + assert cache_module.get_cached_mcp_tools() == [] + + assert "TOPSECRET123" not in caplog.text + + +def _assert_cache_survives_edit(monkeypatch, cfg, sentinel, rewrite): + """Initialize with *sentinel*, apply *rewrite*, and prove nothing was retired.""" + from deerflow.mcp.session_pool import get_session_pool, reset_session_pool + + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + + async def _fake_get_mcp_tools(**_kwargs): + return [sentinel] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + asyncio.run(cache_module.initialize_mcp_tools()) + + old_pool = get_session_pool() + closed: list[bool] = [] + monkeypatch.setattr(old_pool, "close_all_sync", lambda: closed.append(True)) + try: + rewrite() + assert cache_module.get_cached_mcp_tools() == [sentinel] + assert cache_module._mcp_tools_cache == [sentinel] + assert get_session_pool() is old_pool + assert closed == [] + finally: + reset_session_pool() + + +def test_string_and_single_element_interceptor_list_are_equivalent(cache_globals, monkeypatch, tmp_path): + """A bare string and its single-element list select the same interceptor chain.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, interceptors="pkg.interceptor:build") + sentinel = object() + _assert_cache_survives_edit( + monkeypatch, + cfg, + sentinel, + lambda: _write_extensions_config(cfg, {"srv1": _server()}, interceptors=["pkg.interceptor:build"]), + ) + + +def test_missing_and_empty_interceptors_are_equivalent(cache_globals, monkeypatch, tmp_path): + """An absent key and an empty list both select no custom interceptors.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + sentinel = object() + _assert_cache_survives_edit( + monkeypatch, + cfg, + sentinel, + lambda: _write_extensions_config(cfg, {"srv1": _server()}, interceptors=[]), + ) + + +def test_interceptor_order_change_is_still_stale(cache_globals, monkeypatch, tmp_path): + """Order is part of the effective chain and must keep retiring the cache.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}, interceptors=["a:build", "b:build"]) + _initialize_against(monkeypatch, cfg) + + _write_extensions_config(cfg, {"srv1": _server()}, interceptors=["b:build", "a:build"]) + + assert cache_module._is_cache_stale() is True + + +def test_path_switch_with_equivalent_mcp_config_keeps_tools_and_pool(cache_globals, monkeypatch, tmp_path): + """Switching to a file whose MCP slice is identical must not retire sessions.""" + from deerflow.mcp.session_pool import get_session_pool, reset_session_pool + + cfg_a = tmp_path / "a_extensions_config.json" + cfg_b = tmp_path / "b_extensions_config.json" + _write_extensions_config(cfg_a, {"srv1": _server()}, skills={"skill-a": {"enabled": True}}) + _write_extensions_config(cfg_b, {"srv1": _server()}, skills={"skill-a": {"enabled": False}}) + + state: dict[str, object] = {"path": cfg_a} + monkeypatch.setattr( + ExtensionsConfig, + "resolve_config_path", + classmethod(lambda cls, config_path=None: state["path"]), + ) + + sentinel = object() + + async def _fake_get_mcp_tools(**_kwargs): + return [sentinel] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools) + asyncio.run(cache_module.initialize_mcp_tools()) + + old_pool = get_session_pool() + closed: list[bool] = [] + monkeypatch.setattr(old_pool, "close_all_sync", lambda: closed.append(True)) + try: + state["path"] = cfg_b + + assert cache_module.get_cached_mcp_tools() == [sentinel] + assert get_session_pool() is old_pool + assert closed == [] + assert cache_module._config_path == cfg_b + assert cache_module._config_signature == get_config_signature(cfg_b) + finally: + reset_session_pool() + + +def test_path_switch_with_different_mcp_config_is_still_stale(cache_globals, monkeypatch, tmp_path): + """A path switch to a file with a different MCP slice keeps the global reset.""" + cfg_a = tmp_path / "a_extensions_config.json" + cfg_b = tmp_path / "b_extensions_config.json" + _write_extensions_config(cfg_a, {"srv1": _server("npx")}) + _write_extensions_config(cfg_b, {"srv1": _server("uvx")}) + + state: dict[str, object] = {"path": cfg_a} + monkeypatch.setattr( + ExtensionsConfig, + "resolve_config_path", + classmethod(lambda cls, config_path=None: state["path"]), + ) + _initialize_against(monkeypatch, cfg_a) + + state["path"] = cfg_b + + assert cache_module._is_cache_stale() is True + + +def test_unverifiable_signature_is_not_treated_as_stable(cache_globals, monkeypatch, tmp_path): + """A signature without a sha256 digest must never be published as stable.""" + cfg = tmp_path / "extensions_config.json" + _write_extensions_config(cfg, {"srv1": _server()}) + _initialize_against(monkeypatch, cfg) + + real = cache_module._get_config_signature + + def _no_digest(path): + signature = real(path) + return None if signature is None else (signature[0], signature[1], None) + + monkeypatch.setattr(cache_module, "_get_config_signature", _no_digest) + incomplete = _no_digest(cfg) + assert incomplete is not None and incomplete[2] is None + + assert cache_module._read_stable_mcp_snapshot(cfg, incomplete) is None + assert cache_module._is_cache_stale() is True diff --git a/backend/tests/test_mcp_cache_tool_assembly.py b/backend/tests/test_mcp_cache_tool_assembly.py new file mode 100644 index 0000000000..6fb21e375a --- /dev/null +++ b/backend/tests/test_mcp_cache_tool_assembly.py @@ -0,0 +1,111 @@ +"""Regressions for MCP cache handling in ``get_available_tools()``.""" + +import json +import logging +from types import SimpleNamespace + +from deerflow.tools.tools import get_available_tools + + +def _make_config(): + return SimpleNamespace( + tools=[ + SimpleNamespace(name="bash", group="bash", use="deerflow.sandbox.tools:bash_tool"), + SimpleNamespace(name="ls", group="file:read", use="tests:ls_tool"), + ], + models=[], + sandbox=SimpleNamespace(use="deerflow.sandbox.local:LocalSandboxProvider", allow_host_bash=False), + tool_search=SimpleNamespace(enabled=False), + get_model_config=lambda name: None, + ) + + +def test_tool_assembly_refreshes_cache_when_no_server_is_enabled(monkeypatch): + from deerflow.config.extensions_config import ExtensionsConfig + + monkeypatch.setattr("deerflow.tools.tools.get_app_config", lambda: _make_config()) + monkeypatch.setattr( + "deerflow.tools.tools.resolve_variable", + lambda use, _: SimpleNamespace(name="bash" if "bash" in use else "ls"), + ) + monkeypatch.setattr( + ExtensionsConfig, + "from_file", + classmethod(lambda cls, config_path=None: ExtensionsConfig(mcp_servers={}, skills={})), + ) + + calls: list[str] = [] + monkeypatch.setattr( + "deerflow.mcp.cache.refresh_mcp_cache_if_active", + lambda: calls.append("refresh") or False, + ) + + get_available_tools(include_mcp=True, subagent_enabled=False) + + assert calls == ["refresh"] + + +def test_tool_assembly_does_not_leak_credentials_on_malformed_config(monkeypatch, tmp_path, caplog): + """A malformed config must not echo resolved $VAR secrets into the log.""" + cfg = tmp_path / "extensions_config.json" + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg)) + monkeypatch.setenv("MCP_SECRET", "TOPSECRET123") + cfg.write_text( + json.dumps( + { + "mcpServers": { + "srv1": {"enabled": True, "type": "stdio", "command": "npx", "headers": "$MCP_SECRET"}, + }, + "skills": {}, + } + ), + encoding="utf-8", + ) + + monkeypatch.setattr("deerflow.tools.tools.get_app_config", lambda: _make_config()) + monkeypatch.setattr( + "deerflow.tools.tools.resolve_variable", + lambda use, _: SimpleNamespace(name="bash" if "bash" in use else "ls"), + ) + + with caplog.at_level(logging.WARNING): + get_available_tools(include_mcp=True, subagent_enabled=False) + + assert "TOPSECRET123" not in caplog.text + mcp_records = [record for record in caplog.records if "MCP" in record.getMessage()] + assert any("Failed to load MCP extensions config" in record.getMessage() for record in mcp_records) + assert all(record.exc_info is None for record in mcp_records) + + +def test_tool_assembly_logs_diagnostics_for_non_config_failures(monkeypatch, caplog): + """Only config-load failures are sanitized; other MCP errors keep diagnostics.""" + from deerflow.config.extensions_config import ExtensionsConfig + + monkeypatch.setattr("deerflow.tools.tools.get_app_config", lambda: _make_config()) + monkeypatch.setattr( + "deerflow.tools.tools.resolve_variable", + lambda use, _: SimpleNamespace(name="bash" if "bash" in use else "ls"), + ) + monkeypatch.setattr( + ExtensionsConfig, + "from_file", + classmethod( + lambda cls, config_path=None: ExtensionsConfig.model_validate( + { + "mcpServers": {"srv1": {"enabled": True, "type": "stdio", "command": "npx"}}, + "skills": {}, + } + ) + ), + ) + + def _boom(): + raise RuntimeError("retired pool close failed: pipe still open") + + monkeypatch.setattr("deerflow.mcp.cache.get_cached_mcp_tools", _boom) + + with caplog.at_level(logging.ERROR): + get_available_tools(include_mcp=True, subagent_enabled=False) + + assert "retired pool close failed: pipe still open" in caplog.text + assert any(record.exc_info is not None for record in caplog.records) diff --git a/backend/tests/test_mcp_task_runtime_config.py b/backend/tests/test_mcp_task_runtime_config.py index 2989efedbb..d432254666 100644 --- a/backend/tests/test_mcp_task_runtime_config.py +++ b/backend/tests/test_mcp_task_runtime_config.py @@ -137,3 +137,37 @@ def test_mcp_interceptor_changes_remain_hot_reloadable_without_task_tools() -> N validate_mcp_task_config_snapshot(current) finally: set_mcp_task_config_snapshot(None) + + +def _task_server(task_toolsets: list[dict]) -> dict: + return {"mcpServers": {"reports": {"command": "reports-mcp", "task_toolsets": task_toolsets}}} + + +_TASK_TOOLSETS = [ + { + "name": "reports", + "submit_tool": "submit_report", + "status_tool": "status_report", + "cancel_tool": "cancel_report", + } +] + + +def test_equivalent_interceptor_spellings_do_not_invalidate_task_runtime_snapshot() -> None: + startup = ExtensionsConfig.model_validate({**_task_server(_TASK_TOOLSETS), "mcpInterceptors": "example.interceptor:build"}) + current = ExtensionsConfig.model_validate({**_task_server(_TASK_TOOLSETS), "mcpInterceptors": ["example.interceptor:build"]}) + set_mcp_task_config_snapshot(startup) + try: + validate_mcp_task_config_snapshot(current) + finally: + set_mcp_task_config_snapshot(None) + + +def test_transport_alias_does_not_invalidate_task_runtime_snapshot() -> None: + startup = ExtensionsConfig.model_validate({"mcpServers": {"reports": {"type": "stdio", "command": "reports-mcp", "task_toolsets": _TASK_TOOLSETS}}}) + current = ExtensionsConfig.model_validate({"mcpServers": {"reports": {"transport": "stdio", "command": "reports-mcp", "task_toolsets": _TASK_TOOLSETS}}}) + set_mcp_task_config_snapshot(startup) + try: + validate_mcp_task_config_snapshot(current) + finally: + set_mcp_task_config_snapshot(None) diff --git a/frontend/src/content/en/harness/mcp.mdx b/frontend/src/content/en/harness/mcp.mdx index 30dfc6a6a5..b7f37e3b61 100644 --- a/frontend/src/content/en/harness/mcp.mdx +++ b/frontend/src/content/en/harness/mcp.mdx @@ -69,7 +69,11 @@ If the server starts before MCP tools are initialized (e.g., in LangGraph Studio ### Cache invalidation -The MCP tools cache tracks the modification time (`mtime`) of `extensions_config.json`. When the file changes — for example, when a server is enabled or disabled through the Gateway API — the cache is marked stale and tools are reloaded on the next request. +The MCP tools cache tracks the resolved path and a `(mtime, size, sha256)` content signature of `extensions_config.json`. A changed signature is only a signal to re-check the effective MCP configuration: the cache compares the enabled servers' parsed configuration plus the global `mcpInterceptors` list, and retires cached tools and pooled sessions only when that slice actually changed. + +Skills-only or middleware-only edits therefore keep the current MCP tools and sessions, and so does a switch to another config file with an identical MCP configuration. Enabling, disabling, or reconfiguring a server reloads tools on the next request. When no MCP server is enabled, tool assembly retires any previously initialized cache and sessions without starting new tool discovery. + +The Gateway's explicit **Reset MCP tools cache** action (`POST /api/mcp/cache/reset`) is separate from this automatic check and always clears the cache and pooled sessions. This means MCP server changes take effect without restarting the DeerFlow server. diff --git a/frontend/src/content/zh/harness/mcp.mdx b/frontend/src/content/zh/harness/mcp.mdx index eaa68c954e..5d5fbe2dec 100644 --- a/frontend/src/content/zh/harness/mcp.mdx +++ b/frontend/src/content/zh/harness/mcp.mdx @@ -63,7 +63,11 @@ DeerFlow 服务器启动时调用 `initialize_mcp_tools()`。这连接到所有 ### 缓存失效 -MCP 工具缓存追踪 `extensions_config.json` 的修改时间（`mtime`）。当文件更改时——例如通过 Gateway API 启用或禁用服务器时——缓存被标记为过时，下次请求时重新加载工具。 +MCP 工具缓存追踪 `extensions_config.json` 的解析路径和 `(mtime, size, sha256)` 内容签名。签名变化只是触发重新比对的信号：缓存会比较已启用服务器的解析后配置与全局 `mcpInterceptors` 列表，只有该有效切片真正变化时才会淘汰已缓存的工具和会话池。 + +因此仅修改 Skills 或 middleware 不会重建工具或关闭会话；切换到另一份 MCP 配置完全相同的配置文件同样保留会话。启用、禁用或修改某个服务器会在下次请求时重新加载工具。当没有启用任何 MCP 服务器时，工具组装会回收此前已初始化的缓存与会话，但不会发起新的工具发现。 + +Gateway 的显式 **重置 MCP 工具缓存**（`POST /api/mcp/cache/reset`）与此自动检查相互独立，始终清理缓存和会话池。 这意味着 MCP 服务器更改无需重启 DeerFlow 服务器即可生效。