github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-10-26 13:16:24 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(mcp): reuse unchanged per-server tool discovery results (#6795)

Browse Source 
* feat(mcp): add per-server discovery cache scaffolding

Introduce the per-server tool cache entry, canonical parsed-config snapshots on the applied revision, and the shared _binding_matches reuse predicate. Identity reuse is object-identity based on the pool's current binding (no value equality); snapshots and ownership handles stay out of repr. No selection, eviction or publication is wired yet.

* feat(mcp): selectively evict stale per-server tool cache entries

Evict per-server discovery results during selective reconciliation, after the new binding epoch is installed, so unchanged servers keep their exact entry objects while changed, removed or re-bound servers (including lifecycle-token rotations) are dropped. Delegate the binding verdict to _binding_matches so reuse and eviction cannot diverge.

Clear _server_tool_cache on every full reset, and count retained entries in the reconciliation state predicates so a lingering entry can never let a later config change be silently skipped.

* feat(mcp): reuse unchanged per-server results during initialization

Initialize MCP tools from per-server discovery instead of a single full pass: select reusable entries under the claim lock, discover only the missing servers with get_mcp_tools_by_server, then publish atomically after re-validating every candidate binding against the pool's current binding via _binding_matches. Failed servers stay absent (never cached as empty), and superseded or partially bound candidates are discarded whole. Test fixtures now drive the grouped entry point.

* fix(mcp): close T4 review gaps in per-server cache publication

Address four findings from the adversarial review: treat retained entries as unverifiable when the applied baseline is None so an interceptor change still forces a full reset; count retained entries in the claim-time shared-reset predicate; refuse to publish a false-success empty cache when every enabled server failed discovery; and hoist the lazy imports inside the initializer's try/finally so an import failure releases the initialization claim.

* fix(mcp): sanitize cache failure logging and document per-server cache

The outer lazy-initialization handler now logs only the exception type, never the raw message or traceback, so a discovery error cannot print a resolved $VAR credential. The committed-writer planner mirrors the reader's no-baseline guard so retained entries are never reused under an unknown interceptor chain. Document the per-server tool cache (reuse, invalidation, failure handling, remote tool changes) in MCP_SERVER.md and mcp/AGENTS.md, and drop phase-staging wording from the new code and tests.

* docs(mcp): keep the agent-guidance chain under its hard limit

F2's cache notes pushed the effective AGENTS.md instruction chain 905 bytes past its 98304-byte hard limit, failing the agent-guidance CI job. Compress the selective-reconciliation bullet to its invariants (the full cache behaviour lives in docs/MCP_SERVER.md), and reword the total-failure comment so it no longer implies that build-invalid connections were attempted.

* fix(mcp): fence cached tools by interceptor chain

---------

Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
spud 2026-10-14 08:44:21 +08:00 committed by GitHub
parent 6ce385b963
commit eadc1e230e
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
8 changed files with 1547 additions and 77 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
README.md
View File

@ -875,6 +875,8 @@ Targeted updates accept both DeerFlow's `type` field and the MCP-spec `transport
Runtime MCP and skill updates replace `extensions_config.json` atomically, so an interrupted write cannot leave the shared configuration truncated or partially written. Every Gateway worker or instance that reads the same file picks up a new revision on its next read (the cache checks the file's content signature), so MCP and skill changes made through one replica apply to the others without a restart; a missing, truncated, or invalid revision keeps the previous configuration until a complete one lands, including when the file disappears during a reload.
The admin MCP cache reset advances a durable generation marker in the writable config directory. Every Gateway worker mounting that same directory retires its own cached tools and pooled sessions before the next lookup; replicas with independent filesystems are not implicitly covered. If no config path is available, the API reports a process-local reset instead.
The parsed extensions configuration and its recorded content digest come from the same read, so a racing edit followed by a timestamp-preserving backup restore cannot leave a different revision cached indefinitely.
Changes to the global `mcpInterceptors` chain rebuild every server's tool wrappers,
including when a server configuration is edited at the same time.
`extensions_config.json` accepts UTF-8 with or without a leading byte-order mark (BOM), including files saved as UTF-8 with BOM by an editor.

33
backend/docs/MCP_SERVER.md
View File

@ -656,6 +656,39 @@ material already retained by those systems. Restarting or upgrading DeerFlow doe
not rotate credentials or perform historical cleanup; operators must complete
both actions explicitly.
## Tool Discovery Cache
MCP tool discovery is cached per server. The cache is keyed by server name and
holds, for each server, the discovered tools together with a canonical snapshot
of that server's configuration, the normalized global interceptor chain and the
pooled stdio session binding they were discovered against.
- **Reuse.** On every (re)initialization, a server's cached tools are reused only
when its configuration snapshot and global interceptor chain are unchanged
*and* it still owns the process's current binding. The chain is checked against
each entry's published wrappers, including after a failed initialization attempt
or an edit between reconciliation and initialization. An unchanged server keeps
the exact same tool objects and is not re-discovered, so it does not re-fetch
OAuth headers or respawn its stdio process. Only changed servers are re-discovered.
- **Invalidation.** Editing a server's configuration (connection, description,
routing, tool prefixes, ...) drops that server's entry; deleting or disabling a
server drops it; an explicit cache reset or a shared reset marker clears every
server. Reordering servers only changes the flattened tool order.
- **Failure handling.** A server whose discovery fails is left absent for that
pass and its previous entry is dropped, while healthy servers still publish. If
*every* enabled server fails, initialization is aborted rather than publishing
an empty tool set, and the next call retries. A server that legitimately
returns zero tools is cached as a successful empty result and is not
re-discovered.
- **Remote tool changes.** The cache is configuration-driven: a remote MCP server
that changes its `tools/list` response without a configuration change is not
picked up automatically. Use `POST /api/mcp/cache/reset` (or disable/re-enable
or edit the server) to force re-discovery, which the shared reset marker
propagates to every worker mounting the same config directory.
Because reuse is scoped to one server, changing only one server never re-discovers,
re-authenticates, or restarts the others.
## How It Works
MCP servers expose tools that are automatically discovered and integrated into DeerFlow’s agent system at runtime. Once enabled, these tools become available to agents without additional code changes.

5
backend/packages/harness/deerflow/mcp/AGENTS.md
View File

@ -38,9 +38,8 @@ The optional background-task `status` filter flows through `McpTaskSubmitter.lis
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.
- **Persistent stdio binding fence**: pooled session registry keys remain `(server_name, scope_key, owning_loop, ownership_domain)`. Separately, each `(ownership_domain, server_name)` has an opaque `ServerBinding` capability derived from a SHA-256 digest of the base stdio process identity (`transport`, `command`, `args`, configured `cwd`, configured `env`; the resolved values themselves are never stored in the binding repr). Discovery captures the exact pool plus binding before its first await, and durable stdio calls bind before per-call workspace `cwd`/`TMPDIR` augmentation. `get_session()` rejects bindings that are already superseded at admission, owner promotion, or the final post-await check. That final check is a check-before-return guard, not a lease across subsequent `call_tool()` use: a reconciliation racing after it detaches and closes the owner, and the caller then fails through the existing transport-disconnect handling. Binding reconciliation consumes already-classified active fingerprints/removals, atomically installs epochs/tombstones and detaches matching owners under the pool lock, and keeps those detached owners in the pool's teardown reaper so their `__aexit__` stays observable even when the caller drops the returned `PreparedRetirement`; configuration diffing remains a higher-layer responsibility.
- **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`, so presentation-only fields are not connection changes). A claim *installs* the revision it records, so the baseline means those deployment epochs are in place: a stale binding left by a durable caller or an earlier failed attempt is repaired before `ensure_binding()` runs, which fails closed on a mismatch, and with no previous baseline the same lock tombstones the unlisted deployment servers the pool retains (`retire_unlisted=True`). Both epoch-installing decisions consult the frozen durable-task snapshot first: a rejected revision takes the conservative reset, and a claim fails as `get_mcp_tools()` would without touching the pool. The revision is recorded when discovery starts, so even a failed/cancelled discovery can retire a binding it seeded. Classification and application run in one `_init_condition` section, so epoch installation and generation invalidation cannot interleave. An unchanged effective slice adopts the new path/signature and keeps the published tools; a changed interceptor chain, an unreadable/unverifiable revision, a shared-reset-marker change, a frozen-task rejection, or an unbuildable baseline keeps the conservative whole-pool reset; otherwise `reconcile_bindings(active, removed, domain="deployment")` runs without replacing the pool, then the published tools are dropped and `_cache_generation` is bumped so an in-flight discovery cannot publish. Unchanged stdio servers keep their binding, session and owner; changed or removed ones advance an epoch or get a tombstone, and teardown stays with the pool's reaper outside every lock. Claim and transition are deployment-scoped, so a deployment revision never retires personal state; the conservative whole-pool resets do. A new shared-reset generation retires local state first whenever published tools, an applied baseline, or deployment pooled state exist; an empty/personal-only process adopts it. Explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` remain full resets. Gateway config writers call `prepare_mcp_reconciliation()` with the exact committed candidate under the config locks, then `finish_mcp_reconciliation()` outside them; post-lock response wrapping in `finally` keeps teardown from being skipped. This preserves delete→identical-re-add as two transitions and covers durable-only bindings that predate a published tool cache; if the committed handoff fails, `fail_mcp_reconciliation()` detaches local state and the API reports an explicit 500. The shared `mcpLifecycle` ledger (`mcp/lifecycle.py`) extends this cross-worker: readers compare `_AppliedMcpRevision.lifecycle_tokens` before the effective-snapshot shortcut, reader/claim/committed force-rebind changed servers, and every config write must go through a lifecycle-aware writer — during a rolling upgrade, restrict writes to upgraded workers until legacy writers are retired (a legacy delete→identical-re-add does not advance the ledger).
- **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 a real MCP change is classified against the process-local applied-reconciliation baseline and applied selectively (see the selective-reconciliation bullet above) instead of unconditionally replacing the pool. 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.
- **Process-local selective reconciliation**: `mcp/cache.py` keeps an *applied reconciliation baseline* (`_applied_mcp_revision`) apart from the published tool-cache baseline: the effective MCP snapshot, the normalized interceptor chain, and a `normalized_connection_fingerprint()` per enabled stdio server derived from `build_server_params()` (not a whole-config `model_dump()`). A claim *installs* the revision it records — repairing a stale durable/failed binding before `ensure_binding()` fails closed, and tombstoning unlisted deployment servers with no previous baseline — after consulting the frozen durable-task snapshot (a rejection takes the conservative reset; a claim otherwise fails like `get_mcp_tools()`). The revision is recorded when discovery starts, so a failed/cancelled discovery can still retire a binding it seeded. Classification and application share one `_init_condition` section. An unchanged slice adopts the new path/signature and keeps the published tools; an interceptor change, an unreadable/unverifiable revision, a shared-reset-marker change, a frozen rejection, or an unbuildable baseline keeps the conservative whole-pool reset; otherwise `reconcile_bindings(active, removed, domain="deployment")` runs without replacing the pool, then the published tools are dropped and `_cache_generation` is bumped. Unchanged stdio servers keep their binding, session and owner; changed/removed ones advance an epoch or get a tombstone, with teardown left to the pool's reaper outside every lock. Selective transitions never retire personal state; full resets do. A new shared-reset generation retires local state first whenever published tools, an applied baseline, or deployment pooled state exist (an empty/personal-only process adopts it); explicit `reset_mcp_tools_cache()` / `POST /api/mcp/cache/reset` stay full resets. Gateway writers use `prepare_mcp_reconciliation()` / `finish_mcp_reconciliation()` under and after the config locks (delete→identical-re-add stays two transitions; a failed handoff returns 500). The shared `mcpLifecycle` ledger (`mcp/lifecycle.py`) makes this cross-worker: readers compare `_AppliedMcpRevision.lifecycle_tokens` before the snapshot shortcut, reader/claim/committed force-rebind changed servers, and every config write needs a lifecycle-aware writer (restrict writes to upgraded workers mid-upgrade). Per-server results are cached by name and reused only while their canonical snapshot, normalized global interceptor chain and current binding ownership match; selective reconciliation evicts the rest, every full reset clears the cache, and a server that failed discovery is left absent rather than cached as a success.
- **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 a real MCP change is classified against the process-local applied-reconciliation baseline and applied selectively (see the selective-reconciliation bullet above) instead of unconditionally replacing the pool. 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, reuses only the per-server entries whose canonical snapshot, interceptor chain and current binding match, re-discovers the rest through `get_mcp_tools_by_server(extensions_config=..., server_names=missing)`, and publishes the flat list and per-server entries atomically only when the effective MCP snapshot, the `mcpLifecycle` tokens, the shared-reset marker and every candidate binding are still current; otherwise it resets or reconciles the tool cache as before. 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.

