mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-07-31 18:36:00 +00:00
fix: isolate MCP server toggles from invalid peer configs (#4577)
* fix: isolate MCP server toggle updates * fix: write extensions config atomically * fix: normalize MCP transport aliases
This commit is contained in:
Huixin615
GitHub
parent
150f7740c7
commit
133a82c6c2
@ -245,6 +245,14 @@ This section accumulates work toward the **2.1.0** milestone
|
||||
|
||||
### Fixed
|
||||
|
||||
- **mcp:** Isolate Settings > Tools enable/disable updates to one MCP server, so
|
||||
an unrelated disallowed stdio command no longer blocks every switch; allow
|
||||
disabling a disallowed target while still rejecting its re-enable, preserve
|
||||
the raw extensions config, honor the MCP-spec `transport` alias when enabling
|
||||
SSE/HTTP servers, surface backend validation details in the UI, and atomically
|
||||
replace the shared config for MCP, skill, and embedded-client updates so
|
||||
interrupted writes cannot leave it truncated.
|
||||
([#4574])
|
||||
- **runtime:** Thread metadata now switches to `running` only after the run passes
|
||||
the startup barrier, so pending-cancelled runs no longer briefly project
|
||||
`running`; clients may observe the prior thread status during worker startup.
|
||||
|
||||
@ -401,6 +401,9 @@ See the [Sandbox Configuration Guide](backend/docs/CONFIGURATION.md#sandbox) to
|
||||
DeerFlow supports configurable MCP servers and skills to extend its capabilities.
|
||||
For HTTP/SSE MCP servers, OAuth token flows are supported (`client_credentials`, `refresh_token`).
|
||||
For stdio MCP servers, per-tool call timeouts can be configured with `tool_call_timeout`.
|
||||
Settings > Tools updates one MCP server at a time: an invalid stdio command on one server no longer blocks toggling another, while enabling that invalid server remains protected by the command allowlist and surfaces the backend validation message in the UI.
|
||||
Targeted updates accept both DeerFlow's `type` field and the MCP-spec `transport` field for SSE/HTTP servers.
|
||||
Runtime MCP and skill updates replace `extensions_config.json` atomically, so an interrupted write cannot leave the shared configuration truncated or partially written.
|
||||
MCP routing hints can also prefer a specific MCP tool for matching requests without forbidding other tools. When `tool_search` defers MCP schemas, matching routing metadata can auto-promote up to `tool_search.auto_promote_top_k` deferred schemas before the model call.
|
||||
See the [MCP Server Guide](backend/docs/MCP_SERVER.md) for detailed instructions.
|
||||
|
||||
|
||||
@ -486,7 +486,7 @@ Localhost persistence deliberately reads the direct request `Host` and ignores `
|
||||
| **Models** (`/api/models`) | `GET /` - list models; `GET /{name}` - model details |
|
||||
| **Features** (`/api/features`) | `GET /` - report config-gated feature availability (`agents_api.enabled`, `browser_control.enabled`) for frontend UI gating |
|
||||
| **Console** (`/api/console`) | Read-only cross-thread observability for the current user (the data layer for an operations dashboard or external monitoring): `GET /stats` - headline counters (runs/threads/agents/tokens/cost); `GET /runs` - paginated run history joined with thread titles (per-run cost); `GET /usage` - zero-filled daily token series + per-model breakdown with spend. Queries `runs`/`threads_meta` directly as a reporting layer (no new `RunStore` methods); requires a SQL database backend — returns 503 on `database.backend: memory`. Real-cost estimation reads optional `models[*].pricing` (`currency`, `input_per_million`, `output_per_million`, `input_cache_hit_per_million`; `ModelConfig` is `extra="allow"`, so no schema change) and prices each run from its `token_usage_by_model` input/output split. Pricing is **cache-aware**: `RunJournal` accumulates prompt-cache hits from `usage_metadata.input_token_details.cache_read` into a sparse `cache_read_tokens` bucket key (also threaded through `SubagentTokenCollector` → `record_external_llm_usage_records`), and cache-hit input tokens are billed at `input_cache_hit_per_million` (omitted → billed at the miss price, a conservative upper bound). All priced models must use one currency; mixed currencies disable cost reporting and leave cost/currency fields null instead of producing invalid aggregates. Legacy rows fall back to run-level totals at `model_name`; unpriced models yield `cost: null` and cost fields are null when no pricing is configured |
|
||||
| **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - update config (saves to extensions_config.json) |
|
||||
| **MCP** (`/api/mcp`) | `GET /config` - get config; `PUT /config` - replace the full config with whole-payload stdio validation; `PATCH /config` - toggle one server while preserving the raw extensions config and validating only an enabled target; both writes reload config and reset the process-local MCP cache |
|
||||
| **Skills** (`/api/skills`) | `GET /` - list skills; `GET /{name}` - details; `PUT /{name}` - update enabled; `POST /install` - install from .skill archive (accepts standard optional frontmatter like `version`, `author`, `compatibility`); `POST /reload` - admin-only process-local prompt-cache invalidation after trusted external filesystem changes |
|
||||
| **Integrations** (`/api/integrations`) | `GET /lark/status` - inspect managed Lark/Feishu CLI integration state, including `sandbox_runtime_mode` / `sandbox_runtime_ready` (whether `lark-cli` will actually be present in the sandbox at chat time); `POST /lark/install` - admin-only install of the official `lark-*` managed skill pack; `POST /lark/config/start` and `/lark/config/complete` - internal first-time Lark connection setup; `POST /lark/auth/start` and `/lark/auth/complete` - browser device-flow user authorization without terminal access, with optional `domains` / exact `scope` for incremental permission grants |
|
||||
| **Memory** (`/api/memory`) | `GET /` - memory data; `POST /reload` - force reload; `GET /config` - config; `GET /status` - config + data |
|
||||
@ -752,7 +752,7 @@ E2B output sync records remote file versions and actual host file metadata in a
|
||||
add a parallel routing middleware for PR1-style preference hints.
|
||||
- **Stdio file outputs**: Persistent stdio sessions are scoped by `user_id:thread_id`. For stdio transports only, DeerFlow pins the subprocess default `cwd` to the thread workspace and `TMPDIR`/`TMP`/`TEMP` to `workspace/.mcp/tmp/`, unless the operator explicitly configured `cwd` or temp env values. SSE/HTTP transports skip this filesystem prep entirely.
|
||||
- **Stdio path translation**: MCP-returned local file references are not copied. If a `ResourceLink` or conservative free-text path resolves to an existing file inside the thread's mounted user-data tree, it is translated deterministically to `/mnt/user-data/...`; paths outside that tree remain unchanged.
|
||||
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (the `PUT /api/mcp/config` reset only clears the cache in its own worker)
|
||||
- **Runtime updates**: Gateway API saves to extensions_config.json; the Gateway-embedded runtime detects changes via the resolved-path + content-signature check above, so multi-worker / stale-mtime deployments still pick up an added/removed MCP server without a restart (`PUT /api/mcp/config` keeps whole-payload validation, while `PATCH /api/mcp/config` changes only one server's `enabled` field, normalizes the same `type`/MCP-spec `transport` alias as the runtime config model, and validates the target only when enabling it; either endpoint's reset clears the cache only in its own worker). MCP, skill, and embedded-client writers share `atomic_write_extensions_config()`, which writes and fsyncs a same-directory temporary file before `os.replace()` and preserves an existing file's mode and symlink target; failed serialization or replacement leaves the prior config intact and cleans up the temporary file.
|
||||
|
||||
### Skills System (`packages/harness/deerflow/skills/`)
|
||||
|
||||
@ -1219,7 +1219,7 @@ Config is env-driven like the others — `MonocleTracingConfig`, built in `get_t
|
||||
- `skills` - Map of skill name → state (enabled)
|
||||
- `middlewares` - Zero-argument `AgentMiddleware` class paths for lead and subagent runtime extension. `config.yaml -> extensions` can override these fields after validation; overrides are replace-per-field, not list concatenation.
|
||||
|
||||
Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and skill state at runtime; `middlewares` remains an operator-controlled config-file extension point.
|
||||
Gateway API endpoints and `DeerFlowClient` methods can modify MCP servers and skill state at runtime; their `extensions_config.json` writes use the shared atomic replacement helper, while `middlewares` remains an operator-controlled config-file extension point.
|
||||
|
||||
### Embedded Client (`packages/harness/deerflow/client.py`)
|
||||
|
||||
|
||||
@ -7,10 +7,19 @@ from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Request, status
|
||||
from pydantic import BaseModel, ConfigDict, Field
|
||||
from pydantic import BaseModel, ConfigDict, Field, model_validator
|
||||
|
||||
from app.gateway.deps import require_admin_user
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, McpRoutingConfig, McpToolOverride, extensions_config_write_lock, get_extensions_config, reload_extensions_config
|
||||
from deerflow.config.extensions_config import (
|
||||
ExtensionsConfig,
|
||||
McpRoutingConfig,
|
||||
McpToolOverride,
|
||||
atomic_write_extensions_config,
|
||||
extensions_config_write_lock,
|
||||
get_extensions_config,
|
||||
normalize_mcp_transport_alias,
|
||||
reload_extensions_config,
|
||||
)
|
||||
from deerflow.mcp.cache import reset_mcp_tools_cache
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
@ -60,6 +69,12 @@ class McpServerConfigResponse(BaseModel):
|
||||
tool_call_timeout: float | None = Field(default=None, description="Timeout in seconds for individual stdio MCP tool calls")
|
||||
model_config = ConfigDict(extra="allow")
|
||||
|
||||
@model_validator(mode="before")
|
||||
@classmethod
|
||||
def _accept_transport_alias(cls, data: Any) -> Any:
|
||||
"""Keep API parsing aligned with the runtime MCP config model."""
|
||||
return normalize_mcp_transport_alias(data)
|
||||
|
||||
|
||||
class McpConfigResponse(BaseModel):
|
||||
"""Response model for MCP configuration."""
|
||||
@ -79,6 +94,17 @@ class McpConfigUpdateRequest(BaseModel):
|
||||
)
|
||||
|
||||
|
||||
class McpServerStateUpdateRequest(BaseModel):
|
||||
"""Request model for enabling or disabling one MCP server."""
|
||||
|
||||
server_name: str = Field(
|
||||
...,
|
||||
min_length=1,
|
||||
description="Name of the MCP server to update",
|
||||
)
|
||||
enabled: bool = Field(..., description="Whether the MCP server is enabled")
|
||||
|
||||
|
||||
class McpCacheResetResponse(BaseModel):
|
||||
"""Response model for resetting the MCP tools cache."""
|
||||
|
||||
@ -383,9 +409,7 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
|
||||
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)
|
||||
atomic_write_extensions_config(config_path, config_data)
|
||||
|
||||
logger.info(f"MCP configuration updated and saved to: {config_path}")
|
||||
|
||||
@ -396,6 +420,43 @@ def _apply_mcp_config_update(body: McpConfigUpdateRequest) -> dict:
|
||||
return reloaded_config.mcp_servers
|
||||
|
||||
|
||||
def _apply_mcp_server_state_update(body: McpServerStateUpdateRequest) -> dict:
|
||||
"""Update one server state while preserving the raw extensions config."""
|
||||
with extensions_config_write_lock:
|
||||
config_path = ExtensionsConfig.resolve_config_path()
|
||||
if config_path is None or not config_path.exists():
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"MCP server '{body.server_name}' not found",
|
||||
)
|
||||
|
||||
with open(config_path, encoding="utf-8") as f:
|
||||
raw_data = json.load(f)
|
||||
|
||||
raw_servers = raw_data.get("mcpServers", {})
|
||||
raw_server = raw_servers.get(body.server_name) if isinstance(raw_servers, dict) else None
|
||||
if not isinstance(raw_server, dict):
|
||||
raise HTTPException(
|
||||
status_code=status.HTTP_404_NOT_FOUND,
|
||||
detail=f"MCP server '{body.server_name}' not found",
|
||||
)
|
||||
|
||||
if body.enabled:
|
||||
target_server = McpServerConfigResponse(**raw_server)
|
||||
_validate_mcp_update_request(
|
||||
McpConfigUpdateRequest(
|
||||
mcp_servers={body.server_name: target_server},
|
||||
)
|
||||
)
|
||||
|
||||
raw_server["enabled"] = body.enabled
|
||||
atomic_write_extensions_config(config_path, raw_data)
|
||||
|
||||
logger.info("MCP server %s enabled state updated to %s", body.server_name, body.enabled)
|
||||
reloaded_config = reload_extensions_config()
|
||||
return reloaded_config.mcp_servers
|
||||
|
||||
|
||||
@router.post(
|
||||
"/mcp/cache/reset",
|
||||
response_model=McpCacheResetResponse,
|
||||
@ -475,3 +536,25 @@ async def update_mcp_configuration(request: Request, body: McpConfigUpdateReques
|
||||
except Exception as e:
|
||||
logger.error(f"Failed to update MCP configuration: {e}", exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to update MCP configuration: {str(e)}")
|
||||
|
||||
|
||||
@router.patch(
|
||||
"/mcp/config",
|
||||
response_model=McpConfigResponse,
|
||||
summary="Update MCP Server State",
|
||||
description="Enable or disable one MCP server without replacing the full extensions configuration.",
|
||||
)
|
||||
async def update_mcp_server_state(request: Request, body: McpServerStateUpdateRequest) -> McpConfigResponse:
|
||||
"""Enable or disable one MCP server and reload the MCP tool cache."""
|
||||
try:
|
||||
await require_admin_user(request, detail=_ADMIN_REQUIRED_DETAIL)
|
||||
reloaded_servers = await asyncio.to_thread(_apply_mcp_server_state_update, body)
|
||||
|
||||
servers = {name: _mask_server_config(McpServerConfigResponse(**server.model_dump())) for name, server in reloaded_servers.items()}
|
||||
reset_mcp_tools_cache()
|
||||
return McpConfigResponse(mcp_servers=servers)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as e:
|
||||
logger.error("Failed to update MCP server %s state: %s", body.server_name, e, exc_info=True)
|
||||
raise HTTPException(status_code=500, detail=f"Failed to update MCP server state: {str(e)}")
|
||||
|
||||
@ -1,5 +1,4 @@
|
||||
import asyncio
|
||||
import json
|
||||
import logging
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
@ -12,7 +11,14 @@ from app.gateway.deps import get_config, require_admin_user
|
||||
from app.gateway.path_utils import resolve_thread_virtual_path
|
||||
from deerflow.agents.lead_agent.prompt import clear_skills_system_prompt_cache, refresh_skills_system_prompt_cache_async, refresh_user_skills_system_prompt_cache_async
|
||||
from deerflow.config.app_config import AppConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, extensions_config_write_lock, get_extensions_config, reload_extensions_config
|
||||
from deerflow.config.extensions_config import (
|
||||
ExtensionsConfig,
|
||||
SkillStateConfig,
|
||||
atomic_write_extensions_config,
|
||||
extensions_config_write_lock,
|
||||
get_extensions_config,
|
||||
reload_extensions_config,
|
||||
)
|
||||
from deerflow.runtime.user_context import get_effective_user_id
|
||||
from deerflow.skills import Skill
|
||||
from deerflow.skills.installer import SkillAlreadyExistsError, SkillSecurityScanError
|
||||
@ -445,8 +451,7 @@ def _write_extensions_skill_state(skill_name: str, enabled: bool) -> None:
|
||||
|
||||
config_data = extensions_config.to_file_dict()
|
||||
|
||||
with open(config_path, "w", encoding="utf-8") as f:
|
||||
json.dump(config_data, f, indent=2)
|
||||
atomic_write_extensions_config(config_path, config_data)
|
||||
|
||||
logger.info(f"Skills configuration updated and saved to: {config_path}")
|
||||
reload_extensions_config()
|
||||
|
||||
@ -393,6 +393,36 @@ deployment needs additional trusted launchers.
|
||||
}
|
||||
```
|
||||
|
||||
#### Update One MCP Server State
|
||||
|
||||
Enable or disable one configured MCP server without replacing the full
|
||||
extensions configuration.
|
||||
|
||||
```http
|
||||
PATCH /api/mcp/config
|
||||
Content-Type: application/json
|
||||
```
|
||||
|
||||
Requires an authenticated admin session. Enabling a `stdio` server validates
|
||||
that server's `command` against the same allowlist used by the full `PUT`
|
||||
endpoint. Disabling a server does not require its command to be allowlisted, and
|
||||
invalid commands on other servers do not block the update. The endpoint
|
||||
preserves secrets, environment-variable placeholders, skills, custom server
|
||||
fields, and other top-level extensions config. SSE/HTTP targets may use either
|
||||
DeerFlow's `type` field or the MCP-spec `transport` field.
|
||||
|
||||
**Request Body:**
|
||||
```json
|
||||
{
|
||||
"server_name": "semantic-scholar",
|
||||
"enabled": false
|
||||
}
|
||||
```
|
||||
|
||||
The response is the full masked MCP configuration, matching `GET` and `PUT`.
|
||||
An unknown `server_name` returns `404`; attempting to enable a server with a
|
||||
disallowed `stdio` command returns `400`.
|
||||
|
||||
#### Reset MCP Tools Cache
|
||||
|
||||
Clear cached MCP tools and persistent MCP sessions process-wide. This affects
|
||||
|
||||
@ -18,12 +18,10 @@ Usage:
|
||||
import asyncio
|
||||
import concurrent.futures
|
||||
import copy
|
||||
import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
import shutil
|
||||
import tempfile
|
||||
import uuid
|
||||
from collections.abc import Generator, Mapping, Sequence
|
||||
from dataclasses import dataclass, field
|
||||
@ -41,7 +39,13 @@ from deerflow.agents.thread_state import get_thread_state_schema, normalize_midd
|
||||
from deerflow.authz.principal import build_principal_from_context
|
||||
from deerflow.config.agents_config import AGENT_NAME_PATTERN
|
||||
from deerflow.config.app_config import get_app_config, is_trace_correlation_enabled, reload_app_config
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, SkillStateConfig, get_extensions_config, reload_extensions_config
|
||||
from deerflow.config.extensions_config import (
|
||||
ExtensionsConfig,
|
||||
SkillStateConfig,
|
||||
atomic_write_extensions_config,
|
||||
get_extensions_config,
|
||||
reload_extensions_config,
|
||||
)
|
||||
from deerflow.config.paths import get_paths
|
||||
from deerflow.models import create_chat_model
|
||||
from deerflow.runtime import CheckpointStateAccessor
|
||||
@ -229,20 +233,7 @@ class DeerFlowClient:
|
||||
@staticmethod
|
||||
def _atomic_write_json(path: Path, data: dict) -> None:
|
||||
"""Write JSON to *path* atomically (temp file + replace)."""
|
||||
fd = tempfile.NamedTemporaryFile(
|
||||
mode="w",
|
||||
dir=path.parent,
|
||||
suffix=".tmp",
|
||||
delete=False,
|
||||
)
|
||||
try:
|
||||
json.dump(data, fd, indent=2)
|
||||
fd.close()
|
||||
Path(fd.name).replace(path)
|
||||
except BaseException:
|
||||
fd.close()
|
||||
Path(fd.name).unlink(missing_ok=True)
|
||||
raise
|
||||
atomic_write_extensions_config(path, data)
|
||||
|
||||
def _get_runnable_config(self, thread_id: str, **overrides) -> RunnableConfig:
|
||||
"""Build a RunnableConfig for agent invocation."""
|
||||
|
||||
@ -3,6 +3,8 @@
|
||||
import json
|
||||
import logging
|
||||
import os
|
||||
import stat
|
||||
import tempfile
|
||||
import threading
|
||||
from pathlib import Path
|
||||
from typing import Any, Literal
|
||||
@ -14,6 +16,15 @@ from deerflow.config.runtime_paths import existing_project_file
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def normalize_mcp_transport_alias(data: Any) -> Any:
|
||||
"""Promote MCP-spec ``transport`` to ``type`` when ``type`` is absent."""
|
||||
if isinstance(data, dict):
|
||||
transport = data.get("transport")
|
||||
if transport and not data.get("type"):
|
||||
return {**data, "type": transport}
|
||||
return data
|
||||
|
||||
|
||||
class McpRoutingConfig(BaseModel):
|
||||
"""Soft routing hints for MCP tool preference."""
|
||||
|
||||
@ -105,11 +116,7 @@ class McpServerConfig(BaseModel):
|
||||
``stdio`` (the default). This validator normalizes the two so either
|
||||
spelling works, with ``type`` taking precedence when both are provided.
|
||||
"""
|
||||
if isinstance(data, dict):
|
||||
transport = data.get("transport")
|
||||
if transport and not data.get("type"):
|
||||
data = {**data, "type": transport}
|
||||
return data
|
||||
return normalize_mcp_transport_alias(data)
|
||||
|
||||
|
||||
def resolve_effective_mcp_routing(server_config: McpServerConfig | None, original_tool_name: str) -> dict[str, Any]:
|
||||
@ -323,6 +330,70 @@ class ExtensionsConfig(BaseModel):
|
||||
_extensions_config: ExtensionsConfig | None = None
|
||||
|
||||
|
||||
def _fsync_directory_best_effort(directory: Path) -> None:
|
||||
"""Persist a directory entry update where the platform supports it."""
|
||||
if os.name == "nt":
|
||||
return
|
||||
|
||||
try:
|
||||
directory_fd = os.open(directory, os.O_RDONLY)
|
||||
except OSError:
|
||||
return
|
||||
|
||||
try:
|
||||
os.fsync(directory_fd)
|
||||
except OSError:
|
||||
logger.debug("Could not fsync extensions config directory: %s", directory, exc_info=True)
|
||||
finally:
|
||||
try:
|
||||
os.close(directory_fd)
|
||||
except OSError:
|
||||
logger.debug("Could not close extensions config directory: %s", directory, exc_info=True)
|
||||
|
||||
|
||||
def atomic_write_extensions_config(path: Path, data: dict[str, Any]) -> None:
|
||||
"""Write extensions config without exposing a truncated or partial file."""
|
||||
path = Path(path)
|
||||
target_path = path.resolve(strict=False) if path.is_symlink() else path
|
||||
target_path.parent.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
existing_mode: int | None = None
|
||||
try:
|
||||
existing_mode = stat.S_IMODE(target_path.stat().st_mode)
|
||||
except FileNotFoundError:
|
||||
pass
|
||||
|
||||
temporary_path: Path | None = None
|
||||
try:
|
||||
with tempfile.NamedTemporaryFile(
|
||||
mode="w",
|
||||
encoding="utf-8",
|
||||
dir=target_path.parent,
|
||||
prefix=f".{target_path.name}.",
|
||||
suffix=".tmp",
|
||||
delete=False,
|
||||
) as temporary_file:
|
||||
temporary_path = Path(temporary_file.name)
|
||||
json.dump(data, temporary_file, indent=2)
|
||||
if existing_mode is not None:
|
||||
temporary_path.chmod(existing_mode)
|
||||
temporary_file.flush()
|
||||
os.fsync(temporary_file.fileno())
|
||||
|
||||
os.replace(temporary_path, target_path)
|
||||
_fsync_directory_best_effort(target_path.parent)
|
||||
finally:
|
||||
if temporary_path is not None:
|
||||
try:
|
||||
temporary_path.unlink(missing_ok=True)
|
||||
except OSError:
|
||||
logger.warning(
|
||||
"Could not remove temporary extensions config file: %s",
|
||||
temporary_path,
|
||||
exc_info=True,
|
||||
)
|
||||
|
||||
|
||||
def get_extensions_config() -> ExtensionsConfig:
|
||||
"""Get the extensions config instance.
|
||||
|
||||
|
||||
@ -1,9 +1,9 @@
|
||||
"""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
|
||||
The PUT and PATCH handlers resolve the extensions config path, probe its
|
||||
existence, read raw JSON, atomically write it, and reload it — all blocking
|
||||
filesystem IO. They offload the whole read-modify-write via
|
||||
``asyncio.to_thread``; if either 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
|
||||
@ -22,7 +22,13 @@ from types import SimpleNamespace
|
||||
import pytest
|
||||
|
||||
from app.gateway.routers import mcp as mcp_router
|
||||
from app.gateway.routers.mcp import McpConfigUpdateRequest, McpServerConfigResponse, update_mcp_configuration
|
||||
from app.gateway.routers.mcp import (
|
||||
McpConfigUpdateRequest,
|
||||
McpServerConfigResponse,
|
||||
McpServerStateUpdateRequest,
|
||||
update_mcp_configuration,
|
||||
update_mcp_server_state,
|
||||
)
|
||||
|
||||
pytestmark = pytest.mark.asyncio
|
||||
|
||||
@ -52,15 +58,40 @@ async def test_update_mcp_configuration_does_not_block_event_loop(tmp_path: Path
|
||||
assert await asyncio.to_thread(config_path.exists)
|
||||
|
||||
|
||||
async def test_concurrent_mcp_updates_are_serialized(tmp_path: Path, monkeypatch) -> None:
|
||||
async def test_update_mcp_server_state_does_not_block_event_loop(tmp_path: Path, monkeypatch) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
await asyncio.to_thread(
|
||||
config_path.write_text,
|
||||
'{"mcpServers":{"remote":{"enabled":false,"transport":"http","url":"https://example.test/mcp"}},"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)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", lambda: None)
|
||||
|
||||
response = await update_mcp_server_state(
|
||||
request=None,
|
||||
body=McpServerStateUpdateRequest(server_name="remote", enabled=True),
|
||||
)
|
||||
|
||||
assert response.mcp_servers["remote"].enabled is True
|
||||
assert response.mcp_servers["remote"].type == "http"
|
||||
|
||||
|
||||
async def test_concurrent_mcp_put_and_patch_updates_are_serialized(tmp_path: Path, 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. ``extensions_config_write_lock`` restores
|
||||
it — and, being shared with the skills router (the other writer of this file),
|
||||
also serializes against skill toggles: even with several concurrent
|
||||
``PUT /api/mcp/config`` calls, only one RMW is inside the critical section at a
|
||||
time. (Without the lock the tracked max concurrency would exceed 1.)
|
||||
mix of ``PUT /api/mcp/config`` and ``PATCH /api/mcp/config`` calls, only one
|
||||
RMW is inside the critical section at a time. (Without the lock the tracked
|
||||
max concurrency would exceed 1.)
|
||||
|
||||
The tracker is injected *inside* the real worker (via ``reload_extensions_config``,
|
||||
the last step under the lock) rather than replacing ``_apply_mcp_config_update``,
|
||||
@ -68,7 +99,11 @@ async def test_concurrent_mcp_updates_are_serialized(tmp_path: Path, monkeypatch
|
||||
the very thing under test.
|
||||
"""
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
await asyncio.to_thread(config_path.write_text, '{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
await asyncio.to_thread(
|
||||
config_path.write_text,
|
||||
'{"mcpServers":{"s":{"enabled":true,"type":"http","url":"https://example.test/mcp"}},"skills":{}}',
|
||||
encoding="utf-8",
|
||||
)
|
||||
monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(config_path))
|
||||
|
||||
async def _noop_admin(_request, **_kwargs) -> None:
|
||||
@ -96,6 +131,12 @@ async def test_concurrent_mcp_updates_are_serialized(tmp_path: Path, monkeypatch
|
||||
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)])
|
||||
await asyncio.gather(
|
||||
*[update_mcp_configuration(request=None, body=body) for _ in range(4)],
|
||||
update_mcp_server_state(
|
||||
request=None,
|
||||
body=McpServerStateUpdateRequest(server_name="s", enabled=False),
|
||||
),
|
||||
)
|
||||
|
||||
assert counters["max"] == 1, f"config updates were not serialized (max concurrency {counters['max']})"
|
||||
|
||||
@ -20,8 +20,8 @@ performs the same RMW on the same file.
|
||||
go back to separate module-local locks.
|
||||
|
||||
Only the config-infra boundaries (storage / ``get_extensions_config`` / reload /
|
||||
path resolution) are stubbed; the real ``open(config_path, "w")`` write to a tmp
|
||||
file is exercised.
|
||||
path resolution) are stubbed; the real same-directory temporary write and atomic
|
||||
replacement are exercised.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -141,7 +141,7 @@ async def test_update_skill_serializes_concurrent_writes(tmp_path: Path, monkeyp
|
||||
update_skill("demo-skill", SkillUpdateRequest(enabled=True), _admin_request(), SimpleNamespace()),
|
||||
)
|
||||
|
||||
# The shared asyncio.Lock must serialize the offloaded read-modify-write.
|
||||
# The shared threading.Lock must serialize the offloaded read-modify-write.
|
||||
assert counters["max"] == 1
|
||||
|
||||
|
||||
|
||||
144
backend/tests/test_extensions_config_atomic_write.py
Normal file
144
backend/tests/test_extensions_config_atomic_write.py
Normal file
@ -0,0 +1,144 @@
|
||||
"""Regression tests for crash-safe extensions config writes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
import os
|
||||
import stat
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from deerflow.config import extensions_config as extensions_config_module
|
||||
from deerflow.config.extensions_config import atomic_write_extensions_config
|
||||
|
||||
|
||||
def _temporary_files_for(path: Path) -> list[Path]:
|
||||
return list(path.parent.glob(f".{path.name}.*.tmp"))
|
||||
|
||||
|
||||
def test_atomic_write_replaces_config_without_leaving_temp_files(tmp_path: Path) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
config_path.write_text('{"old": true}', encoding="utf-8")
|
||||
|
||||
atomic_write_extensions_config(
|
||||
config_path,
|
||||
{
|
||||
"mcpServers": {"github": {"enabled": False}},
|
||||
"skills": {"research": {"enabled": True}},
|
||||
},
|
||||
)
|
||||
|
||||
assert json.loads(config_path.read_text(encoding="utf-8")) == {
|
||||
"mcpServers": {"github": {"enabled": False}},
|
||||
"skills": {"research": {"enabled": True}},
|
||||
}
|
||||
assert _temporary_files_for(config_path) == []
|
||||
|
||||
|
||||
def test_atomic_write_preserves_original_when_json_dump_fails_mid_write(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
original = '{"mcpServers": {"github": {"enabled": true}}, "skills": {}}'
|
||||
config_path.write_text(original, encoding="utf-8")
|
||||
|
||||
def fail_after_partial_write(_data, file_handle, **_kwargs) -> None:
|
||||
file_handle.write('{"mcpServers":')
|
||||
file_handle.flush()
|
||||
raise OSError("disk full")
|
||||
|
||||
monkeypatch.setattr(extensions_config_module.json, "dump", fail_after_partial_write)
|
||||
|
||||
with pytest.raises(OSError, match="disk full"):
|
||||
atomic_write_extensions_config(
|
||||
config_path,
|
||||
{"mcpServers": {"github": {"enabled": False}}, "skills": {}},
|
||||
)
|
||||
|
||||
assert config_path.read_text(encoding="utf-8") == original
|
||||
assert _temporary_files_for(config_path) == []
|
||||
|
||||
|
||||
def test_atomic_write_preserves_original_when_replace_fails(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
original = '{"mcpServers": {}, "skills": {}}'
|
||||
config_path.write_text(original, encoding="utf-8")
|
||||
|
||||
def fail_replace(_source, _destination) -> None:
|
||||
raise OSError("replace failed")
|
||||
|
||||
monkeypatch.setattr(extensions_config_module.os, "replace", fail_replace)
|
||||
|
||||
with pytest.raises(OSError, match="replace failed"):
|
||||
atomic_write_extensions_config(
|
||||
config_path,
|
||||
{"mcpServers": {"github": {"enabled": True}}, "skills": {}},
|
||||
)
|
||||
|
||||
assert config_path.read_text(encoding="utf-8") == original
|
||||
assert _temporary_files_for(config_path) == []
|
||||
|
||||
|
||||
def test_atomic_write_preserves_original_when_file_fsync_fails(
|
||||
monkeypatch: pytest.MonkeyPatch,
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
original = '{"mcpServers": {}, "skills": {}}'
|
||||
config_path.write_text(original, encoding="utf-8")
|
||||
|
||||
def fail_fsync(_file_descriptor) -> None:
|
||||
raise OSError("fsync failed")
|
||||
|
||||
monkeypatch.setattr(extensions_config_module.os, "fsync", fail_fsync)
|
||||
|
||||
with pytest.raises(OSError, match="fsync failed"):
|
||||
atomic_write_extensions_config(
|
||||
config_path,
|
||||
{"mcpServers": {"github": {"enabled": True}}, "skills": {}},
|
||||
)
|
||||
|
||||
assert config_path.read_text(encoding="utf-8") == original
|
||||
assert _temporary_files_for(config_path) == []
|
||||
|
||||
|
||||
@pytest.mark.skipif(os.name == "nt", reason="POSIX mode bits unavailable")
|
||||
def test_atomic_write_preserves_existing_file_mode(tmp_path: Path) -> None:
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
config_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
config_path.chmod(0o640)
|
||||
|
||||
atomic_write_extensions_config(
|
||||
config_path,
|
||||
{"mcpServers": {}, "skills": {"research": {"enabled": False}}},
|
||||
)
|
||||
|
||||
assert stat.S_IMODE(config_path.stat().st_mode) == 0o640
|
||||
|
||||
|
||||
def test_atomic_write_updates_symlink_target_without_replacing_symlink(
|
||||
tmp_path: Path,
|
||||
) -> None:
|
||||
target_path = tmp_path / "actual-extensions-config.json"
|
||||
target_path.write_text('{"mcpServers": {}, "skills": {}}', encoding="utf-8")
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
try:
|
||||
config_path.symlink_to(target_path)
|
||||
except OSError as error:
|
||||
pytest.skip(f"Symlinks are unavailable: {error}")
|
||||
|
||||
atomic_write_extensions_config(
|
||||
config_path,
|
||||
{"mcpServers": {"github": {"enabled": False}}, "skills": {}},
|
||||
)
|
||||
|
||||
assert config_path.is_symlink()
|
||||
assert json.loads(target_path.read_text(encoding="utf-8")) == {
|
||||
"mcpServers": {"github": {"enabled": False}},
|
||||
"skills": {},
|
||||
}
|
||||
@ -3,6 +3,8 @@
|
||||
Verifies that GET /api/mcp/config masks sensitive fields (env values,
|
||||
header values, OAuth secrets) and that PUT /api/mcp/config correctly
|
||||
preserves existing secrets when the frontend round-trips masked values.
|
||||
PATCH /api/mcp/config coverage pins targeted state changes, raw-config
|
||||
preservation, transport aliases, authorization, and command validation.
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@ -21,13 +23,15 @@ from app.gateway.routers.mcp import (
|
||||
McpConfigUpdateRequest,
|
||||
McpOAuthConfigResponse,
|
||||
McpServerConfigResponse,
|
||||
McpServerStateUpdateRequest,
|
||||
_mask_server_config,
|
||||
_merge_preserving_secrets,
|
||||
_validate_mcp_update_request,
|
||||
reset_mcp_tools_cache_endpoint,
|
||||
update_mcp_configuration,
|
||||
update_mcp_server_state,
|
||||
)
|
||||
from deerflow.config.extensions_config import ExtensionsConfig
|
||||
from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _mask_server_config
|
||||
@ -589,6 +593,202 @@ async def test_update_mcp_configuration_preserves_server_extra_fields(monkeypatc
|
||||
assert response.mcp_servers["playwright"].model_extra["api_key"] == "***"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("enabled", [False, True])
|
||||
async def test_update_mcp_server_state_updates_valid_target_despite_unrelated_disallowed_command(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
enabled: bool,
|
||||
):
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
original = {
|
||||
"mcpServers": {
|
||||
"semantic-scholar": {
|
||||
"enabled": True,
|
||||
"type": "stdio",
|
||||
"command": "s2-mcp-server",
|
||||
"env": {"S2_API_KEY": "$S2_API_KEY"},
|
||||
"customFlag": "keep-me",
|
||||
},
|
||||
"github": {
|
||||
"enabled": not enabled,
|
||||
"type": "stdio",
|
||||
"command": "npx",
|
||||
"args": ["-y", "@modelcontextprotocol/server-github"],
|
||||
},
|
||||
},
|
||||
"skills": {"research": {"enabled": False}},
|
||||
"middlewares": ["example.middleware:Middleware"],
|
||||
"customTopLevel": {"preserve": True},
|
||||
}
|
||||
config_path.write_text(json.dumps(original), encoding="utf-8")
|
||||
reset_calls = 0
|
||||
|
||||
def fake_reload_extensions_config():
|
||||
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
||||
|
||||
def fake_reset_mcp_tools_cache():
|
||||
nonlocal reset_calls
|
||||
reset_calls += 1
|
||||
|
||||
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
||||
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
||||
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
||||
|
||||
response = await update_mcp_server_state(
|
||||
_request_with_role("admin"),
|
||||
McpServerStateUpdateRequest(server_name="github", enabled=enabled),
|
||||
)
|
||||
|
||||
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
assert persisted["mcpServers"]["github"]["enabled"] is enabled
|
||||
assert persisted["mcpServers"]["semantic-scholar"] == original["mcpServers"]["semantic-scholar"]
|
||||
assert persisted["skills"] == original["skills"]
|
||||
assert persisted["middlewares"] == original["middlewares"]
|
||||
assert persisted["customTopLevel"] == original["customTopLevel"]
|
||||
assert response.mcp_servers["github"].enabled is enabled
|
||||
assert response.mcp_servers["semantic-scholar"].env == {"S2_API_KEY": "***"}
|
||||
assert reset_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_mcp_server_state_allows_disabling_but_rejects_enabling_disallowed_command(monkeypatch, tmp_path):
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
config_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"mcpServers": {
|
||||
"semantic-scholar": {
|
||||
"enabled": True,
|
||||
"type": "stdio",
|
||||
"command": "s2-mcp-server",
|
||||
}
|
||||
},
|
||||
"skills": {},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
reset_calls = 0
|
||||
|
||||
def fake_reload_extensions_config():
|
||||
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
||||
|
||||
def fake_reset_mcp_tools_cache():
|
||||
nonlocal reset_calls
|
||||
reset_calls += 1
|
||||
|
||||
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
||||
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
||||
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
||||
|
||||
response = await update_mcp_server_state(
|
||||
_request_with_role("admin"),
|
||||
McpServerStateUpdateRequest(server_name="semantic-scholar", enabled=False),
|
||||
)
|
||||
assert response.mcp_servers["semantic-scholar"].enabled is False
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await update_mcp_server_state(
|
||||
_request_with_role("admin"),
|
||||
McpServerStateUpdateRequest(server_name="semantic-scholar", enabled=True),
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "s2-mcp-server" in exc_info.value.detail
|
||||
persisted = json.loads(config_path.read_text(encoding="utf-8"))
|
||||
assert persisted["mcpServers"]["semantic-scholar"]["enabled"] is False
|
||||
assert reset_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
@pytest.mark.parametrize("transport", ["sse", "http"])
|
||||
async def test_update_mcp_server_state_enables_raw_transport_alias(
|
||||
monkeypatch,
|
||||
tmp_path,
|
||||
transport: str,
|
||||
):
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
original_server = {
|
||||
"enabled": False,
|
||||
"transport": transport,
|
||||
"url": "https://mcp.example.com/mcp",
|
||||
"customFlag": "keep-me",
|
||||
}
|
||||
config_path.write_text(
|
||||
json.dumps(
|
||||
{
|
||||
"mcpServers": {"remote": original_server},
|
||||
"skills": {},
|
||||
}
|
||||
),
|
||||
encoding="utf-8",
|
||||
)
|
||||
reset_calls = 0
|
||||
|
||||
def fake_reload_extensions_config():
|
||||
return ExtensionsConfig.model_validate(json.loads(config_path.read_text(encoding="utf-8")))
|
||||
|
||||
def fake_reset_mcp_tools_cache():
|
||||
nonlocal reset_calls
|
||||
reset_calls += 1
|
||||
|
||||
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
||||
monkeypatch.setattr(mcp_router, "reload_extensions_config", fake_reload_extensions_config)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
||||
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
||||
|
||||
response = await update_mcp_server_state(
|
||||
_request_with_role("admin"),
|
||||
McpServerStateUpdateRequest(server_name="remote", enabled=True),
|
||||
)
|
||||
|
||||
persisted_server = json.loads(config_path.read_text(encoding="utf-8"))["mcpServers"]["remote"]
|
||||
assert persisted_server == {**original_server, "enabled": True}
|
||||
assert "type" not in persisted_server
|
||||
assert response.mcp_servers["remote"].enabled is True
|
||||
assert response.mcp_servers["remote"].type == transport
|
||||
assert reset_calls == 1
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_mcp_server_state_returns_404_without_writing_or_resetting_cache(monkeypatch, tmp_path):
|
||||
config_path = tmp_path / "extensions_config.json"
|
||||
original_text = '{"mcpServers": {}, "skills": {}}'
|
||||
config_path.write_text(original_text, encoding="utf-8")
|
||||
reset_calls = 0
|
||||
|
||||
def fake_reset_mcp_tools_cache():
|
||||
nonlocal reset_calls
|
||||
reset_calls += 1
|
||||
|
||||
monkeypatch.setattr(mcp_router.ExtensionsConfig, "resolve_config_path", lambda: config_path)
|
||||
monkeypatch.setattr(mcp_router, "reset_mcp_tools_cache", fake_reset_mcp_tools_cache)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await update_mcp_server_state(
|
||||
_request_with_role("admin"),
|
||||
McpServerStateUpdateRequest(server_name="missing", enabled=True),
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 404
|
||||
assert config_path.read_text(encoding="utf-8") == original_text
|
||||
assert reset_calls == 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_update_mcp_server_state_requires_admin():
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
await update_mcp_server_state(
|
||||
_request_with_role("user"),
|
||||
McpServerStateUpdateRequest(server_name="github", enabled=False),
|
||||
)
|
||||
|
||||
assert exc_info.value.status_code == 403
|
||||
|
||||
|
||||
def test_validate_mcp_update_allows_default_npx_stdio_command(monkeypatch):
|
||||
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
||||
request = McpConfigUpdateRequest(
|
||||
@ -689,3 +889,48 @@ def test_validate_mcp_update_ignores_remote_transports(monkeypatch):
|
||||
)
|
||||
|
||||
_validate_mcp_update_request(request)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("raw_server", "expected_type"),
|
||||
[
|
||||
({"transport": "sse", "url": "https://mcp.example.com/sse"}, "sse"),
|
||||
({"transport": "http", "url": "https://mcp.example.com/mcp"}, "http"),
|
||||
({"transport": "stdio", "command": "npx"}, "stdio"),
|
||||
({"type": "http", "transport": "sse", "url": "https://mcp.example.com/mcp"}, "http"),
|
||||
({}, "stdio"),
|
||||
],
|
||||
)
|
||||
def test_api_and_runtime_mcp_models_normalize_transport_consistently(
|
||||
raw_server: dict[str, object],
|
||||
expected_type: str,
|
||||
):
|
||||
api_server = McpServerConfigResponse.model_validate(raw_server)
|
||||
runtime_server = McpServerConfig.model_validate(raw_server)
|
||||
|
||||
assert api_server.type == expected_type
|
||||
assert runtime_server.type == expected_type
|
||||
assert api_server.type == runtime_server.type
|
||||
if "transport" in raw_server:
|
||||
assert api_server.model_extra["transport"] == raw_server["transport"]
|
||||
assert runtime_server.model_extra["transport"] == raw_server["transport"]
|
||||
|
||||
|
||||
def test_validate_mcp_update_enforces_stdio_transport_alias(monkeypatch):
|
||||
monkeypatch.delenv(_MCP_STDIO_COMMAND_ALLOWLIST_ENV, raising=False)
|
||||
request = McpConfigUpdateRequest.model_validate(
|
||||
{
|
||||
"mcp_servers": {
|
||||
"disallowed": {
|
||||
"transport": "stdio",
|
||||
"command": "custom-mcp-server",
|
||||
}
|
||||
}
|
||||
}
|
||||
)
|
||||
|
||||
with pytest.raises(HTTPException) as exc_info:
|
||||
_validate_mcp_update_request(request)
|
||||
|
||||
assert exc_info.value.status_code == 400
|
||||
assert "custom-mcp-server" in exc_info.value.detail
|
||||
|
||||
@ -72,7 +72,11 @@ The frontend is a stateful chat application. Users create **threads** (conversat
|
||||
Formal artifact content is refreshed once when the run finishes; transient `write-file:` previews remain message-driven.
|
||||
3. `useThreadHistory` loads persisted conversation pages from `GET /api/threads/{id}/messages/page`, preserving the backend's thread-global event `seq`; rendering overlays checkpoint/live copies at their matching canonical identities (a summarized checkpoint may contain a protected early input plus a recent tail). Context-compaction rescue diffs every retained visible identity rather than slicing at the first anchor, and keeps a run-scoped ledger of committed visible messages so replacement updates and repeated rolling checkpoint windows cannot erase an already displayed step. The resolver suppresses checkpoint/transient prefixes whose canonical position is still behind an unloaded cursor page instead of collapsing that unknown gap before a recent anchor, then adds optimistic messages without timestamp re-sorting. History invalidation preserves already-loaded pages so their established ordering positions are not discarded.
|
||||
4. Stop actions call the LangGraph SDK stream stop path; `core/threads/hooks.ts` invalidates current-thread, thread-history, token-usage, and sidebar/search caches immediately and schedules one follow-up refetch because SDK stop may finish via abort + fire-and-forget cancel before backend title finalization commits
|
||||
5. TanStack Query manages server state; localStorage stores user settings
|
||||
5. TanStack Query manages server state; localStorage stores user settings. The
|
||||
Settings > Tools MCP switch calls the targeted `PATCH /api/mcp/config`
|
||||
mutation, disables switches until that mutation's success refetch completes,
|
||||
displays the backend error `detail` through a toast, and invalidates
|
||||
`["mcpConfig"]` only after success.
|
||||
6. Components subscribe to thread state and render updates
|
||||
|
||||
Run duration is run-scoped UI metadata even though the compatibility field `additional_kwargs.turn_duration` is repeated on historical AI messages. `core/messages/run-duration.ts` folds those copies into one display anchored after the run's last visible message group. `MessageList` owns the temporary client-side duration for a just-completed live turn until authoritative history arrives. The duration is total run wall-clock time, not per-message reasoning time; reasoning disclosure and run activity/duration are rendered separately.
|
||||
|
||||
@ -47,7 +47,7 @@ function MCPServerList({
|
||||
servers?: Record<string, MCPServerConfig>;
|
||||
}) {
|
||||
const { t } = useI18n();
|
||||
const { mutate: enableMCPServer } = useEnableMCPServer();
|
||||
const { isPending, mutate: enableMCPServer } = useEnableMCPServer();
|
||||
const entries = Object.entries(servers ?? {});
|
||||
if (entries.length === 0) {
|
||||
return (
|
||||
@ -73,7 +73,9 @@ function MCPServerList({
|
||||
<ItemActions>
|
||||
<Switch
|
||||
checked={config.enabled}
|
||||
disabled={env.NEXT_PUBLIC_STATIC_WEBSITE_ONLY === "true"}
|
||||
disabled={
|
||||
env.NEXT_PUBLIC_STATIC_WEBSITE_ONLY === "true" || isPending
|
||||
}
|
||||
onCheckedChange={(checked) =>
|
||||
enableMCPServer({ serverName: name, enabled: checked })
|
||||
}
|
||||
|
||||
@ -52,3 +52,26 @@ export async function updateMCPConfig(config: MCPConfig) {
|
||||
}
|
||||
return response.json();
|
||||
}
|
||||
|
||||
export async function updateMCPServerState(
|
||||
serverName: string,
|
||||
enabled: boolean,
|
||||
) {
|
||||
const response = await fetch(`${getBackendBaseURL()}/api/mcp/config`, {
|
||||
method: "PATCH",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
server_name: serverName,
|
||||
enabled,
|
||||
}),
|
||||
});
|
||||
if (!response.ok) {
|
||||
throw new MCPConfigRequestError(
|
||||
response.status,
|
||||
await readErrorDetail(response, "Failed to update MCP server"),
|
||||
);
|
||||
}
|
||||
return response.json() as Promise<MCPConfig>;
|
||||
}
|
||||
|
||||
@ -1,6 +1,16 @@
|
||||
import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query";
|
||||
import {
|
||||
type QueryClient,
|
||||
useMutation,
|
||||
useQuery,
|
||||
useQueryClient,
|
||||
} from "@tanstack/react-query";
|
||||
import { toast } from "sonner";
|
||||
|
||||
import { loadMCPConfig, MCPConfigRequestError, updateMCPConfig } from "./api";
|
||||
import {
|
||||
loadMCPConfig,
|
||||
MCPConfigRequestError,
|
||||
updateMCPServerState,
|
||||
} from "./api";
|
||||
|
||||
export function useMCPConfig() {
|
||||
const { data, isLoading, error } = useQuery({
|
||||
@ -12,35 +22,23 @@ export function useMCPConfig() {
|
||||
return { config: data, isLoading, error };
|
||||
}
|
||||
|
||||
interface EnableMCPServerVariables {
|
||||
serverName: string;
|
||||
enabled: boolean;
|
||||
}
|
||||
|
||||
export function getEnableMCPServerMutationOptions(queryClient: QueryClient) {
|
||||
return {
|
||||
mutationFn: ({ serverName, enabled }: EnableMCPServerVariables) =>
|
||||
updateMCPServerState(serverName, enabled),
|
||||
onSuccess: () => queryClient.invalidateQueries({ queryKey: ["mcpConfig"] }),
|
||||
onError: (error: Error) => {
|
||||
toast.error(error.message);
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function useEnableMCPServer() {
|
||||
const queryClient = useQueryClient();
|
||||
const { config } = useMCPConfig();
|
||||
return useMutation({
|
||||
mutationFn: async ({
|
||||
serverName,
|
||||
enabled,
|
||||
}: {
|
||||
serverName: string;
|
||||
enabled: boolean;
|
||||
}) => {
|
||||
if (!config) {
|
||||
throw new Error("MCP config not found");
|
||||
}
|
||||
if (!config.mcp_servers[serverName]) {
|
||||
throw new Error(`MCP server ${serverName} not found`);
|
||||
}
|
||||
await updateMCPConfig({
|
||||
mcp_servers: {
|
||||
...config.mcp_servers,
|
||||
[serverName]: {
|
||||
...config.mcp_servers[serverName],
|
||||
enabled,
|
||||
},
|
||||
},
|
||||
});
|
||||
},
|
||||
onSuccess: () => {
|
||||
void queryClient.invalidateQueries({ queryKey: ["mcpConfig"] });
|
||||
},
|
||||
});
|
||||
return useMutation(getEnableMCPServerMutationOptions(queryClient));
|
||||
}
|
||||
|
||||
@ -0,0 +1,82 @@
|
||||
import { afterEach, describe, expect, it, rs } from "@rstest/core";
|
||||
import { cleanup, fireEvent, render, screen } from "@testing-library/react";
|
||||
|
||||
import { ToolSettingsPage } from "@/components/workspace/settings/tool-settings-page";
|
||||
|
||||
const mcpMockState = rs.hoisted(() => ({
|
||||
isPending: false,
|
||||
mutate: rs.fn(),
|
||||
}));
|
||||
|
||||
rs.mock("@/core/i18n/hooks", () => ({
|
||||
useI18n: () => ({
|
||||
t: {
|
||||
common: { loading: "Loading" },
|
||||
settings: {
|
||||
tools: {
|
||||
title: "Tools",
|
||||
description: "Manage MCP tools",
|
||||
adminRequired: "Admin required",
|
||||
empty: "No tools",
|
||||
},
|
||||
},
|
||||
},
|
||||
}),
|
||||
}));
|
||||
|
||||
rs.mock("@/core/mcp/hooks", () => ({
|
||||
useMCPConfig: () => ({
|
||||
config: {
|
||||
mcp_servers: {
|
||||
github: { enabled: true, description: "GitHub tools" },
|
||||
remote: { enabled: false, description: "Remote tools" },
|
||||
},
|
||||
},
|
||||
isLoading: false,
|
||||
error: null,
|
||||
}),
|
||||
useEnableMCPServer: () => ({
|
||||
isPending: mcpMockState.isPending,
|
||||
mutate: mcpMockState.mutate,
|
||||
}),
|
||||
}));
|
||||
|
||||
rs.mock("@/env", () => ({
|
||||
env: { NEXT_PUBLIC_STATIC_WEBSITE_ONLY: "false" },
|
||||
}));
|
||||
|
||||
afterEach(() => {
|
||||
mcpMockState.isPending = false;
|
||||
mcpMockState.mutate.mockReset();
|
||||
cleanup();
|
||||
});
|
||||
|
||||
describe("ToolSettingsPage MCP switches", () => {
|
||||
it("disables every switch while a targeted update is pending", () => {
|
||||
mcpMockState.isPending = true;
|
||||
|
||||
render(<ToolSettingsPage />);
|
||||
|
||||
const switches = screen.getAllByRole("switch");
|
||||
expect(switches).toHaveLength(2);
|
||||
for (const item of switches) {
|
||||
expect((item as HTMLButtonElement).disabled).toBe(true);
|
||||
}
|
||||
});
|
||||
|
||||
it("submits only the selected server state when idle", () => {
|
||||
render(<ToolSettingsPage />);
|
||||
|
||||
const switches = screen.getAllByRole("switch");
|
||||
const githubSwitch = switches[0];
|
||||
expect(githubSwitch).toBeDefined();
|
||||
expect((githubSwitch as HTMLButtonElement).disabled).toBe(false);
|
||||
|
||||
fireEvent.click(githubSwitch!);
|
||||
|
||||
expect(mcpMockState.mutate).toHaveBeenCalledWith({
|
||||
serverName: "github",
|
||||
enabled: false,
|
||||
});
|
||||
});
|
||||
});
|
||||
@ -28,6 +28,7 @@ import {
|
||||
MCPConfigRequestError,
|
||||
loadMCPConfig,
|
||||
updateMCPConfig,
|
||||
updateMCPServerState,
|
||||
} from "@/core/mcp/api";
|
||||
|
||||
const mockedFetch = rs.mocked(fetcher);
|
||||
@ -116,3 +117,41 @@ describe("updateMCPConfig", () => {
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
describe("updateMCPServerState", () => {
|
||||
test("patches only the requested server state", async () => {
|
||||
mockedFetch.mockResolvedValueOnce(
|
||||
jsonResponse(200, {
|
||||
mcp_servers: { github: { enabled: false } },
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(updateMCPServerState("github", false)).resolves.toEqual({
|
||||
mcp_servers: { github: { enabled: false } },
|
||||
});
|
||||
expect(mockedFetch).toHaveBeenCalledWith("/api/mcp/config", {
|
||||
method: "PATCH",
|
||||
headers: {
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
body: JSON.stringify({
|
||||
server_name: "github",
|
||||
enabled: false,
|
||||
}),
|
||||
});
|
||||
});
|
||||
|
||||
test("surfaces backend validation detail", async () => {
|
||||
const detail =
|
||||
"MCP server 'semantic-scholar' uses disallowed stdio command 's2-mcp-server'.";
|
||||
mockedFetch.mockResolvedValueOnce(jsonResponse(400, { detail }));
|
||||
|
||||
await expect(
|
||||
updateMCPServerState("semantic-scholar", true),
|
||||
).rejects.toMatchObject({
|
||||
name: "MCPConfigRequestError",
|
||||
status: 400,
|
||||
message: detail,
|
||||
});
|
||||
});
|
||||
});
|
||||
|
||||
@ -1,14 +1,23 @@
|
||||
import { beforeEach, describe, expect, it, rs } from "@rstest/core";
|
||||
import { QueryClient } from "@tanstack/react-query";
|
||||
import { toast } from "sonner";
|
||||
|
||||
rs.mock("@/core/api/fetcher", () => ({
|
||||
fetch: rs.fn(),
|
||||
}));
|
||||
|
||||
rs.mock("sonner", () => ({
|
||||
toast: {
|
||||
error: rs.fn(),
|
||||
},
|
||||
}));
|
||||
|
||||
import { fetch } from "@/core/api/fetcher";
|
||||
import { MCPConfigRequestError, loadMCPConfig } from "@/core/mcp/api";
|
||||
import { getEnableMCPServerMutationOptions } from "@/core/mcp/hooks";
|
||||
|
||||
const mockedFetch = rs.mocked(fetch);
|
||||
const mockedToastError = rs.mocked(toast.error);
|
||||
|
||||
function makeClient() {
|
||||
return new QueryClient({
|
||||
@ -23,6 +32,7 @@ function makeClient() {
|
||||
describe("useMCPConfig retry policy", () => {
|
||||
beforeEach(() => {
|
||||
mockedFetch.mockReset();
|
||||
mockedToastError.mockReset();
|
||||
});
|
||||
|
||||
it("does not retry when loadMCPConfig throws MCPConfigRequestError (403)", async () => {
|
||||
@ -82,3 +92,50 @@ describe("useMCPConfig retry policy", () => {
|
||||
expect(mockedFetch).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
});
|
||||
|
||||
describe("MCP server state mutation", () => {
|
||||
beforeEach(() => {
|
||||
mockedFetch.mockReset();
|
||||
mockedToastError.mockReset();
|
||||
});
|
||||
|
||||
it("invalidates MCP config after a successful targeted update", async () => {
|
||||
mockedFetch.mockResolvedValue(
|
||||
new Response(JSON.stringify({ mcp_servers: {} }), { status: 200 }),
|
||||
);
|
||||
const client = makeClient();
|
||||
const invalidateQueries = rs
|
||||
.spyOn(client, "invalidateQueries")
|
||||
.mockResolvedValue();
|
||||
const mutation = client
|
||||
.getMutationCache()
|
||||
.build(client, getEnableMCPServerMutationOptions(client));
|
||||
|
||||
await mutation.execute({ serverName: "github", enabled: false });
|
||||
|
||||
expect(invalidateQueries).toHaveBeenCalledWith({
|
||||
queryKey: ["mcpConfig"],
|
||||
});
|
||||
expect(mockedToastError).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("shows the backend error detail when a targeted update fails", async () => {
|
||||
const detail =
|
||||
"MCP server 'semantic-scholar' uses disallowed stdio command 's2-mcp-server'.";
|
||||
mockedFetch.mockResolvedValue(
|
||||
new Response(JSON.stringify({ detail }), { status: 400 }),
|
||||
);
|
||||
const client = makeClient();
|
||||
const invalidateQueries = rs.spyOn(client, "invalidateQueries");
|
||||
const mutation = client
|
||||
.getMutationCache()
|
||||
.build(client, getEnableMCPServerMutationOptions(client));
|
||||
|
||||
await expect(
|
||||
mutation.execute({ serverName: "semantic-scholar", enabled: true }),
|
||||
).rejects.toThrow(detail);
|
||||
|
||||
expect(mockedToastError).toHaveBeenCalledWith(detail);
|
||||
expect(invalidateQueries).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user