github/deer-flow
1
0
Fork 0
mirror of https://github.com/bytedance/deer-flow.git synced 2026-07-27 00:17:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(mcp): offload blocking filesystem IO in MCP config update (#3552)

Browse Source 
* fix(mcp): offload blocking filesystem IO in MCP config update

update_mcp_configuration resolved the extensions config path, probed its
existence, read the raw JSON, wrote the merged config, and reloaded it — all
blocking filesystem IO on the event loop (PUT /api/mcp/config). The whole
read-modify-write after the async admin check has no interleaved awaits, so it
moves into one _apply_mcp_config_update helper dispatched via asyncio.to_thread;
the masked response is built on the loop. The secret-preserving merge, error
codes, and the stdio command allowlist are unchanged.

Found via `make detect-blocking-io`. Same class as #3457 / #3529 / #3551.

Add tests/blocking_io/test_mcp_router.py anchor, verified red->green under the
strict Blockbuster gate.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(mcp): serialize concurrent config updates with a write lock

Address review on #3552: offloading the read-modify-write to a worker thread
dropped the implicit serialization the single-threaded event loop provided, so
two concurrent PUT /api/mcp/config calls could interleave and clobber each
other. Guard the offloaded RMW with a module-level asyncio.Lock to restore
within-process atomicity (cross-process writers remain a separate, pre-existing
concern).

Add a serialization regression test (red->green: without the lock the tracked
max concurrency exceeds 1).

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix: address mcp config blocking io review

---------

Co-authored-by: ly-wang19 <ly-wang19@users.noreply.github.com>
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: Willem Jiang <willem.jiang@gmail.com>
This commit is contained in:
ly-wang19 2026-07-15 20:41:56 +08:00 committed by GitHub
parent a0acdda103
commit 289adcbb02
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
2 changed files with 170 additions and 52 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

130
backend/app/gateway/routers/mcp.py
View File