243
backend/packages/harness/deerflow/mcp/cache.py
View File

@ -22,6 +22,7 @@ from deerflow.mcp.lifecycle import (
lifecycle_delta,
parse_mcp_lifecycle,
)
from deerflow.mcp.tools import ServerDiscoveryResult
if TYPE_CHECKING:
from deerflow.mcp.session_pool import MCPSessionPool
@ -44,16 +45,23 @@ class _AppliedMcpRevision:
digest ``ensure_binding``/``reconcile_bindings`` consume. Those values
are one-way digests, so resolved ``$VAR`` credentials never appear there.
``effective_snapshot`` is *not* secret-safe (it is the fully resolved MCP
slice), so it is excluded from ``repr`` — the same contract
``_mcp_config_snapshot`` documents. Never log or persist this object's
snapshot field.
``server_snapshots`` maps every enabled server to
``_server_config_snapshot(...)`` — a canonical digest of its *parsed*
configuration, so a metadata/routing change is distinguishable from an
equivalent ``type``/``transport`` respelling. It drives selective eviction;
it is never used as connection identity (``stdio_connections`` owns that).
``effective_snapshot`` and ``server_snapshots`` are *not* secret-safe (they
are the fully resolved MCP slice), so they are excluded from ``repr`` — the
same contract ``_mcp_config_snapshot`` documents. Never log or persist these
fields.
"""
effective_snapshot: str = field(repr=False)
interceptors: str
stdio_connections: Mapping[str, str]
lifecycle_tokens: Mapping[str, str]
server_snapshots: Mapping[str, str] = field(repr=False)
# Reconciliation kinds produced by ``_plan_mcp_reconciliation_locked``. The
@ -89,7 +97,29 @@ class McpReconciliationPending:
retired_pool: "MCPSessionPool | None" = None
@dataclass(frozen=True, slots=True)
class _ServerToolCacheEntry:
"""A published per-server discovery result and its config identity.
``snapshot`` is the server's canonical config identity at publication (see
``_server_config_snapshot``), ``interceptors`` records the normalized global
chain used to wrap the tools, and ``result`` is the exact
``ServerDiscoveryResult`` published for it. All are excluded from ``repr``:
the snapshot may embed resolved credentials and the result carries ownership
handles.
"""
snapshot: str = field(repr=False)
interceptors: str = field(repr=False)
result: ServerDiscoveryResult = field(repr=False)
_mcp_tools_cache: list[BaseTool] | None = None
# Per-server published discovery results, keyed by server name. Deliberately
# separate from the flattened ``_mcp_tools_cache`` so an unchanged server's exact
# ``BaseTool`` objects can be reused without re-running discovery. The
# reconciliation and initialization paths populate and selectively evict it.
_server_tool_cache: dict[str, _ServerToolCacheEntry] = {}
_cache_initialized = False
_init_lock = threading.RLock() # Guards cache state transitions.
_init_condition = threading.Condition(_init_lock)
@ -245,6 +275,21 @@ def _interceptors_snapshot(config) -> str:
)
def _server_config_snapshot(server) -> str:
"""Canonical, comparable identity for one parsed MCP server config.
Built from ``normalize_mcp_server_config`` so an equivalent
``type``/``transport`` spelling and differing JSON key order compare equal,
while any field that can affect the discovered tools (connection params,
``description``, ``routing``, ``tools``, ``tool_name_prefix``, ...) compares
unequal. ``sort_keys`` normalizes nested key order.
The result may embed resolved ``$VAR`` credentials; it is only ever compared
for equality and must never be logged or persisted.
"""
return json.dumps(normalize_mcp_server_config(server), sort_keys=True, ensure_ascii=False)
def _mcp_revision_from_config(config) -> _AppliedMcpRevision:
"""Build the applied-reconciliation revision for a parsed config.
@ -255,11 +300,13 @@ def _mcp_revision_from_config(config) -> _AppliedMcpRevision:
"""
stdio_connections = _stdio_fingerprints(config)
ledger = parse_mcp_lifecycle(config.model_extra or {})
server_snapshots = {name: _server_config_snapshot(server) for name, server in config.get_enabled_mcp_servers().items()}
return _AppliedMcpRevision(
effective_snapshot=_effective_mcp_config_snapshot(config),
interceptors=_interceptors_snapshot(config),
stdio_connections=stdio_connections,
lifecycle_tokens=effective_tokens(config, ledger),
server_snapshots=server_snapshots,
)
@ -391,6 +438,65 @@ def _deployment_binding_delta(
return active, tuple(sorted(removed))
def _binding_matches(
result: ServerDiscoveryResult,
name: str,
fingerprint: str | None,
pool: "MCPSessionPool | None",
) -> bool:
"""Return whether ``result`` may be reused for ``name`` against ``pool``.
``fingerprint`` is the server's *current* connection digest: ``None`` for
HTTP/SSE servers, which hold no pooled stdio binding.
For stdio, reuse requires the *exact* binding object to still be the pool's
current one. ``ServerBinding`` is declared ``eq=False`` precisely so this
check is object identity, never dataclass value equality: a cross-worker
delete/re-add of an identical server re-epochs the binding to a new object
carrying the same fingerprint, and only identity comparison rejects the
superseded result (the ABA case).
This deliberately calls the pool's private ``_binding_is_current`` — an
intentional, read-only, in-package coupling shared by the reuse check and the
pre-publication check so the two cannot diverge. Callers must already hold
``_init_condition``; ``_binding_is_current`` then acquires the pool lock,
preserving the established ``_init_condition -> pool._lock`` order.
"""
if fingerprint is None: # HTTP/SSE: no pooled stdio binding may be attached.
return result.pool is None and result.binding is None
binding = result.binding
return pool is not None and result.pool is pool and binding is not None and binding.domain == "deployment" and binding.server_name == name and binding.fingerprint == fingerprint and pool._binding_is_current(binding)
def _evict_server_tool_entries_locked(revision: _AppliedMcpRevision, pool: "MCPSessionPool") -> None:
"""Drop per-server results a newly applied revision no longer justifies.
Caller must hold ``_init_condition``. This runs *after* the caller has
installed the new binding epoch (see ``_apply_mcp_reconciliation_locked``),
so a lifecycle-token rotation that leaves the connection fingerprint
untouched has already replaced the binding object and is caught by the
identity check below. Retained entries keep their exact identity so an
unchanged server's published objects are reused verbatim.
An entry survives only when both hold:
* its canonical config snapshot equals the new revision's
(``stdio_connections`` alone misses presentation-only changes), and
* ``_binding_matches`` proves the result still belongs to the pool's current
deployment binding (HTTP/SSE: no pooled binding at all).
Names absent from ``revision.server_snapshots`` (deleted, disabled or never
declared) are evicted because ``.get`` returns ``None``. The binding verdict
is delegated to ``_binding_matches`` so the reuse and eviction paths cannot
diverge.
"""
snapshots = revision.server_snapshots
stale = [name for name, entry in _server_tool_cache.items() if snapshots.get(name) != entry.snapshot or not _binding_matches(entry.result, name, revision.stdio_connections.get(name), pool)]
for name in stale:
del _server_tool_cache[name]
def _install_claimed_revision_locked(pool, revision: _AppliedMcpRevision, *, previous: _AppliedMcpRevision | None) -> None:
"""Install ``revision``'s deployment epochs before discovery runs.
@ -429,20 +535,24 @@ def _shared_reset_has_local_state_locked(pool) -> bool:
is deliberately not a reason — a deployment-side reset must not retire it.
Caller must hold ``_init_condition``.
"""
return _cache_initialized or _applied_mcp_revision is not None or bool(pool.retained_server_names(domain="deployment"))
return _cache_initialized or _applied_mcp_revision is not None or bool(_server_tool_cache) or bool(pool.retained_server_names(domain="deployment"))
def _has_mcp_reconciliation_state() -> bool:
"""True when this process holds state a config observation could affect.
Published tools, a recorded applied revision and an in-flight discovery are
all reasons to keep watching the config. A process that never initialized
MCP pays no config-read cost (``refresh_mcp_cache_if_active`` contract).
Published tools, a recorded applied revision, an in-flight discovery and a
retained per-server entry are all reasons to keep watching the config. The
entry check matters because ``initialize_mcp_tools`` can null the applied
baseline (an unbuildable revision) without clearing retained entries; without
counting them, a later config change would be silently skipped. A process that
never initialized MCP pays no config-read cost
(``refresh_mcp_cache_if_active`` contract).
"""
from deerflow.mcp.session_pool import current_session_pool
pool = current_session_pool()
return _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or (pool is not None and bool(pool.retained_server_names(domain="deployment")))
return _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(_server_tool_cache) or (pool is not None and bool(pool.retained_server_names(domain="deployment")))
def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None:
@ -536,6 +646,13 @@ def _plan_mcp_reconciliation_locked() -> _McpReconciliation | None:
return _McpReconciliation(_RECONCILE_FULL)
if applied is None:
if _server_tool_cache:
# Retained entries do not record the global interceptor chain they
# were built under, and there is no applied baseline to compare it
# against, so they are unverifiable. Fall back to the conservative
# full reset rather than risk serving a stale interceptor chain.
logger.info("MCP configuration changed with no applicable baseline but retained server entries; resetting conservatively")
return _McpReconciliation(_RECONCILE_FULL)
# No trusted baseline: conservatively re-epoch every active deployment
# binding (same fingerprint included) and tombstone unlisted deployment
# state, without replacing the pool or touching personal state.
@ -609,6 +726,11 @@ def _apply_mcp_reconciliation_locked(plan: _McpReconciliation):
force_rebind=plan.force_rebind,
)
# Evict only after the new epoch is installed: a lifecycle-token rotation
# keeps the fingerprint but replaces the binding object, so evaluating
# reuse against the pre-transition pool would wrongly retain the entry.
_evict_server_tool_entries_locked(revision, pool)
_applied_mcp_revision = revision
_config_path, _config_signature = plan.path, plan.signature
_initialized_without_config = plan.path is None
@ -640,7 +762,7 @@ def _plan_committed_mcp_reconciliation_locked(
from deerflow.mcp.session_pool import get_session_pool
pool = get_session_pool()
has_local_state = _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(pool.retained_server_names(domain="deployment"))
has_local_state = _cache_initialized or _applied_mcp_revision is not None or _initializing_generation is not None or bool(_server_tool_cache) or bool(pool.retained_server_names(domain="deployment"))
if not has_local_state:
# No visible local state, but a durable-task caller can bind between this
# observation and the record below (``ensure_binding`` takes only the pool
@ -668,8 +790,13 @@ def _plan_committed_mcp_reconciliation_locked(
return _McpReconciliation(_RECONCILE_FULL)
if applied is None:
# Mirrors the reader's no-baseline rule: conservatively re-epoch active
# deployment bindings and tombstone unlisted ones, personal untouched.
if _server_tool_cache:
# Mirrors the reader's no-baseline rule: retained entries do not
# record the global interceptor chain they were built under, so
# without a baseline they are unverifiable and force a full reset.
return _McpReconciliation(_RECONCILE_FULL)
# Conservatively re-epoch active deployment bindings and tombstone
# unlisted ones, personal untouched.
return _McpReconciliation(
_RECONCILE_SELECTIVE,
revision=revision,
@ -775,7 +902,7 @@ async def initialize_mcp_tools() -> list[BaseTool]:
"""
global _mcp_tools_cache, _cache_initialized, _config_path, _config_signature
global _initializing_generation, _cache_generation, _mcp_config_snapshot, _initialized_without_config
global _cache_reset_marker_signature, _applied_mcp_revision
global _cache_reset_marker_signature, _applied_mcp_revision, _server_tool_cache
while True:
with _init_condition:
@ -792,13 +919,14 @@ 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.session_pool import StaleMCPBindingError, get_session_pool
from deerflow.mcp.tools import get_mcp_tools
loaded_tools = None
loaded_snapshot = None
loaded_lifecycle = None
loaded_fingerprints: dict[str, str] = {}
loaded_revision = None
reusable_entries: dict[str, _ServerToolCacheEntry] = {}
pending_entries: dict[str, _ServerToolCacheEntry] = {}
candidate_groups: dict[str, ServerDiscoveryResult] = {}
post_path = None
post_sig = None
post_snapshot = None
@ -807,6 +935,10 @@ async def initialize_mcp_tools() -> list[BaseTool]:
post_reset_signature = None
init_succeeded = False
try:
from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.mcp.session_pool import StaleMCPBindingError, get_session_pool
from deerflow.mcp.tools import _flatten_server_tool_groups, get_mcp_tools_by_server
logger.info("Initializing MCP tools...")
# Read the exact revision we hand to discovery. Comparing pre/post file
# snapshots alone cannot prove which revision produced the tools, because
@ -824,6 +956,7 @@ async def initialize_mcp_tools() -> list[BaseTool]:
)
raise RuntimeError("Extensions config could not be loaded for MCP tool discovery") from None
loaded_snapshot = _effective_mcp_config_snapshot(loaded_config)
loaded_fingerprints = _stdio_fingerprints(loaded_config)
try:
loaded_lifecycle = effective_tokens(loaded_config, parse_mcp_lifecycle(loaded_config.model_extra or {}))
except InvalidMcpLifecycle:
@ -900,15 +1033,40 @@ async def initialize_mcp_tools() -> list[BaseTool]:
raise rejection
_install_claimed_revision_locked(claimed_pool, loaded_revision, previous=previous_revision)
_applied_mcp_revision = loaded_revision
# Select the entries this exact revision may reuse. The revision
# and the pool are both fixed under this lock, and ``_binding_matches``
# proves the entry still owns the pool's current binding. An
# untrusted revision (``loaded_revision is None``) reuses nothing.
# Compare each entry's wrapper chain, not the applied baseline:
# a claim can observe an interceptor edit after reconciliation,
# and a failed claim already advanced that baseline on retry.
if loaded_revision is not None:
for name in loaded_config.get_enabled_mcp_servers():
entry = _server_tool_cache.get(name)
if (
entry is not None
and loaded_revision.server_snapshots.get(name) == entry.snapshot
and loaded_revision.interceptors == entry.interceptors
and _binding_matches(entry.result, name, loaded_fingerprints.get(name), claimed_pool)
):
reusable_entries[name] = entry
finally:
if retired_before_claim is not None:
retired_before_claim.close_all_sync()
enabled_names = list(loaded_config.get_enabled_mcp_servers())
missing_names = [name for name in enabled_names if name not in reusable_entries]
discovered_groups: dict[str, ServerDiscoveryResult] = {}
try:
loaded_tools = await get_mcp_tools(
extensions_config=loaded_config,
session_pool=claimed_pool,
)
if missing_names:
# Only the missing servers are re-discovered; ``get_mcp_tools_by_server``
# already isolates per-server failures (a failed server is absent
# from the mapping, a zero-tool success is present with tools=()).
discovered_groups = await get_mcp_tools_by_server(
extensions_config=loaded_config,
server_names=missing_names,
session_pool=claimed_pool,
)
except StaleMCPBindingError:
# The pool this claim owns was retired while discovery was running.
# That is the same cache race as a generation change, so keep the
@ -921,6 +1079,25 @@ async def initialize_mcp_tools() -> list[BaseTool]:
logger.info("MCP cache was reset during binding installation; discarding superseded initialization")
return []
raise
# Local candidate: reused results plus the successful discoveries. A
# failed server is simply absent, so it is neither flattened nor cached.
candidate_groups = {name: entry.result for name, entry in reusable_entries.items()}
candidate_groups.update(discovered_groups)
pending_entries = dict(reusable_entries)
if loaded_revision is not None:
for name, result in discovered_groups.items():
snapshot = loaded_revision.server_snapshots.get(name)
if snapshot is not None:
pending_entries[name] = _ServerToolCacheEntry(snapshot=snapshot, interceptors=loaded_revision.interceptors, result=result)
if enabled_names and not candidate_groups:
# No enabled server produced a successful discovery result: this is
# a total discovery failure, not a legitimately empty configuration
# (a build-invalid connection is never even attempted), so refuse to
# publish a false-success empty cache.
# Raising aborts this attempt (the ``finally`` below releases the
# claim) without spinning inside a single ``get_cached_mcp_tools``.
raise RuntimeError("MCP discovery produced no results for any enabled server")
loaded_tools = _flatten_server_tool_groups(enabled_names, candidate_groups)
post_path, post_sig = _current_config_state()
post_reset_signature = _current_cache_reset_marker_signature(post_path)
if post_path is not None and post_sig is not None:
@ -965,6 +1142,19 @@ async def initialize_mcp_tools() -> list[BaseTool]:
return []
publish = loaded_snapshot is not None and post_snapshot is not None and loaded_snapshot == post_snapshot and loaded_lifecycle == post_lifecycle and loaded_reset_signature == post_reset_signature
if publish:
# Re-validate every published result -- reused or freshly
# discovered -- against the pool's *current* binding. Selection
# and publication share ``_binding_matches``, so a binding
# rotation racing the lock-free discovery makes the whole
# candidate stale: no sibling is published either.
stale_binding = next(
(name for name, result in candidate_groups.items() if not _binding_matches(result, name, loaded_fingerprints.get(name), claimed_pool)),
None,
)
if stale_binding is not None:
logger.warning("MCP binding for server '%s' rotated during discovery; discarding the whole candidate", stale_binding)
publish = False
if not publish:
logger.warning("MCP config or shared reset generation changed during initialization; discarding stale result")
# Reconcile against the revision observed now instead of forcing
@ -976,6 +1166,7 @@ async def initialize_mcp_tools() -> list[BaseTool]:
else:
retired_pool = _apply_mcp_reconciliation_locked(plan)
else:
_server_tool_cache = pending_entries
_mcp_tools_cache = loaded_tools
_cache_initialized = True
_config_path, _config_signature = post_path, post_sig
@ -1060,8 +1251,13 @@ def get_cached_mcp_tools() -> list[BaseTool]:
future.result()
else:
loop.run_until_complete(initialize_mcp_tools())
except Exception:
logger.exception("Failed to lazy-initialize MCP tools")
except Exception as exc:
# Never emit the raw exception message or traceback here. Config
# parsing resolves ``$VAR`` values before validation and discovery
# failures can embed request URLs, so an exception's text may carry
# resolved credentials. The sanitized origins (config load, invalid
# ledger) already log their own bounded detail.
logger.error("Failed to lazy-initialize MCP tools (%s)", type(exc).__name__)
return []
with _init_lock:
@ -1112,6 +1308,7 @@ def _reset_mcp_tools_cache_state() -> None:
global _cache_reset_marker_signature, _applied_mcp_revision
_mcp_tools_cache = None
_server_tool_cache.clear()
_cache_initialized = False
_config_path = None
_config_signature = None

