mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-10-26 21:26:39 +00:00
fix(mcp): invalidate cache only for effective MCP config changes (#5750)
* fix(mcp): avoid cache resets for unrelated config changes extensions_config.json carries MCP servers, skills and middleware settings, but the MCP cache treated any whole-file signature change as MCP state change, so toggling a skill retired every cached tool and closed all pooled sessions. Record a snapshot of the effective MCP slice (enabled servers' parsed config plus mcpInterceptors) and retire MCP state only when that slice changes. Hand the exact config revision to tool discovery so a change during discovery cannot publish tools built from a superseded revision, and add a non-initializing refresh entry point so disabling the last enabled server still converges the cache and pool on the next tool assembly. * docs(mcp): align cache guide and restore error diagnostics Review feedback on #5750 flagged three non-blocking issues: the MCP module guide still described the pre-semantic-snapshot invalidation contract; the tools.py sanitization covered every exception in the MCP block, discarding diagnostics for failures that cannot carry resolved credentials; and _is_cache_stale()'s docstring presented it as a pure predicate despite adopting a new file signature. Narrow the type-only logging to ExtensionsConfig.from_file() and restore logger.exception() for tool caching, pool cleanup and plugin filtering, with a regression test asserting the message and traceback survive. Document the effective-MCP-slice comparison, the stable config handoff to discovery, refresh_mcp_cache_if_active() and the _initialized_without_config handling in the module guide, and note the signature-adoption side effect in the docstring. * fix(mcp): unify config normalization and verify signatures Review follow-up on #5750. - Normalize mcpInterceptors (bare string == single-element list; missing key == empty list) and the type/transport alias through one shared module used by the tools cache, the interceptor builder and the durable-task config snapshot, so equivalent spellings no longer retire tools/sessions or raise McpTaskConfigurationError. - Treat a resolved-path switch as a signal to re-read the effective MCP slice instead of an invalidation reason: an identical slice adopts the new path and signature and keeps pooled sessions. - Refuse to publish or adopt a signature whose sha256 digest is missing ((mtime, size, None) from a file that could not be read). - Sync the English and Chinese MCP user docs with the effective-config comparison, and document that the Gateway admin reset stays unconditional. * docs(mcp): note the unverifiable-signature fail-closed branch A present config file whose content cannot be hashed is treated as stale, unlike a deleted file which keeps the last-known-good cache. Record that asymmetry in the module guide. * docs(mcp): scope the no-hash rationale to never-initialized processes The cost-free early return in refresh_mcp_cache_if_active() only covers a process that never published an MCP cache. A process that did publish one (including an empty cache from the zero-server or LangGraph Studio lazy-init path) still pays one stat+sha256 per tool assembly. Correct the rationale in the docstring and the matching comment in get_available_tools().
This commit is contained in:
spud
GitHub
parent
fe70cfc5ae
commit
8204dcbb20
@ -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.<server>.tool_name_prefix` defaults to `true`, preserving the collision-safe `<server_name>_` 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 `<token_type> <access_token>` 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 <pkg> -c '<command>'` **runs** the command (`-p` is `npm exec`'s `--package`, so `<pkg>` 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`.
|
||||
|
||||
@ -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()
|
||||
|
||||
|
||||
@ -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
|
||||
@ -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:
|
||||
|
||||
@ -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
|
||||
|
||||
|
||||
|
||||
@ -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)
|
||||
|
||||
|
||||
@ -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] = []
|
||||
|
||||
@ -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
|
||||
|
||||
111
backend/tests/test_mcp_cache_tool_assembly.py
Normal file
111
backend/tests/test_mcp_cache_tool_assembly.py
Normal file
@ -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)
|
||||
@ -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)
|
||||
|
||||
@ -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.
|
||||
|
||||
|
||||
@ -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 服务器即可生效。
|
||||
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user