@ -1,3 +1,4 @@
import asyncio
import json
import logging
import os
@ -15,6 +16,12 @@ from deerflow.mcp.cache import reset_mcp_tools_cache
logger = logging.getLogger(__name__)
router = APIRouter(prefix="/api", tags=["mcp"])
# Serializes the read-modify-write of extensions_config.json within this worker
# process. Offloading the RMW to a thread removed the implicit serialization the
# single-threaded event loop used to provide, so two concurrent
# PUT /api/mcp/config calls could otherwise interleave and clobber each other.
# (Cross-process writers remain a separate, pre-existing concern.)
_mcp_config_write_lock = asyncio.Lock()
_ADMIN_REQUIRED_DETAIL = "Admin privileges required to manage MCP configuration."
@ -330,6 +337,70 @@ async def get_mcp_configuration(request: Request) -> McpConfigResponse:
return McpConfigResponse(mcp_servers=servers)
def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
"""Worker-thread body for :func:`update_mcp_configuration`.
Resolving the config path, the existence probe, reading the raw JSON,
writing the merged config, and reloading it are all blocking filesystem IO
that must stay off the event loop. The merge is pure in-memory work but
lives here too so the whole read-modify-write is a single worker hop.
Returns the reloaded MCP server configs for the response.
"""
# Get the current config path (or determine where to save it)
config_path = ExtensionsConfig.resolve_config_path()
# If no config file exists, create one in the parent directory (project root)
if config_path is None:
config_path = Path.cwd().parent / "extensions_config.json"
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
# Load current config to preserve skills
current_config = get_extensions_config()
# Load raw (un-resolved) JSON from disk to use as the merge source.
# This preserves $VAR placeholders in env values and top-level keys
# like mcpInterceptors that would otherwise be lost.
raw_servers: dict[str, dict] = {}
raw_other_keys: dict = {}
if config_path is not None and config_path.exists():
with open(config_path, encoding="utf-8") as f:
raw_data = json.load(f)
raw_servers = raw_data.get("mcpServers", {})
# Preserve any top-level keys beyond mcpServers/skills
for key, value in raw_data.items():
if key not in ("mcpServers", "skills"):
raw_other_keys[key] = value
# Merge incoming server configs with raw on-disk secrets
merged_servers: dict[str, McpServerConfigResponse] = {}
for name, incoming in body.mcp_servers.items():
raw_server = raw_servers.get(name)
if raw_server is not None:
merged_servers[name] = _merge_preserving_secrets(
incoming,
McpServerConfigResponse(**raw_server),
)
else:
merged_servers[name] = incoming
# Build config data preserving all top-level keys from the original file
config_data = dict(raw_other_keys)
config_data["mcpServers"] = {name: server.model_dump() for name, server in merged_servers.items()}
config_data["skills"] = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()}
# Write the configuration to file
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config_data, f, indent=2)
logger.info(f"MCP configuration updated and saved to: {config_path}")
# Reload the Gateway configuration and update the global cache. The
# agent runtime lives in Gateway, so this keeps API reads and tool
# execution aligned after extensions_config.json changes.
reloaded_config = reload_extensions_config()
return reloaded_config.mcp_servers
@router.post(
"/mcp/cache/reset",
response_model=McpCacheResetResponse,
@ -393,60 +464,15 @@ async def update_mcp_configuration(request: Request, body: McpConfigUpdateReques
await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL)
_validate_mcp_update_request(body)
# Get the current config path (or determine where to save it)
config_path = ExtensionsConfig.resolve_config_path()
# Offload the blocking read-modify-write of extensions_config.json
# (path resolve, existence probe, raw read, merged write, reload). The
# lock serializes concurrent updates within this process so the RMW stays
# atomic now that it no longer runs inline on the event loop.
async with _mcp_config_write_lock:
reloaded_servers = await asyncio.to_thread(_apply_mcp_config_update, body)
# If no config file exists, create one in the parent directory (project root)
if config_path is None:
config_path = Path.cwd().parent / "extensions_config.json"
logger.info(f"No existing extensions config found. Creating new config at: {config_path}")
# Load current config to preserve skills
current_config = get_extensions_config()
# Load raw (un-resolved) JSON from disk to use as the merge source.
# This preserves $VAR placeholders in env values and top-level keys
# like mcpInterceptors that would otherwise be lost.
raw_servers: dict[str, dict] = {}
raw_other_keys: dict = {}
if config_path is not None and config_path.exists():
with open(config_path, encoding="utf-8") as f:
raw_data = json.load(f)
raw_servers = raw_data.get("mcpServers", {})
# Preserve any top-level keys beyond mcpServers/skills
for key, value in raw_data.items():
if key not in ("mcpServers", "skills"):
raw_other_keys[key] = value
# Merge incoming server configs with raw on-disk secrets
merged_servers: dict[str, McpServerConfigResponse] = {}
for name, incoming in body.mcp_servers.items():
raw_server = raw_servers.get(name)
if raw_server is not None:
merged_servers[name] = _merge_preserving_secrets(
incoming,
McpServerConfigResponse(**raw_server),
)
else:
merged_servers[name] = incoming
# Build config data preserving all top-level keys from the original file
config_data = dict(raw_other_keys)
config_data["mcpServers"] = {name: server.model_dump() for name, server in merged_servers.items()}
config_data["skills"] = {name: {"enabled": skill.enabled} for name, skill in current_config.skills.items()}
# Write the configuration to file
with open(config_path, "w", encoding="utf-8") as f:
json.dump(config_data, f, indent=2)
logger.info(f"MCP configuration updated and saved to: {config_path}")
# Reload the Gateway configuration and update the global cache. The
# agent runtime lives in Gateway, so this keeps API reads and tool
# execution aligned after extensions_config.json changes.
reloaded_config = reload_extensions_config()
servers = {name: _mask_server_config(McpServerConfigResponse(**server.model_dump())) for name, server in reloaded_servers.items()}
reset_mcp_tools_cache()
servers = {name: _mask_server_config(McpServerConfigResponse(**server.model_dump())) for name, server in reloaded_config.mcp_servers.items()}
return McpConfigResponse(mcp_servers=servers)
except HTTPException:

92
backend/tests/blocking_io/test_mcp_router.py Normal file
View File

@ -0,0 +1,92 @@
"""Regression anchor: updating MCP config must not block the event loop.
``update_mcp_configuration`` resolves the extensions config path, probes its
existence, reads the raw JSON, writes the merged config, and reloads it — all
blocking filesystem IO. The handler offloads the whole read-modify-write via
``asyncio.to_thread``; if it regresses back onto the event loop, the strict
Blockbuster gate raises ``BlockingError`` and this test fails.
The admin check is patched to a no-op so the anchor exercises the handler's own
filesystem IO, not the authz layer. Imports sit at module top so any import-time
IO runs at collection, outside the gate.
"""
from __future__ import annotations
import asyncio
import threading
import time
from pathlib import Path
import pytest
from app.gateway.routers import mcp as mcp_router
from app.gateway.routers.mcp import McpConfigUpdateRequest, McpServerConfigResponse, update_mcp_configuration
pytestmark = pytest.mark.asyncio
async def test_update_mcp_configuration_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
config_path = tmp_path / "extensions_config.json"
# resolve_config_path() requires the env-pointed file to exist; seed a minimal one.
await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8")
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
async def _noop_admin(_request, **_kwargs) -> None:
return None
monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin)
# An http transport skips the stdio command allowlist check, so the anchor
# stays focused on the filesystem offload rather than command validation.
body = McpConfigUpdateRequest(
mcp_servers={"test-server": McpServerConfigResponse(type="http", url="https://example.test/mcp", description="anchor")},
)
resp = await update_mcp_configuration(request=None, body=body)
assert "test-server" in resp.mcp_servers
# The merged config was actually written to the env-pointed path (offload the
# stat so the assertion itself doesn't trip the gate).
assert await asyncio.to_thread(config_path.exists)
async def test_concurrent_mcp_updates_are_serialized(monkeypatch) -> None:
"""The write lock keeps the offloaded read-modify-write atomic within the process.
Offloading the RMW to a worker thread dropped the implicit serialization the
single-threaded event loop provided. ``_mcp_config_write_lock`` restores it:
even with several concurrent ``PUT /api/mcp/config`` calls, only one
``_apply_mcp_config_update`` runs at a time. (Without the lock the tracked
max concurrency would exceed 1.)
"""
async def _noop_admin(_request, **_kwargs) -> None:
return None
monkeypatch.setattr(mcp_router, "require_admin_user", _noop_admin)
monkeypatch.setattr(mcp_router, "_validate_mcp_update_request", lambda _body: None)
state_lock = threading.Lock()
active = 0
max_active = 0
def _tracking_apply(_body) -> dict:
nonlocal active, max_active
with state_lock:
active += 1
max_active = max(max_active, active)
time.sleep(0.02) # worker thread (off-loop): hold long enough to expose any overlap
with state_lock:
active -= 1
return {}
monkeypatch.setattr(mcp_router, "_apply_mcp_config_update", _tracking_apply)
body = McpConfigUpdateRequest(
mcp_servers={"s": McpServerConfigResponse(type="http", url="https://example.test/mcp")},
)
await asyncio.gather(*[update_mcp_configuration(request=None, body=body) for _ in range(5)])
assert max_active == 1, f"config updates were not serialized (max concurrency {max_active})"