100
backend/tests/test_mcp_cache.py
View File

@ -54,6 +54,7 @@ _TRACKED_GLOBALS = (
"_initialized_without_config",
"_cache_reset_marker_signature",
"_applied_mcp_revision",
"_server_tool_cache",
)
@ -81,6 +82,7 @@ def cache_globals():
cache_module._mcp_tools_cache = None
cache_module._cache_initialized = False
cache_module._server_tool_cache = {}
for name in (
"_config_path",
"_config_signature",
@ -115,6 +117,36 @@ def cache_globals():
setattr(cache_module, name, value)
def _patch_discovery(monkeypatch, flat_fake):
"""Route a flat discovery fake through the grouped per-server entry point.
These cache tests stub ``get_mcp_tools`` with "discovery returns these
tools". Discovery now happens per server, so mirror the live prologue here:
install a
real binding for every selected stdio server and attribute the fake's flat
tool list to the first selected server (the rest succeed with zero tools).
The flattened ordering the tests assert is unchanged.
"""
from deerflow.mcp.client import build_servers_config
from deerflow.mcp.session_pool import get_session_pool, normalized_connection_fingerprint
from deerflow.mcp.tools import ServerDiscoveryResult
async def grouped(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None):
pool = session_pool if session_pool is not None else get_session_pool()
connections = build_servers_config(extensions_config)
selected = [name for name in connections if server_names is None or name in set(server_names)]
tools = await flat_fake(extensions_config=extensions_config, session_pool=pool)
groups = {}
for index, name in enumerate(selected):
connection = connections[name]
is_stdio = connection.get("transport", "stdio") == "stdio"
binding = pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment") if is_stdio else None
groups[name] = ServerDiscoveryResult(tools=tuple(tools) if index == 0 else (), pool=pool if is_stdio else None, binding=binding)
return groups
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools_by_server", grouped)
def _initialize_against(monkeypatch, config_path: Path) -> None:
"""Populate the cache against ``config_path`` via the real init entry point.
@ -127,7 +159,7 @@ def _initialize_against(monkeypatch, config_path: Path) -> None:
async def _fake_get_mcp_tools(**_kwargs):
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
asyncio.run(cache_module.initialize_mcp_tools())
assert cache_module._cache_initialized is True
@ -348,7 +380,7 @@ class TestCrossLoopReinitialization:
async def _fake_tools(**_kwargs):
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
asyncio.run(cache_module.initialize_mcp_tools())
assert cache_module._cache_initialized is True
@ -392,7 +424,7 @@ class TestCrossLoopReinitialization:
await asyncio.sleep(0.01)
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
async def _contended_init():
await asyncio.gather(cache_module.initialize_mcp_tools(), cache_module.initialize_mcp_tools())
@ -436,7 +468,7 @@ class TestCrossLoopReinitialization:
async def _fake_tools(**_kwargs):
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
result1 = cache_module.get_cached_mcp_tools()
assert result1 == []
@ -473,7 +505,7 @@ def test_config_change_during_initialization_discards_stale_tools(cache_globals,
return ["old-tools"]
return ["new-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
first = asyncio.run(cache_module.initialize_mcp_tools())
assert first == []
@ -554,7 +586,7 @@ def test_config_change_during_initialization_reconciles_in_place_for_same_server
_write_extensions_config(cfg, {"same": _server("uvx")})
return [f"session-{session.command}"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
try:
first = asyncio.run(cache_module.initialize_mcp_tools())
@ -597,7 +629,7 @@ def test_superseded_initializer_cannot_seed_replacement_pool(cache_globals, monk
_write_extensions_config(cfg, {"A": _server("old-cmd")})
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
real_get_mcp_tools = tools_module.get_mcp_tools
real_get_mcp_tools_by_server = tools_module.get_mcp_tools_by_server
class FakeClient:
def __init__(self, connections, **kwargs) -> None:
@ -621,9 +653,9 @@ def test_superseded_initializer_cannot_seed_replacement_pool(cache_globals, monk
# while discovery still holds the config revision it read for G0.
entered.set()
await release.wait()
return await real_get_mcp_tools(*args, **kwargs)
return await real_get_mcp_tools_by_server(*args, **kwargs)
monkeypatch.setattr(tools_module, "get_mcp_tools", _gated_get_mcp_tools)
monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _gated_get_mcp_tools)
monkeypatch.setattr("deerflow.mcp.tools.get_initial_oauth_headers", AsyncMock(return_value={}))
monkeypatch.setattr("deerflow.mcp.tools.build_mcp_tool_interceptors", lambda *a, **k: [])
monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient)
@ -673,7 +705,7 @@ def test_reset_mcp_tools_cache_does_not_wait_for_in_flight_initialization(cache_
await asyncio.to_thread(finish.wait)
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
worker = threading.Thread(target=lambda: asyncio.run(cache_module.initialize_mcp_tools()))
worker.start()
@ -724,7 +756,7 @@ def test_automatic_stale_invalidation_retires_session_pool_before_reinitializing
loaded_pools.append(session_pool_module.get_session_pool())
return ["new-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
try:
result = cache_module.get_cached_mcp_tools()
@ -762,7 +794,7 @@ def test_reset_mcp_tools_cache_retires_session_pool_before_releasing_initializer
race_results.append(asyncio.run(cache_module.initialize_mcp_tools()))
return real_reset_session_pool()
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
monkeypatch.setattr(session_pool_module, "reset_session_pool", _reset_with_concurrent_initializer)
try:
@ -798,7 +830,7 @@ def test_cancelled_initializer_releases_generation_claim(cache_globals, monkeypa
await release.wait()
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
async def _cancel_and_retry():
owner = asyncio.create_task(cache_module.initialize_mcp_tools())
@ -844,7 +876,7 @@ def test_skills_only_change_keeps_tools_and_pool(cache_globals, monkeypatch, tmp
async def _fake_get_mcp_tools(**_kwargs):
return [sentinel_tool]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
asyncio.run(cache_module.initialize_mcp_tools())
old_pool = get_session_pool()
@ -972,7 +1004,7 @@ def test_shared_reset_during_initialization_discards_stale_tools(cache_globals,
_write_remote_cache_reset_marker(cfg, "changed-during-discovery")
return ["stale-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
result = asyncio.run(cache_module.initialize_mcp_tools())
@ -1080,7 +1112,7 @@ def test_skills_only_change_during_initialization_publishes_loaded_tools(cache_g
_write_extensions_config(cfg, {"srv1": _server()}, skills={"skill-a": {"enabled": False}})
return ["loaded-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
result = asyncio.run(cache_module.initialize_mcp_tools())
@ -1105,7 +1137,7 @@ def test_refresh_during_initialization_discards_stale_publish(cache_globals, mon
await asyncio.sleep(0.01)
return ["stale-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
async def _run():
owner = asyncio.create_task(cache_module.initialize_mcp_tools())
@ -1130,7 +1162,7 @@ def test_initialization_without_a_readable_snapshot_discards_result(cache_global
async def _fake_get_mcp_tools(**_kwargs):
return ["loaded-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
monkeypatch.setattr(cache_module, "_read_stable_mcp_revision", lambda path, signature: None)
result = asyncio.run(cache_module.initialize_mcp_tools())
@ -1197,7 +1229,7 @@ def test_initialization_hands_the_snapshotted_config_to_discovery(cache_globals,
_write_extensions_config(cfg, {"srv1": _server("npx")})
return ["loaded-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
result = asyncio.run(cache_module.initialize_mcp_tools())
@ -1223,7 +1255,7 @@ def test_config_appearing_after_unconfigured_init_is_stale(cache_globals, monkey
async def _fake_get_mcp_tools(**_kwargs):
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
asyncio.run(cache_module.initialize_mcp_tools())
assert cache_module._cache_initialized is True
@ -1244,7 +1276,7 @@ def test_initialization_with_unreadable_post_signature_discards_result(cache_glo
async def _fake_get_mcp_tools(**_kwargs):
return ["loaded-tools"]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
monkeypatch.setattr(cache_module, "_current_config_state", lambda: (cfg, None))
result = asyncio.run(cache_module.initialize_mcp_tools())
@ -1288,7 +1320,7 @@ def _assert_cache_survives_edit(monkeypatch, cfg, sentinel, rewrite):
async def _fake_get_mcp_tools(**_kwargs):
return [sentinel]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
asyncio.run(cache_module.initialize_mcp_tools())
old_pool = get_session_pool()
@ -1362,7 +1394,7 @@ def test_path_switch_with_equivalent_mcp_config_keeps_tools_and_pool(cache_globa
async def _fake_get_mcp_tools(**_kwargs):
return [sentinel]
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_get_mcp_tools)
_patch_discovery(monkeypatch, _fake_get_mcp_tools)
asyncio.run(cache_module.initialize_mcp_tools())
old_pool = get_session_pool()
@ -1442,7 +1474,7 @@ class TestLazyInitializationFailure:
calls.append(threading.current_thread().name)
raise RuntimeError("discovery failed")
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
return calls
def test_runtime_error_from_discovery_without_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog):
@ -1453,9 +1485,13 @@ class TestLazyInitializationFailure:
assert calls == [threading.current_thread().name]
assert cache_module._cache_initialized is False
logged = [record for record in caplog.records if record.exc_info]
logged = [record for record in caplog.records if record.name == "deerflow.mcp.cache" and record.levelname == "ERROR"]
assert len(logged) == 1
assert str(logged[0].exc_info[1]) == "discovery failed"
# The failure must be attributed to the discovery error by type, never to
# a misleading asyncio loop error -- and the raw message is deliberately
# not emitted because it can carry resolved credentials.
assert "RuntimeError" in logged[0].getMessage()
assert "discovery failed" not in logged[0].getMessage()
def test_runtime_error_from_discovery_inside_running_loop_runs_once(self, cache_globals, monkeypatch, tmp_path, caplog):
calls = self._install_failing_discovery(monkeypatch, tmp_path)
@ -1470,9 +1506,13 @@ class TestLazyInitializationFailure:
assert cache_module._cache_initialized is False
# The logged failure must be the discovery error, not a misleading
# "asyncio.run() cannot be called from a running event loop".
logged = [record for record in caplog.records if record.exc_info]
logged = [record for record in caplog.records if record.name == "deerflow.mcp.cache" and record.levelname == "ERROR"]
assert len(logged) == 1
assert str(logged[0].exc_info[1]) == "discovery failed"
# The failure must be attributed to the discovery error by type, never to
# a misleading asyncio loop error -- and the raw message is deliberately
# not emitted because it can carry resolved credentials.
assert "RuntimeError" in logged[0].getMessage()
assert "discovery failed" not in logged[0].getMessage()
def test_thread_without_event_loop_still_initializes_once(self, cache_globals, monkeypatch, tmp_path):
"""The no-loop fallback stays intact: a worker thread with no loop set initializes via ``asyncio.run``."""
@ -1485,7 +1525,7 @@ class TestLazyInitializationFailure:
calls.append(threading.current_thread().name)
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
outcome: dict[str, object] = {}
def _worker():
@ -1521,7 +1561,7 @@ class TestLazyInitializationFailure:
calls.append(threading.current_thread().name)
return []
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", _fake_tools)
_patch_discovery(monkeypatch, _fake_tools)
closed_loop = asyncio.new_event_loop()
closed_loop.close()
asyncio.set_event_loop(closed_loop)

837
backend/tests/test_mcp_cache_discovery.py Normal file
View File

@ -0,0 +1,837 @@
"""Coverage for the per-server MCP discovery cache.
The cache may reuse a server's already-published ``ServerDiscoveryResult`` only
when all of these still hold:
* the server's canonical configuration identity is unchanged
(``_server_config_snapshot``),
* the normalized global interceptor chain is unchanged, and
* the discovery result is still owned by the pool's *current* session binding
(``_binding_matches``).
This module covers both the reuse predicates and their use by
``initialize_mcp_tools`` (selection, per-server rediscovery, atomic
publication).
``ServerBinding`` is declared ``eq=False``: two bindings with identical field
values are still distinct objects, and ABA safety depends on comparing object
identity (``pool._binding_is_current``), never dataclass value equality. The
superseded-binding test below would still pass under value equality and is
therefore the load-bearing one.
"""
from __future__ import annotations
import asyncio
import builtins
import json
import logging
import threading
import pytest
from langchain_core.tools import StructuredTool
from deerflow.config.extensions_config import ExtensionsConfig
from deerflow.mcp import cache as cache_module
from deerflow.mcp import session_pool as session_pool_module
from deerflow.mcp.cache import (
_AppliedMcpRevision,
_binding_matches,
_ServerToolCacheEntry,
)
from deerflow.mcp.client import build_servers_config
from deerflow.mcp.session_pool import MCPSessionPool, normalized_connection_fingerprint
from deerflow.mcp.tools import ServerDiscoveryResult
def _config(servers: dict[str, dict]) -> ExtensionsConfig:
return ExtensionsConfig.model_validate({"mcpServers": servers})
def _server(config: ExtensionsConfig, name: str):
return config.get_enabled_mcp_servers()[name]
def _snapshot(servers: dict[str, dict], name: str) -> str:
return cache_module._server_config_snapshot(_server(_config(servers), name))
def _pool_with_binding(
name: str = "A",
fingerprint: str = "fp",
domain: str = "deployment",
) -> tuple[MCPSessionPool, object]:
pool = MCPSessionPool()
binding = pool.ensure_binding(name, fingerprint, domain=domain)
return pool, binding
# ---------------------------------------------------------------------------
# Canonical config identity
# ---------------------------------------------------------------------------
def test_server_snapshot_equal_across_transport_alias_and_field_order():
typed = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "args": ["x"]}}, "A")
aliased = _snapshot({"A": {"enabled": True, "transport": "stdio", "command": "srv", "args": ["x"]}}, "A")
reordered = _snapshot({"A": {"command": "srv", "args": ["x"], "enabled": True, "type": "stdio"}}, "A")
# ``transport`` is an MCP-spec alias for ``type`` and JSON key order is
# irrelevant: neither spelling difference may look like a config change.
assert typed == aliased == reordered
def test_server_snapshot_changes_on_presentation_metadata_change():
base = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "description": "one"}}, "A")
changed = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "description": "two"}}, "A")
# Presentation-only fields still change how tools are advertised, so the
# snapshot must distinguish them (metadata-only updates rebuild the tools).
assert base != changed
def test_server_snapshot_changes_on_routing_and_connection_change():
base = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv"}}, "A")
routed = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv", "routing": {"mode": "off", "priority": 5}}}, "A")
reconnected = _snapshot({"A": {"enabled": True, "type": "stdio", "command": "srv2"}}, "A")
assert base != routed
assert base != reconnected
def test_server_snapshot_is_stable_string():
snapshot = _snapshot({"A": {"enabled": True, "type": "http", "url": "https://example.test/mcp"}}, "A")
assert isinstance(snapshot, str) and snapshot
# ---------------------------------------------------------------------------
# Binding ownership identity
# ---------------------------------------------------------------------------
def test_binding_matches_accepts_http_result_without_pool_or_binding():
result = ServerDiscoveryResult(tools=(), pool=None, binding=None)
assert _binding_matches(result, "A", None, None) is True
def test_binding_matches_rejects_http_result_that_carries_ownership():
pool, binding = _pool_with_binding()
assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=None), "A", None, None) is False
assert _binding_matches(ServerDiscoveryResult(tools=(), pool=None, binding=binding), "A", None, None) is False
assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=binding), "A", None, None) is False
def test_binding_matches_accepts_current_stdio_binding():
pool, binding = _pool_with_binding(name="A", fingerprint="fp")
result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding)
assert _binding_matches(result, "A", "fp", pool) is True
def test_binding_matches_rejects_foreign_pool():
pool, binding = _pool_with_binding(name="A", fingerprint="fp")
foreign = MCPSessionPool()
result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding)
assert _binding_matches(result, "A", "fp", foreign) is False
assert _binding_matches(result, "A", "fp", None) is False
def test_binding_matches_rejects_mismatched_pool_and_binding_handles():
owner = MCPSessionPool()
binding = owner.ensure_binding("A", "fp", domain="deployment")
other_pool = MCPSessionPool()
# A result whose pool handle disagrees with the binding's owning pool must
# never be trusted, even when the *passed* pool still considers that exact
# binding current. This makes the ``result.pool is pool`` guard load-bearing.
result = ServerDiscoveryResult(tools=(), pool=other_pool, binding=binding)
assert _binding_matches(result, "A", "fp", owner) is False
def test_binding_matches_rejects_fingerprint_and_name_mismatch():
pool, binding = _pool_with_binding(name="A", fingerprint="fp")
result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding)
assert _binding_matches(result, "A", "other-fp", pool) is False
assert _binding_matches(result, "B", "fp", pool) is False
def test_binding_matches_rejects_superseded_binding_object_with_same_fingerprint():
pool = MCPSessionPool()
stale = pool.ensure_binding("A", "fp", domain="deployment")
# A cross-worker delete/re-add of an identical server re-epochs the binding
# to a *new object* carrying the same fingerprint (the ABA case).
pool.reconcile_bindings({"A": "fp"}, [], domain="deployment", force_rebind=("A",))
current = pool._bindings[("deployment", "A")]
assert current is not stale
assert current.fingerprint == stale.fingerprint
# The stale object has identical field values but is no longer current, so
# value equality must never let it through.
assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=stale), "A", "fp", pool) is False
assert _binding_matches(ServerDiscoveryResult(tools=(), pool=pool, binding=current), "A", "fp", pool) is True
def test_binding_matches_rejects_retired_pool_binding():
pool, binding = _pool_with_binding(name="A", fingerprint="fp")
pool.retire_all()
result = ServerDiscoveryResult(tools=(), pool=pool, binding=binding)
assert _binding_matches(result, "A", "fp", pool) is False
def test_binding_matches_rejects_personal_domain_binding():
pool = MCPSessionPool()
personal = pool.ensure_binding("A", "fp", domain="personal")
result = ServerDiscoveryResult(tools=(), pool=pool, binding=personal)
# A deployment discovery may never reuse a same-named personal binding.
assert _binding_matches(result, "A", "fp", pool) is False
# ---------------------------------------------------------------------------
# Cache entry scaffolding + repr secrecy
# ---------------------------------------------------------------------------
def test_server_tool_cache_starts_as_a_dict():
assert isinstance(cache_module._server_tool_cache, dict)
def test_server_cache_entry_repr_hides_snapshot_and_ownership_handles():
pool, binding = _pool_with_binding(name="A", fingerprint="secret-fingerprint")
entry = _ServerToolCacheEntry(
snapshot='{"env": {"TOKEN": "super-secret-token"}}',
interceptors="[]",
result=ServerDiscoveryResult(tools=(), pool=pool, binding=binding),
)
rendered = repr(entry)
assert "super-secret-token" not in rendered
assert "secret-fingerprint" not in rendered
assert "snapshot" not in rendered
def test_applied_revision_repr_hides_resolved_server_snapshots():
revision = _AppliedMcpRevision(
effective_snapshot='{"env": {"TOKEN": "super-secret-token"}}',
interceptors="[]",
stdio_connections={"A": "digest"},
lifecycle_tokens={},
server_snapshots={"A": '{"env": {"TOKEN": "super-secret-token"}}'},
)
rendered = repr(revision)
assert "super-secret-token" not in rendered
assert "server_snapshots" not in rendered
assert "effective_snapshot" not in rendered
# ---------------------------------------------------------------------------
# Discovery reuse + atomic publication through ``initialize_mcp_tools``
# ---------------------------------------------------------------------------
_DISCOVERY_MISSING = object()
_DISCOVERY_TRACKED_GLOBALS = (
"_mcp_tools_cache",
"_cache_initialized",
"_config_path",
"_config_signature",
"_init_lock",
"_init_condition",
"_initializing_generation",
"_cache_generation",
"_mcp_config_snapshot",
"_initialized_without_config",
"_cache_reset_marker_signature",
"_applied_mcp_revision",
"_server_tool_cache",
)
@pytest.fixture()
def discovery_cache(tmp_path, monkeypatch):
"""Isolate cache coordination state, ``_server_tool_cache`` and the pool singleton."""
saved = {name: getattr(cache_module, name, _DISCOVERY_MISSING) for name in _DISCOVERY_TRACKED_GLOBALS}
cache_module._mcp_tools_cache = None
cache_module._cache_initialized = False
cache_module._server_tool_cache = {}
for name in (
"_config_path",
"_config_signature",
"_mcp_config_snapshot",
"_initialized_without_config",
"_cache_reset_marker_signature",
"_applied_mcp_revision",
):
if hasattr(cache_module, name):
setattr(cache_module, name, None)
cache_module._init_lock = threading.RLock()
cache_module._init_condition = threading.Condition(cache_module._init_lock)
cache_module._initializing_generation = None
cache_module._cache_generation = 0
session_pool_module.reset_session_pool()
cfg = tmp_path / "extensions_config.json"
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
try:
yield cfg
finally:
session_pool_module.reset_session_pool()
for name, value in saved.items():
if value is _DISCOVERY_MISSING:
if hasattr(cache_module, name):
delattr(cache_module, name)
else:
setattr(cache_module, name, value)
def _stdio(name: str, command: str | None = None, **extra) -> dict:
return {"enabled": True, "type": "stdio", "command": command or f"cmd-{name}", **extra}
def _write_config(cfg, servers: dict, *, tokens: dict | None = None, interceptors: list[str] | str | None = None) -> None:
payload: dict = {"mcpServers": servers, "skills": {}}
if tokens is not None:
payload["mcpLifecycle"] = {"version": 1, "servers": tokens}
if interceptors is not None:
payload["mcpInterceptors"] = interceptors
cfg.write_text(json.dumps(payload), encoding="utf-8")
class _DevTool:
def __init__(self, name: str) -> None:
self.name = name
def __repr__(self) -> str: # pragma: no cover - debugging aid
return f"<tool {self.name}>"
class _Discovery:
"""Deterministic grouped-discovery stand-in that records per-call selections."""
def __init__(self, *, empty=(), fail=()) -> None:
self.calls: list[tuple[str, ...]] = []
self.before_return: dict[str, object] = {}
self.empty = set(empty)
self.fail = set(fail)
async def __call__(self, *, extensions_config, server_names=None, session_pool=None, personal_user_id=None):
connections = build_servers_config(extensions_config)
selected = [name for name in connections if server_names is None or name in set(server_names)]
self.calls.append(tuple(selected))
bindings = {}
for name in selected:
connection = connections[name]
if connection.get("transport", "stdio") == "stdio":
bindings[name] = session_pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment")
for name in selected:
hook = self.before_return.get(name)
if hook is not None:
await hook(session_pool)
groups = {}
for name in selected:
if name in self.fail:
continue
is_stdio = connections[name].get("transport", "stdio") == "stdio"
tools = () if name in self.empty else (_DevTool(name),)
groups[name] = ServerDiscoveryResult(tools=tools, pool=session_pool if is_stdio else None, binding=bindings.get(name))
return groups
def _install(monkeypatch, discovery: _Discovery) -> _Discovery:
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools_by_server", discovery)
return discovery
def _names(tools) -> list[str]:
return [getattr(tool, "name", None) for tool in tools]
def _fingerprint(servers: dict, name: str) -> str:
return normalized_connection_fingerprint(build_servers_config(ExtensionsConfig.model_validate({"mcpServers": servers}))[name])
@pytest.mark.asyncio
async def test_unchanged_server_is_reused_by_identity(discovery_cache, monkeypatch):
"""Changing A re-discovers only A; B keeps its exact tool object and binding."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
discovery = _install(monkeypatch, _Discovery())
tools1 = await cache_module.initialize_mcp_tools()
assert _names(tools1) == ["A", "B"]
assert discovery.calls == [("A", "B")]
tool_b = next(tool for tool in tools1 if tool.name == "B")
pool = session_pool_module.get_session_pool()
binding_b = pool._bindings[("deployment", "B")]
_write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools2 = await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A", "B"), ("A",)]
assert _names(tools2) == ["A", "B"]
assert next(tool for tool in tools2 if tool.name == "B") is tool_b
assert pool._bindings[("deployment", "B")] is binding_b
@pytest.mark.asyncio
async def test_metadata_only_change_rebuilds_only_that_server(discovery_cache, monkeypatch):
"""A presentation-only change rebuilds A but reuses B's entry."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
discovery = _install(monkeypatch, _Discovery())
tools1 = await cache_module.initialize_mcp_tools()
tool_a = next(tool for tool in tools1 if tool.name == "A")
tool_b = next(tool for tool in tools1 if tool.name == "B")
_write_config(cfg, {"A": _stdio("A", description="renamed"), "B": _stdio("B")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools2 = await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A", "B"), ("A",)]
assert next(tool for tool in tools2 if tool.name == "A") is not tool_a
assert next(tool for tool in tools2 if tool.name == "B") is tool_b
@pytest.mark.asyncio
async def test_lifecycle_token_change_with_identical_connection_rebuilds_only_a(discovery_cache, monkeypatch):
"""A same-fingerprint lifecycle ABA still re-discovers A and reuses B."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}, tokens={"A": "1" * 32, "B": "2" * 32})
discovery = _install(monkeypatch, _Discovery())
tools1 = await cache_module.initialize_mcp_tools()
tool_b = next(tool for tool in tools1 if tool.name == "B")
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")}, tokens={"A": "3" * 32, "B": "2" * 32})
assert cache_module.refresh_mcp_cache_if_active() is True
tools2 = await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A", "B"), ("A",)]
assert next(tool for tool in tools2 if tool.name == "B") is tool_b
@pytest.mark.asyncio
async def test_declaration_reorder_rediscovers_nothing(discovery_cache, monkeypatch):
"""Reordering servers only re-flattens; zero discovery calls happen."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
discovery = _install(monkeypatch, _Discovery())
tools1 = await cache_module.initialize_mcp_tools()
tool_a = next(tool for tool in tools1 if tool.name == "A")
tool_b = next(tool for tool in tools1 if tool.name == "B")
_write_config(cfg, {"B": _stdio("B"), "A": _stdio("A")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools2 = await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A", "B")]
assert _names(tools2) == ["B", "A"]
assert tools2[0] is tool_b and tools2[1] is tool_a
@pytest.mark.asyncio
async def test_failed_discovery_publishes_siblings_without_caching_failure(discovery_cache, monkeypatch):
"""A failed server is absent (not a cached empty success); its sibling still publishes."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
discovery = _install(monkeypatch, _Discovery())
await cache_module.initialize_mcp_tools()
discovery.fail.add("A")
_write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools = await cache_module.initialize_mcp_tools()
assert _names(tools) == ["B"]
assert "A" not in cache_module._server_tool_cache
assert "B" in cache_module._server_tool_cache
@pytest.mark.asyncio
async def test_zero_tool_success_is_cached_and_not_rediscovered(discovery_cache, monkeypatch):
"""A successful zero-tool discovery is cached and reused (not re-requested)."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
discovery = _install(monkeypatch, _Discovery(empty={"A"}))
await cache_module.initialize_mcp_tools()
assert cache_module._server_tool_cache["A"].result.tools == ()
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B", command="cmd-B2")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools = await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A", "B"), ("B",)]
assert _names(tools) == ["B"]
@pytest.mark.asyncio
async def test_candidate_superseded_by_generation_is_not_published(discovery_cache, monkeypatch):
"""A generation bump mid-discovery discards the whole candidate.
The trigger invalidates the claim without changing the config bytes or
retiring the pool, so the generation check is the only guard that can reject
the candidate (the post-read and binding checks would both still pass).
"""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A")})
discovery = _install(monkeypatch, _Discovery())
async def supersede(_pool):
with cache_module._init_condition:
cache_module._invalidate_published_tools_locked()
discovery.before_return["A"] = supersede
tools = await cache_module.initialize_mcp_tools()
assert tools == []
assert cache_module._cache_initialized is False
@pytest.mark.asyncio
async def test_binding_rotation_during_discovery_discards_candidate(discovery_cache, monkeypatch):
"""A rotated binding for a *reused* sibling fails the whole publication."""
cfg = discovery_cache
servers = {"A": _stdio("A"), "B": _stdio("B")}
_write_config(cfg, servers)
discovery = _install(monkeypatch, _Discovery())
await cache_module.initialize_mcp_tools()
tool_b = cache_module._server_tool_cache["B"].result.tools[0]
_write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")})
assert cache_module.refresh_mcp_cache_if_active() is True
async def rotate_b(pool):
pool.reconcile_bindings({"B": _fingerprint(servers, "B")}, [], domain="deployment", force_rebind=("B",))
discovery.before_return["A"] = rotate_b
tools = await cache_module.initialize_mcp_tools()
assert tools == []
assert cache_module._cache_initialized is False
# The stale B entry must never be re-published.
assert cache_module._server_tool_cache["B"].result.tools[0] is tool_b
@pytest.mark.asyncio
async def test_explicit_reset_forces_full_rediscovery(discovery_cache, monkeypatch):
"""An explicit reset clears entries so the next init discovers every server again."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
discovery = _install(monkeypatch, _Discovery())
await cache_module.initialize_mcp_tools()
cache_module.reset_mcp_tools_cache()
await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A", "B"), ("A", "B")]
@pytest.mark.asyncio
async def test_cancelled_initializer_releases_claim(discovery_cache, monkeypatch):
"""Cancelling mid-discovery releases the claim so the next initializer proceeds."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A")})
discovery = _install(monkeypatch, _Discovery())
entered = asyncio.Event()
release = asyncio.Event()
async def gate(_pool):
entered.set()
await release.wait()
discovery.before_return["A"] = gate
task = asyncio.create_task(cache_module.initialize_mcp_tools())
await asyncio.wait_for(entered.wait(), 2)
task.cancel()
with pytest.raises(asyncio.CancelledError):
await task
assert cache_module._initializing_generation is None
assert cache_module._cache_initialized is False
discovery.before_return.pop("A")
tools = await cache_module.initialize_mcp_tools()
assert _names(tools) == ["A"]
assert cache_module._cache_initialized is True
@pytest.mark.asyncio
async def test_personal_binding_is_never_reused_as_deployment(discovery_cache, monkeypatch):
"""A same-named personal binding must not satisfy the deployment reuse check."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A")})
discovery = _install(monkeypatch, _Discovery())
await cache_module.initialize_mcp_tools()
pool = session_pool_module.get_session_pool()
personal = pool.ensure_binding("A", "personal-fp", domain="personal")
binding_a = pool._bindings[("deployment", "A")]
_write_config(cfg, {"A": _stdio("A", command="cmd-A2")})
assert cache_module.refresh_mcp_cache_if_active() is True
await cache_module.initialize_mcp_tools()
assert discovery.calls == [("A",), ("A",)]
assert pool._bindings[("personal", "A")] is personal
assert pool._bindings[("deployment", "A")] is not binding_a
def _structured_tool(name: str) -> StructuredTool:
async def _call() -> str:
return name
return StructuredTool.from_function(coroutine=_call, name=name, description=name)
@pytest.mark.asyncio
async def test_unchanged_server_oauth_is_not_refetched(discovery_cache, monkeypatch):
"""A partial re-init only fetches OAuth headers for the re-discovered server."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
oauth_requests: list[set[str]] = []
async def fake_headers(extensions_config, *, token_manager=None, server_names=None):
oauth_requests.append(set(server_names or ()))
return {}
monkeypatch.setattr("deerflow.mcp.tools.get_initial_oauth_headers", fake_headers)
monkeypatch.setattr("deerflow.mcp.tools.build_mcp_tool_interceptors", lambda *args, **kwargs: [])
class FakeClient:
def __init__(self, connections, **kwargs) -> None:
self.callbacks = kwargs.get("callbacks") or []
self.tool_interceptors = kwargs.get("tool_interceptors") or []
self.tool_name_prefix = kwargs.get("tool_name_prefix", True)
async def get_tools(self, *, server_name=None):
return [_structured_tool(server_name)]
monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient)
await cache_module.initialize_mcp_tools()
assert oauth_requests[-1] == {"A", "B"}
_write_config(cfg, {"A": _stdio("A", command="cmd-A2"), "B": _stdio("B")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools = await cache_module.initialize_mcp_tools()
assert oauth_requests[-1] == {"A"} # B's OAuth headers are not refetched
assert sorted(tool.name for tool in tools) == ["A", "B"]
def _http(name: str, url: str | None = None) -> dict:
return {"enabled": True, "type": "http", "url": url or f"https://{name.lower()}.test/mcp"}
@pytest.mark.asyncio
@pytest.mark.parametrize(
"before,after,fail_once",
[
([], ["pkg.one:build"], False),
(["pkg.one:build"], [], False),
(["pkg.one:build"], ["pkg.two:build"], False),
(["pkg.one:build", "pkg.two:build"], ["pkg.two:build", "pkg.one:build"], False),
("pkg.one:build", ["pkg.one:build"], False),
([], ["pkg.one:build"], True),
],
ids=["add", "remove", "replace", "reorder", "equivalent-spelling", "failed-claim-retry"],
)
async def test_claim_time_interceptor_change_rewraps_unchanged_servers(discovery_cache, monkeypatch, before, after, fail_once):
"""A global change between reconciliation and the claim must reach every tool."""
from langchain_mcp_adapters.tools import convert_mcp_tool_to_langchain_tool
from mcp.types import CallToolResult, TextContent, Tool
cfg = discovery_cache
servers = {"A": _http("A"), "B": _http("B")}
_write_config(cfg, servers, interceptors=before)
discovered: list[str] = []
invoked: list[tuple[str, str]] = []
fail_discovery = False
class OfflineSession:
async def call_tool(self, name, arguments, **kwargs):
return CallToolResult(content=[TextContent(type="text", text="ok")])
class OfflineClient:
def __init__(self, connections, *, tool_interceptors, **kwargs):
self.tool_interceptors = tool_interceptors
async def get_tools(self, *, server_name):
discovered.append(server_name)
if fail_discovery:
raise RuntimeError("offline discovery failure")
return [
convert_mcp_tool_to_langchain_tool(
OfflineSession(),
Tool(name="probe", inputSchema={"type": "object", "properties": {}}),
server_name=server_name,
tool_interceptors=self.tool_interceptors,
tool_name_prefix=True,
)
]
def resolve_builder(path):
async def interceptor(request, handler):
invoked.append((request.server_name, path))
return await handler(request)
return lambda: interceptor
monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", OfflineClient)
monkeypatch.setattr("deerflow.mcp.tools.resolve_variable", resolve_builder)
initial_tools = await cache_module.initialize_mcp_tools()
initial_b = next(tool for tool in initial_tools if tool.name == "B_probe")
servers["A"] = _http("A", url="https://a.test/changed")
_write_config(cfg, servers, interceptors=before)
assert cache_module.refresh_mcp_cache_if_active() is True
assert "B" in cache_module._server_tool_cache
real_from_file = ExtensionsConfig.from_file
change_pending = True
def change_before_claim(*args, **kwargs):
nonlocal change_pending
if change_pending:
change_pending = False
_write_config(cfg, servers, interceptors=after)
return real_from_file(*args, **kwargs)
monkeypatch.setattr(ExtensionsConfig, "from_file", change_before_claim)
if fail_once:
fail_discovery = True
with pytest.raises(RuntimeError, match="no results"):
await cache_module.initialize_mcp_tools()
assert cache_module._initializing_generation is None
fail_discovery = False
updated_tools = await cache_module.initialize_mcp_tools()
for tool in updated_tools:
await tool.ainvoke({})
assert invoked == [(name, path) for name in ("A", "B") for path in after]
repeated_tools = await asyncio.to_thread(cache_module.get_cached_mcp_tools)
assert all(current is repeated for current, repeated in zip(updated_tools, repeated_tools, strict=True))
invoked.clear()
for tool in repeated_tools:
await tool.ainvoke({})
assert invoked == [(name, path) for name in ("A", "B") for path in after]
changed = ([before] if isinstance(before, str) else before) != after
assert discovered == ["A", "B", *(["A", "B"] if fail_once else []), "A", *(["B"] if changed else [])]
updated_b = next(tool for tool in updated_tools if tool.name == "B_probe")
assert (updated_b is initial_b) is (not changed)
@pytest.mark.asyncio
async def test_interceptor_change_without_baseline_drops_retained_entries(discovery_cache, monkeypatch):
"""With no applied baseline, a retained HTTP entry must not outlive an interceptor change."""
cfg = discovery_cache
_write_config(cfg, {"A": _http("A")})
_install(monkeypatch, _Discovery())
await cache_module.initialize_mcp_tools()
assert "A" in cache_module._server_tool_cache
# Conservative state: entries retained but the comparing baseline is gone.
cache_module._applied_mcp_revision = None
cache_module._cache_initialized = False
cfg.write_text(json.dumps({"mcpServers": {"A": _http("A")}, "skills": {}, "mcpInterceptors": ["pkg.mod:build"]}), encoding="utf-8")
assert cache_module.refresh_mcp_cache_if_active() is True
assert cache_module._server_tool_cache == {}
def test_shared_reset_counts_retained_entries(discovery_cache):
"""A retained per-server entry must make a shared reset consider local state."""
pool = session_pool_module.get_session_pool()
assert cache_module._shared_reset_has_local_state_locked(pool) is False
cache_module._server_tool_cache["A"] = cache_module._ServerToolCacheEntry(
snapshot="orphan",
interceptors="[]",
result=ServerDiscoveryResult(tools=()),
)
assert cache_module._shared_reset_has_local_state_locked(pool) is True
@pytest.mark.asyncio
async def test_total_discovery_failure_does_not_publish_empty_success(discovery_cache, monkeypatch):
"""Every enabled server failing is a phase-level failure, not a cached empty success."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A"), "B": _stdio("B")})
_install(monkeypatch, _Discovery(fail={"A", "B"}))
with pytest.raises(RuntimeError):
await cache_module.initialize_mcp_tools()
assert cache_module._cache_initialized is False
assert cache_module._initializing_generation is None
assert cache_module._server_tool_cache == {}
@pytest.mark.asyncio
async def test_import_failure_after_claim_releases_generation(discovery_cache, monkeypatch):
"""A lazy-import failure must not strand the initialization claim."""
cfg = discovery_cache
_write_config(cfg, {"A": _stdio("A")})
real_import = builtins.__import__
def failing_import(name, *args, **kwargs):
if name == "deerflow.mcp.tools":
raise ImportError("simulated missing adapter module")
return real_import(name, *args, **kwargs)
monkeypatch.setattr(builtins, "__import__", failing_import)
with pytest.raises(ImportError):
await cache_module.initialize_mcp_tools()
assert cache_module._initializing_generation is None
def test_lazy_init_failure_does_not_log_resolved_credentials(discovery_cache, monkeypatch, caplog):
"""A stage-level discovery failure must not print a resolved-credential message."""
cfg = discovery_cache
_write_config(cfg, {"A": _http("A")})
secret = "s3cr3t-pass-123"
async def boom(*args, **kwargs):
raise RuntimeError(f"discovery failed for url 'https://alice:{secret}@remote.test/mcp'")
monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools_by_server", boom)
with caplog.at_level(logging.DEBUG):
assert cache_module.get_cached_mcp_tools() == []
assert secret not in caplog.text
@pytest.mark.asyncio
async def test_committed_interceptor_change_without_baseline_drops_retained_entries(discovery_cache, monkeypatch):
"""The committed-writer path mirrors the reader's no-baseline interceptor guard."""
cfg = discovery_cache
_write_config(cfg, {"A": _http("A")})
_install(monkeypatch, _Discovery())
await cache_module.initialize_mcp_tools()
assert "A" in cache_module._server_tool_cache
cache_module._applied_mcp_revision = None
cache_module._cache_initialized = False
committed = ExtensionsConfig.model_validate({"mcpServers": {"A": _http("A")}, "mcpInterceptors": ["pkg.mod:build"]})
cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg))
assert cache_module._server_tool_cache == {}

368
backend/tests/test_mcp_cache_reconciliation.py
View File

@ -55,6 +55,7 @@ from deerflow.mcp.session_pool import (
normalized_connection_fingerprint,
)
from deerflow.mcp.tasks.runtime import McpTaskConfigurationError, set_mcp_task_config_snapshot
from deerflow.mcp.tools import ServerDiscoveryResult
_MISSING = object()
@ -74,6 +75,7 @@ _TRACKED_GLOBALS = (
"_initialized_without_config",
"_cache_reset_marker_signature",
"_applied_mcp_revision",
"_server_tool_cache",
)
@ -114,6 +116,7 @@ def reconciler():
cache_module._mcp_tools_cache = None
cache_module._cache_initialized = False
cache_module._server_tool_cache = {}
for name in (
"_config_path",
"_config_signature",
@ -214,23 +217,28 @@ def _install_discovery(
create_sessions: bool = True,
hooks: dict | None = None,
):
"""Replace ``tools.get_mcp_tools`` with a deterministic stand-in.
"""Replace the grouped and flat discovery entry points with deterministic stand-ins.
It mirrors the real discovery prologue (``build_servers_config`` +
``ensure_binding`` before the first awaited session creation) so the cache
layer is exercised against the same ordering discovery itself guarantees.
The grouped fake honours ``server_names`` selection, so per-server reuse is
exercised exactly as the initializer drives it.
"""
from deerflow.mcp import tools as tools_module
async def _fake_get_mcp_tools(*, extensions_config, session_pool=None):
async def _run(*, extensions_config, server_names, session_pool):
pool = session_pool if session_pool is not None else session_pool_module.get_session_pool()
servers = build_servers_config(extensions_config)
selected = [name for name in servers if server_names is None or name in set(server_names)]
if hooks is not None:
hooks["servers"] = servers
hooks.setdefault("selected", []).append(tuple(selected))
if before_bindings is not None:
await before_bindings(servers, pool)
bindings = {}
for name, connection in servers.items():
for name in selected:
connection = servers[name]
if connection.get("transport", "stdio") == "stdio":
bindings[name] = pool.ensure_binding(
name,
@ -243,16 +251,26 @@ def _install_discovery(
await after_bindings(servers, pool, bindings)
if fail_after_bindings:
raise RuntimeError("discovery failed after binding install")
if not create_sessions:
return []
sessions = {}
for name, binding in bindings.items():
sessions[name] = await pool.get_session(name, "u:t", servers[name], binding=binding)
if hooks is not None:
hooks["sessions"] = sessions
return [_FakeTool(name) for name in servers]
tool_list: dict[str, _FakeTool] = {}
if create_sessions:
sessions = {}
for name in selected:
if name in bindings:
sessions[name] = await pool.get_session(name, "u:t", servers[name], binding=bindings[name])
tool_list[name] = _FakeTool(name)
if hooks is not None:
hooks["sessions"] = sessions
return {name: ServerDiscoveryResult(tools=((tool_list[name],) if name in tool_list else ()), pool=pool, binding=bindings.get(name)) for name in selected}
async def _fake_get_mcp_tools(*, extensions_config, session_pool=None):
groups = await _run(extensions_config=extensions_config, server_names=None, session_pool=session_pool)
return [tool for name in groups for tool in groups[name].tools]
async def _fake_get_mcp_tools_by_server(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None):
return await _run(extensions_config=extensions_config, server_names=server_names, session_pool=session_pool)
monkeypatch.setattr(tools_module, "get_mcp_tools", _fake_get_mcp_tools)
monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _fake_get_mcp_tools_by_server)
def _write_remote_marker(config_path: Path, generation: str) -> Path:
@ -1689,7 +1707,7 @@ async def test_discovery_failure_does_not_lock_the_generation_claim(reconciler,
async def _boom(**_kwargs):
raise RuntimeError("discovery failed")
monkeypatch.setattr(tools_module, "get_mcp_tools", _boom)
monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _boom)
with pytest.raises(RuntimeError):
await cache_module.initialize_mcp_tools()
@ -1716,7 +1734,7 @@ async def test_stale_binding_error_without_supersession_is_not_swallowed(reconci
async def _conflict(**_kwargs):
raise StaleMCPBindingError("A")
monkeypatch.setattr(tools_module, "get_mcp_tools", _conflict)
monkeypatch.setattr(tools_module, "get_mcp_tools_by_server", _conflict)
with pytest.raises(StaleMCPBindingError):
await cache_module.initialize_mcp_tools()
@ -2466,3 +2484,327 @@ async def test_tombstone_only_change_advances_applied_tokens(reconciler, monkeyp
assert _entry(pool, "B") is entry_b
# ...but the applied token map must not lag the consumed shared version.
assert cache_module._applied_mcp_revision.lifecycle_tokens["A"] == raw_b2[LIFECYCLE_KEY]["servers"]["A"]
# ---------------------------------------------------------------------------
# Selective eviction of per-server discovery results
#
# ``_server_tool_cache`` is populated by the publish path. These tests seed it
# the way a real publish would and then drive the *reconciliation* entry points,
# so eviction is exercised through the same locked transition the cache hot paths
# use - never by calling the eviction helper directly.
# ---------------------------------------------------------------------------
def _server_snapshot(servers: dict, name: str) -> str:
config = ExtensionsConfig.model_validate({"mcpServers": servers})
return cache_module._server_config_snapshot(config.get_enabled_mcp_servers()[name])
def _seed_entry(name: str, snapshot: str, *, pool=None, binding=None):
entry = cache_module._ServerToolCacheEntry(
snapshot=snapshot,
interceptors="[]",
result=ServerDiscoveryResult(tools=(_FakeTool(name),), pool=pool, binding=binding),
)
cache_module._server_tool_cache[name] = entry
return entry
def _seed_stdio_entry(pool, servers: dict, name: str):
return _seed_entry(name, _server_snapshot(servers, name), pool=pool, binding=_binding(pool, name))
def _lifecycle_raw(servers: dict, tokens: dict) -> dict:
from deerflow.mcp.lifecycle import LIFECYCLE_KEY
return {"mcpServers": servers, "skills": {}, LIFECYCLE_KEY: {"version": 1, "servers": tokens}}
async def _initialize_with_ledger(monkeypatch, cfg, servers: dict, log: dict, tokens: dict):
"""Initialize against a config that already carries an explicit lifecycle ledger."""
cfg.write_text(json.dumps(_lifecycle_raw(servers, tokens)), encoding="utf-8")
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(cfg))
_record_sessions(monkeypatch, log)
_install_discovery(monkeypatch, hooks={})
return await cache_module.initialize_mcp_tools()
@pytest.mark.asyncio
async def test_selective_eviction_metadata_change_evicts_only_changed_server(reconciler, monkeypatch, tmp_path):
"""A snapshot-only change drops that server's entry; an unchanged peer is reused by identity."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
entry_a = _seed_stdio_entry(pool, servers, "A")
entry_b = _seed_stdio_entry(pool, servers, "B")
assert entry_a is not entry_b
_write_config(cfg, {"A": _stdio("cmd-A", description="renamed"), "B": _stdio("cmd-B")})
assert cache_module.refresh_mcp_cache_if_active() is True
assert "A" not in cache_module._server_tool_cache
assert cache_module._server_tool_cache["B"] is entry_b
@pytest.mark.asyncio
async def test_selective_eviction_fingerprint_change_evicts_and_rebinds(reconciler, monkeypatch, tmp_path):
"""A connection-identity change invalidates both the binding and the entry."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
binding_a = _binding(pool, "A")
_seed_stdio_entry(pool, servers, "A")
entry_b = _seed_stdio_entry(pool, servers, "B")
_write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")})
assert cache_module.refresh_mcp_cache_if_active() is True
assert "A" not in cache_module._server_tool_cache
assert _binding(pool, "A") is not binding_a
assert cache_module._server_tool_cache["B"] is entry_b
@pytest.mark.asyncio
async def test_selective_eviction_read_side_lifecycle_token_change(reconciler, monkeypatch, tmp_path):
"""A peer worker's token change with an identical connection still evicts A (reader path)."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize_with_ledger(monkeypatch, cfg, servers, _session_log(), {"A": "1" * 32, "B": "2" * 32})
entry_b = _seed_stdio_entry(pool, servers, "B")
_seed_stdio_entry(pool, servers, "A")
binding_a = _binding(pool, "A")
# Identical server config, but a peer advanced A's lifecycle token while B's
# is unchanged, so only A must be re-discovered.
cfg.write_text(json.dumps(_lifecycle_raw(servers, {"A": "3" * 32, "B": "2" * 32})), encoding="utf-8")
assert cache_module.refresh_mcp_cache_if_active() is True
assert "A" not in cache_module._server_tool_cache
assert cache_module._server_tool_cache["B"] is entry_b
assert _binding(pool, "A") is not binding_a
@pytest.mark.asyncio
async def test_selective_eviction_committed_revision_lifecycle_token_change(reconciler, monkeypatch, tmp_path):
"""The committed-writer path evicts A on a token change with an identical connection."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize_with_ledger(monkeypatch, cfg, servers, _session_log(), {"A": "1" * 32, "B": "2" * 32})
entry_b = _seed_stdio_entry(pool, servers, "B")
_seed_stdio_entry(pool, servers, "A")
binding_a = _binding(pool, "A")
cfg.write_text(json.dumps(_lifecycle_raw(servers, {"A": "3" * 32, "B": "2" * 32})), encoding="utf-8")
committed = ExtensionsConfig.from_file(str(cfg))
cache_module.finish_mcp_reconciliation(cache_module.prepare_mcp_reconciliation(committed, config_path=cfg))
assert "A" not in cache_module._server_tool_cache
assert cache_module._server_tool_cache["B"] is entry_b
assert _binding(pool, "A") is not binding_a
@pytest.mark.asyncio
async def test_selective_eviction_removed_server_keeps_others(reconciler, monkeypatch, tmp_path):
"""A deleted server's entry is dropped; a still-declared peer's entry survives."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
_seed_stdio_entry(pool, servers, "A")
entry_b = _seed_stdio_entry(pool, servers, "B")
_write_config(cfg, {"B": _stdio("cmd-B")})
assert cache_module.refresh_mcp_cache_if_active() is True
assert "A" not in cache_module._server_tool_cache
assert cache_module._server_tool_cache["B"] is entry_b
@pytest.mark.asyncio
async def test_selective_eviction_declaration_order_keeps_all_entries(reconciler, monkeypatch, tmp_path):
"""Reordering servers rebuilds the flattened list but reuses every entry by identity."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
entry_a = _seed_stdio_entry(pool, servers, "A")
entry_b = _seed_stdio_entry(pool, servers, "B")
_write_config(cfg, {"B": _stdio("cmd-B"), "A": _stdio("cmd-A")})
assert cache_module.refresh_mcp_cache_if_active() is True
assert cache_module._server_tool_cache["A"] is entry_a
assert cache_module._server_tool_cache["B"] is entry_b
@pytest.mark.asyncio
async def test_skills_only_change_evicts_no_server_entries(reconciler, monkeypatch, tmp_path):
"""A skills-only edit leaves the effective MCP slice untouched, so nothing is evicted."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
entry_a = _seed_stdio_entry(pool, servers, "A")
entry_b = _seed_stdio_entry(pool, servers, "B")
_write_config(cfg, servers, skills={"public": {"enabled": True}})
assert cache_module.refresh_mcp_cache_if_active() is False
assert cache_module._server_tool_cache["A"] is entry_a
assert cache_module._server_tool_cache["B"] is entry_b
@pytest.mark.asyncio
async def test_interceptor_change_clears_all_server_entries(reconciler, monkeypatch, tmp_path):
"""A global interceptor change falls back to the full reset, clearing every entry."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
_seed_stdio_entry(pool, servers, "A")
_seed_stdio_entry(pool, servers, "B")
_write_config(cfg, servers, interceptors=["my.module:interceptor"])
assert cache_module.refresh_mcp_cache_if_active() is True
assert cache_module._server_tool_cache == {}
@pytest.mark.asyncio
async def test_explicit_reset_clears_all_server_entries(reconciler, monkeypatch, tmp_path):
"""The process-local explicit reset clears every server entry."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
_seed_stdio_entry(pool, servers, "A")
_seed_stdio_entry(pool, servers, "B")
cache_module.reset_mcp_tools_cache()
assert cache_module._server_tool_cache == {}
@pytest.mark.asyncio
async def test_shared_reset_marker_clears_all_server_entries(reconciler, monkeypatch, tmp_path):
"""A reset published by another worker clears this worker's entries on observation."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
_seed_stdio_entry(pool, servers, "A")
_seed_stdio_entry(pool, servers, "B")
_write_remote_marker(cfg, "peer-reset")
assert cache_module.refresh_mcp_cache_if_active() is True
assert cache_module._server_tool_cache == {}
@pytest.mark.asyncio
async def test_selective_eviction_drops_entries_for_unlisted_servers(reconciler, monkeypatch, tmp_path):
"""An entry for a server no longer declared can never survive a reconciliation."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
_seed_entry("X", "stale-snapshot", pool=pool, binding=pool.ensure_binding("X", "fpX", domain="deployment"))
entry_b = _seed_stdio_entry(pool, servers, "B")
_write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")})
assert cache_module.refresh_mcp_cache_if_active() is True
assert "X" not in cache_module._server_tool_cache
assert cache_module._server_tool_cache["B"] is entry_b
@pytest.mark.asyncio
async def test_selective_eviction_leaves_personal_domain_untouched(reconciler, monkeypatch, tmp_path):
"""Deployment eviction must not disturb a same-named personal binding/session."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
servers = {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}
await _initialize(monkeypatch, cfg, servers, _session_log())
personal = pool.ensure_binding("A", "personal-fp", domain="personal")
_seed_stdio_entry(pool, servers, "A")
entry_b = _seed_stdio_entry(pool, servers, "B")
_write_config(cfg, {"A": _stdio("cmd-A2"), "B": _stdio("cmd-B")})
assert cache_module.refresh_mcp_cache_if_active() is True
assert "A" not in cache_module._server_tool_cache
assert cache_module._server_tool_cache["B"] is entry_b
assert pool._bindings[("personal", "A")] is personal
def test_reconciliation_state_counts_server_entries_without_baseline(reconciler):
"""A lingering entry must keep config reconciliation active on its own.
``initialize_mcp_tools`` can null ``_applied_mcp_revision`` (unbuildable
revision) without clearing the retained entries, so the state predicate must
count the cache itself or a later config change would be silently skipped.
"""
assert cache_module._has_mcp_reconciliation_state() is False
cache_module._server_tool_cache["A"] = cache_module._ServerToolCacheEntry(
snapshot="orphan",
interceptors="[]",
result=ServerDiscoveryResult(tools=()),
)
assert cache_module._has_mcp_reconciliation_state() is True
@pytest.mark.asyncio
async def test_metadata_change_reuses_b_and_keeps_a_session(reconciler, monkeypatch, tmp_path):
"""A metadata-only change rebuilds A but keeps A's stdio session and reuses B."""
cfg = tmp_path / "extensions_config.json"
pool = MCPSessionPool()
session_pool_module._pool = pool
log = _session_log()
await _initialize(monkeypatch, cfg, {"A": _stdio("cmd-A"), "B": _stdio("cmd-B")}, log)
tool_b = cache_module._server_tool_cache["B"].result.tools[0]
binding_a = _binding(pool, "A")
session_a = _entry(pool, "A")[0]
_write_config(cfg, {"A": _stdio("cmd-A", description="renamed"), "B": _stdio("cmd-B")})
assert cache_module.refresh_mcp_cache_if_active() is True
tools = await cache_module.initialize_mcp_tools()
assert [tool.name for tool in tools] == ["A", "B"]
assert cache_module._server_tool_cache["B"].result.tools[0] is tool_b
assert _binding(pool, "A") is binding_a
assert _entry(pool, "A")[0] is session_a
assert log["created"]["cmd-A"] == 1 # A's stdio process was not respawned
assert session_pool_module.get_session_pool() is pool

36
backend/tests/test_mcp_lifecycle_invalid.py
View File

@ -37,11 +37,21 @@ _INVALID_PROGRAM = textwrap.dedent(
calls = {"n": 0}
async def _fake_get_mcp_tools(**_kwargs):
calls["n"] += 1
return ["should-not-publish"]
from deerflow.mcp.client import build_servers_config
from deerflow.mcp.session_pool import normalized_connection_fingerprint
from deerflow.mcp.tools import ServerDiscoveryResult
tools_module.get_mcp_tools = _fake_get_mcp_tools
async def _fake_by_server(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None):
calls["n"] += 1
groups = {}
for name, connection in build_servers_config(extensions_config).items():
binding = None
if connection.get("transport", "stdio") == "stdio":
binding = session_pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment")
groups[name] = ServerDiscoveryResult(tools=("should-not-publish",), pool=session_pool if binding is not None else None, binding=binding)
return groups
tools_module.get_mcp_tools_by_server = _fake_by_server
result = cache_module.get_cached_mcp_tools()
print(json.dumps({
@ -66,11 +76,21 @@ _RECOVERY_PROGRAM = textwrap.dedent(
cfg = Path(os.environ["DEER_FLOW_EXTENSIONS_CONFIG_PATH"])
calls = {"n": 0}
async def _fake_get_mcp_tools(**_kwargs):
calls["n"] += 1
return ["tool"]
from deerflow.mcp.client import build_servers_config
from deerflow.mcp.session_pool import normalized_connection_fingerprint
from deerflow.mcp.tools import ServerDiscoveryResult
tools_module.get_mcp_tools = _fake_get_mcp_tools
async def _fake_by_server(*, extensions_config, server_names=None, session_pool=None, personal_user_id=None):
calls["n"] += 1
groups = {}
for name, connection in build_servers_config(extensions_config).items():
binding = None
if connection.get("transport", "stdio") == "stdio":
binding = session_pool.ensure_binding(name, normalized_connection_fingerprint(connection), domain="deployment")
groups[name] = ServerDiscoveryResult(tools=("tool",), pool=session_pool if binding is not None else None, binding=binding)
return groups
tools_module.get_mcp_tools_by_server = _fake_by_server
first = cache_module.get_cached_mcp_tools()
first_calls = calls["n"]