diff --git a/README.md b/README.md index 121628c7ce..ac4a4f33b0 100644 --- a/README.md +++ b/README.md @@ -645,6 +645,9 @@ Signed-in users' notification toggle, default model, conversation mode, and reas Capability Center groups plugins by office collaboration, documents and knowledge, search and research, business and data, and development and operations. The directory includes setup references alongside existing MCP configurations and Lark. Recommended integrations and built-in support do not imply an installed or verified connection; the Installed filter shows configured MCP entries and installed Lark only. +Personal MCP connections configured in the web interface are persisted per user. +Deployment tools remain shared. See [connection ownership](docs/capability-center.md#personal-and-deployment-mcp-configuration). + For plugin manifests, adapter registration, and Agent capability selection, see [Capability Center integration contract](docs/capability-center.md). @@ -655,7 +658,7 @@ contacts. Saving configuration performs no external write. These plugins reuse the existing MCP lifecycle and require no separate plugin service. See the integration contract above for required fields, scopes, and feature boundaries. -Plugin brand icons are bundled locally. When adding or editing one MCP plugin, administrators can upload a PNG, JPG, or WebP image (up to 2 MB), preview it, or restore the default icon. Changes take effect only after Save; custom icons persist across browsers as a normalized 128px PNG in the server entry's display-only `presentation.icon` metadata. They are not sent to the MCP transport. +Plugin brand icons are bundled locally. When adding or editing one personal MCP plugin, users can upload a PNG, JPG, or WebP image (up to 2 MB), preview it, or restore the default icon. Changes take effect only after Save; custom icons persist across browsers as a normalized 128px PNG in the server entry's display-only `presentation.icon` metadata. They are not sent to the MCP transport. Capability Center > Plugins adds, replaces, and deletes one MCP server at a time through targeted mutations that preserve concurrent sibling changes; deletes use a bodyless URL-addressed request. 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. diff --git a/backend/app/gateway/AGENTS.md b/backend/app/gateway/AGENTS.md index 95df8dd1e4..2a574c5eae 100644 --- a/backend/app/gateway/AGENTS.md +++ b/backend/app/gateway/AGENTS.md @@ -211,10 +211,9 @@ archive/search behavior, read [Thread lifecycle invariants](../../docs/THREAD_LI It owns lineage and settled-checkpoint rules, legacy fallback boundaries, archive filtering before pagination, owner isolation, and activity-time preservation. -Capability installation IDs must be unique for MCP create/replace/state writes. -Single-server DELETE keeps schema validation but allows residual identity -collisions, so legacy duplicate groups can be repaired incrementally. Runtime -explicit selections still exclude ambiguous identities until repaired. +MCP create/replace/state writes require unique installation IDs. DELETE validates +schema but permits residual legacy ID collisions. Explicit selections exclude +ambiguous IDs until repaired. Capability skill discovery (`/api/capabilities/installations/skills`) reuses `routers/skills.py::_filter_visible_skills` after its off-thread storage read. @@ -234,3 +233,5 @@ model, while provider failures honor `authorization.fail_closed`. Skill listing authorization mirrors the models pattern: `routers/skills.py` routes resolve `(provider, principal)` through `authz.py::resolve_skill_authorization()` — a thin sibling of `resolve_model_authorization`, both delegating to the shared `_resolve_route_scoped_authorization` core — and the shared `_filter_visible_skills` helper filters the user-scoped catalog (public + caller's custom skills) through `filter_resources(principal, "skill", ...)` by name. Three user-facing surfaces apply it: `GET /api/skills` (the frontend skill list / slash-command autocomplete), `GET /api/skills/custom`, and `GET /api/skills/{name}` — the detail endpoint returns the standard 404 for an invisible skill so it cannot become an existence oracle the filtered list closed (`get_model`'s 403 is an execution decision via `authorize("model", "use")`, which has no skills equivalent in this layer). Anonymous callers are unfiltered (mirroring `list_models`), and provider resolution/decision errors follow `authorization.fail_closed` (fail-closed → empty listing / 404, fail-open → full listing). Skill management endpoints (`install`/`upload`/`reload`/custom-skill CRUD) stay `require_admin_user`-gated, and runtime activation authorization is a separate layer. The built-in RBAC provider maps this to the per-role `skills` policy key. Tests: `tests/test_skills_listing_authorization.py`. Batch workers pin `app.state.extensions`; never persist snapshots. + +Personal MCP uses owner-only files/calls; platform MCP stays shared. diff --git a/backend/app/gateway/app.py b/backend/app/gateway/app.py index f748c75bff..af5c27fb8f 100644 --- a/backend/app/gateway/app.py +++ b/backend/app/gateway/app.py @@ -40,6 +40,7 @@ from app.gateway.routers import ( mcp_tasks, memory, models, + personal_mcp, plugins, project_documents, project_thread_files, @@ -526,7 +527,9 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: logger.warning("Upload staging file cleanup skipped", exc_info=True) # Initialize LangGraph runtime components (StreamBridge, RunManager, checkpointer, store) - async with _runtime_with_mcp_pool_shutdown(app, startup_config): + from app.gateway.personal_mcp_access import personal_mcp_authority + + async with personal_mcp_authority(), _runtime_with_mcp_pool_shutdown(app, startup_config): logger.info("LangGraph runtime initialised") # Check admin bootstrap state and migrate orphan threads after admin exists. @@ -603,7 +606,6 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: OrdinaryMcpTaskDriver, ) from deerflow.mcp.tasks.runtime import ( - configured_task_toolset_count, set_mcp_task_config_snapshot, set_mcp_task_submitter, validate_mcp_task_runtime_configuration, @@ -622,11 +624,10 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]: ) if mcp_task_repo is not None: mcp_task_drivers = McpTaskDriverRegistry() - if configured_task_toolset_count(task_extensions_config): - mcp_task_drivers.register( - ORDINARY_MCP_TASK_DRIVER, - OrdinaryMcpTaskDriver(McpTaskToolCaller(task_extensions_config)), - ) + mcp_task_drivers.register( + ORDINARY_MCP_TASK_DRIVER, + OrdinaryMcpTaskDriver(McpTaskToolCaller(task_extensions_config)), + ) mcp_task_service = McpTaskService( repository=mcp_task_repo, drivers=mcp_task_drivers, @@ -1053,6 +1054,7 @@ This gateway provides runtime endpoints for agent runs plus custom endpoints for # MCP API is mounted at /api/mcp app.include_router(capabilities.router) app.include_router(mcp.router) + app.include_router(personal_mcp.router) # Durable MCP tasks are scoped to their owning thread. app.include_router(mcp_tasks.router) diff --git a/backend/app/gateway/capabilities.py b/backend/app/gateway/capabilities.py index 61425e42c1..8dfc600657 100644 --- a/backend/app/gateway/capabilities.py +++ b/backend/app/gateway/capabilities.py @@ -2,13 +2,13 @@ import asyncio from dataclasses import dataclass -from typing import Any, Protocol +from typing import Any, Literal, Protocol from uuid import uuid4 from fastapi import HTTPException, Request from pydantic import BaseModel, Field, ValidationError -from app.gateway.deps import is_admin_user +from app.gateway.deps import get_current_user_from_request, is_admin_user from app.gateway.routers import integrations, mcp, skills from deerflow.capabilities.business import connection_config from deerflow.capabilities.catalog import PluginManifest @@ -46,6 +46,7 @@ class AdapterContext: request: Request config: AppConfig user_id: str + scope: Literal["deployment", "user"] = "deployment" class CapabilityAdapter(Protocol): @@ -87,7 +88,16 @@ def validate_mcp_connection(configuration: dict[str, Any]) -> None: class MCPAdapter: async def list_installations(self, context: AdapterContext) -> list[CapabilityInstallation]: - servers = await asyncio.to_thread(mcp._load_raw_mcp_server_responses) + if context.scope == "user": + from deerflow.mcp.user_config import read_user_mcp_config + + try: + raw_config = await asyncio.to_thread(read_user_mcp_config, context.user_id) + except ValueError as exc: + mcp._raise_invalid_mcp_configuration(str(exc), cause=exc) + servers = mcp._mcp_server_responses_from_raw(raw_config) + else: + servers = await asyncio.to_thread(mcp._load_raw_mcp_server_responses) result = [] ambiguous = ambiguous_installation_ids({name: server.model_dump() for name, server in servers.items()}) for name, server in servers.items(): @@ -113,6 +123,7 @@ class MCPAdapter: health="ambiguous" if identity in ambiguous else "unknown", plugin_id=metadata.get("plugin_id") if isinstance(metadata.get("plugin_id"), str) else None, adapter="mcp", + scope=context.scope, name=name, reference=name, description=server.description or "", @@ -134,7 +145,12 @@ class MCPAdapter: body = mcp.McpConfigUpdateRequest(mcp_servers={name: mcp.McpServerConfigResponse.model_validate(definition)}) except ValidationError as error: raise HTTPException(422, "Invalid MCP configuration") from error - await mcp.create_mcp_servers(context.request, body) + if context.scope == "user": + from app.gateway.routers.personal_mcp import create_servers + + await create_servers(context.request, body) + else: + await mcp.create_mcp_servers(context.request, body) class BusinessAdapter(MCPAdapter): @@ -222,7 +238,12 @@ registry.register("lark", LarkAdapter()) registry.register("skills", SkillAdapter()) -async def list_installations(adapter: str, request: Request, config: AppConfig) -> InstallationList: - context = AdapterContext(request, config, get_effective_user_id()) +async def list_installations(adapter: str, request: Request, config: AppConfig, scope: Literal["deployment", "user", "all"] = "deployment") -> InstallationList: + user_id = get_effective_user_id() + if scope != "deployment": + user_id = str((await get_current_user_from_request(request)).id) + context = AdapterContext(request, config, user_id, "user" if scope == "user" else "deployment") items = await registry.get(adapter).list_installations(context) - return InstallationList(items=items, can_manage=await is_admin_user(request)) + if scope == "all" and adapter in {"mcp", "business"}: + items += await registry.get(adapter).list_installations(AdapterContext(request, config, user_id, "user")) + return InstallationList(items=items, can_manage=scope == "user" or await is_admin_user(request)) diff --git a/backend/app/gateway/personal_mcp_access.py b/backend/app/gateway/personal_mcp_access.py new file mode 100644 index 0000000000..d8399a7cd8 --- /dev/null +++ b/backend/app/gateway/personal_mcp_access.py @@ -0,0 +1,33 @@ +"""Bind personal MCP authority to fresh Gateway account records.""" + +import asyncio +from contextlib import asynccontextmanager + +from app.gateway.auth_disabled import AUTH_DISABLED_USER_ID, is_auth_disabled +from app.gateway.deps import get_local_provider +from deerflow.mcp.personal_access import set_personal_mcp_admin_checker + + +async def _is_current_admin(owner: str) -> bool: + if is_auth_disabled() and owner == AUTH_DISABLED_USER_ID: + return True + user = await get_local_provider().get_user(owner) + return user is not None and user.system_role == "admin" + + +@asynccontextmanager +async def personal_mcp_authority(): + loop = asyncio.get_running_loop() + + async def check(owner: str) -> bool: + if asyncio.get_running_loop() is loop: + return await _is_current_admin(owner) + # Synchronous tool discovery runs on assembly workers with their own + # loops. Keep the shared database pool on the Gateway's owning loop. + return await asyncio.wrap_future(asyncio.run_coroutine_threadsafe(_is_current_admin(owner), loop)) + + previous = set_personal_mcp_admin_checker(check) + try: + yield + finally: + set_personal_mcp_admin_checker(previous) diff --git a/backend/app/gateway/routers/capabilities.py b/backend/app/gateway/routers/capabilities.py index ea071ce47e..24cce2de67 100644 --- a/backend/app/gateway/routers/capabilities.py +++ b/backend/app/gateway/routers/capabilities.py @@ -1,13 +1,13 @@ """Capability discovery is public to authenticated users; mutations reuse existing policy.""" import asyncio -from typing import Any +from typing import Any, Literal from fastapi import APIRouter, Depends, HTTPException, Request from pydantic import BaseModel, ConfigDict, Field from app.gateway.capabilities import AdapterContext, InstallationList, list_installations, registry -from app.gateway.deps import get_config, require_admin_user +from app.gateway.deps import get_config, get_current_user_from_request, require_admin_user from deerflow.capabilities.catalog import PluginManifest, load_catalog from deerflow.config.app_config import AppConfig from deerflow.runtime.user_context import get_effective_user_id @@ -20,6 +20,7 @@ class InstallRequest(BaseModel): plugin_id: str name: str = Field(default="", max_length=128) configuration: dict[str, Any] = Field(default_factory=dict) + scope: Literal["deployment", "user"] = "deployment" @router.get("/catalog", response_model=list[PluginManifest]) @@ -28,17 +29,23 @@ async def catalog() -> list[PluginManifest]: @router.get("/installations/{adapter}", response_model=InstallationList) -async def installations(adapter: str, request: Request, config: AppConfig = Depends(get_config)) -> InstallationList: - return await list_installations(adapter, request, config) +async def installations(adapter: str, request: Request, config: AppConfig = Depends(get_config), scope: Literal["deployment", "user", "all"] = "deployment") -> InstallationList: + return await list_installations(adapter, request, config, scope) @router.post("/installations", response_model=InstallationList) async def install(body: InstallRequest, request: Request, config: AppConfig = Depends(get_config)) -> InstallationList: - await require_admin_user(request, detail="Admin privileges required to install capabilities.") + user_id = get_effective_user_id() + if body.scope == "deployment": + await require_admin_user(request, detail="Admin privileges required to install capabilities.") + else: + user_id = str((await get_current_user_from_request(request)).id) entries = await asyncio.to_thread(load_catalog) manifest = next((entry for entry in entries if entry.id == body.plugin_id), None) if manifest is None: raise HTTPException(404, "Plugin not found") - context = AdapterContext(request, config, get_effective_user_id()) + if body.scope == "user" and manifest.adapter not in {"mcp", "business"}: + await require_admin_user(request, detail="Admin privileges required to install this capability.") + context = AdapterContext(request, config, user_id, body.scope) await registry.get(manifest.adapter).install(context, manifest, body.name, body.configuration) - return await list_installations(manifest.adapter, request, config) + return await list_installations(manifest.adapter, request, config, body.scope) diff --git a/backend/app/gateway/routers/personal_mcp.py b/backend/app/gateway/routers/personal_mcp.py new file mode 100644 index 0000000000..81585ea2de --- /dev/null +++ b/backend/app/gateway/routers/personal_mcp.py @@ -0,0 +1,126 @@ +"""Owner-only MCP configuration. Deployment MCP routes remain admin-only.""" + +import asyncio +from typing import Literal +from uuid import uuid4 + +from fastapi import APIRouter, HTTPException, Request +from pydantic import ValidationError + +from app.gateway.deps import get_current_user_from_request, is_admin_user +from app.gateway.routers import mcp +from deerflow.capabilities.runtime import ambiguous_installation_ids +from deerflow.config.extensions_config import ExtensionsConfig, atomic_write_extensions_config, extensions_config_file_lock, extensions_config_write_lock +from deerflow.mcp.user_config import read_user_mcp_config, user_mcp_config_path + +router = APIRouter(prefix="/api/mcp/personal/config", tags=["mcp"]) + + +async def _owner(request: Request) -> str: + user = await get_current_user_from_request(request) + return str(user.id) + + +def _response(raw: dict) -> mcp.McpConfigResponse: + return mcp.McpConfigResponse(mcp_servers={name: mcp._mask_server_config(server) for name, server in mcp._mcp_server_responses_from_raw(raw).items()}) + + +def _read_personal_config(user_id: str) -> dict: + try: + return read_user_mcp_config(user_id) + except ValueError as exc: + mcp._raise_invalid_mcp_configuration(str(exc), cause=exc) + + +def _validate_personal_server(server: mcp.McpServerConfigResponse, *, admin: bool) -> None: + # A personal store must not grant ordinary users the deployment operator's + # ability to launch arbitrary host packages or query internal services. + if not admin: + from deerflow.capabilities.business import is_bundled_connection + + bundled = server.type == "stdio" and is_bundled_connection(server.command, server.args, server.env) and not server.cwd + if (server.type not in {"http", "sse"} and not bundled) or server.oauth is not None: + raise HTTPException(403, "Only an administrator may configure host commands or OAuth token endpoints") + from deerflow.community.url_safety import validate_public_http_url + + if not bundled and (not server.url or validate_public_http_url(server.url, action="connect to")): + raise HTTPException(400, "Personal MCP connections require a public HTTP(S) endpoint") + mcp._validate_mcp_update_request(mcp.McpConfigUpdateRequest(mcp_servers={"personal": server})) + + +def _mutate(user_id: str, operation: Literal["create", "update", "delete", "state"], body, *, admin: bool) -> dict: + path = user_mcp_config_path(user_id) + path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) + with extensions_config_write_lock, extensions_config_file_lock(path): + raw = _read_personal_config(user_id) + servers = mcp._raw_mcp_servers(raw) + if operation == "create": + for name, incoming in body.mcp_servers.items(): + if name in servers: + raise HTTPException(409, f"MCP server '{name}' already exists") + mcp._ensure_no_masked_secrets(incoming) + _validate_personal_server(incoming, admin=admin) + definition = incoming.model_dump() + metadata = definition.get("capability") + definition["capability"] = {**(metadata if isinstance(metadata, dict) else {}), "id": str(uuid4())} + definition["personal_public_network"] = not admin + servers[name] = definition + else: + name = body if operation == "delete" else body.server_name + if name not in servers: + raise HTTPException(404, "MCP server not found") + if operation == "delete": + del servers[name] + elif operation == "update": + merged = mcp._merge_preserving_secrets(body.server, mcp._mcp_server_response_from_raw(name, servers[name]), preserve_omitted_fields=False) + _validate_personal_server(merged, admin=admin) + servers[name] = {**merged.model_dump(), "personal_public_network": not admin} + else: + if body.enabled: + _validate_personal_server(mcp._mcp_server_response_from_raw(name, servers[name]), admin=admin) + servers[name]["personal_public_network"] = not admin + servers[name]["enabled"] = body.enabled + candidate = {"mcpServers": servers} + try: + # No environment expansion: these are caller-owned literal values. + ExtensionsConfig.model_validate(candidate) + except ValidationError: + raise HTTPException(400, "Invalid personal MCP configuration") from None + if ambiguous_installation_ids(servers): + raise HTTPException(400, "Duplicate MCP installation IDs") + atomic_write_extensions_config(path, candidate) + path.chmod(0o600) + return candidate + + +@router.get("", response_model=mcp.McpConfigResponse) +async def get_configuration(request: Request): + owner = await _owner(request) + return _response(await asyncio.to_thread(_read_personal_config, owner)) + + +async def _write(request: Request, operation: str, body): + owner = await _owner(request) + admin = await is_admin_user(request) + raw = await asyncio.to_thread(_mutate, owner, operation, body, admin=admin) + return _response(raw) + + +@router.post("/servers", response_model=mcp.McpConfigResponse) +async def create_servers(request: Request, body: mcp.McpConfigUpdateRequest): + return await _write(request, "create", body) + + +@router.put("/server", response_model=mcp.McpConfigResponse) +async def update_server(request: Request, body: mcp.McpServerConfigUpdateRequest): + return await _write(request, "update", body) + + +@router.patch("", response_model=mcp.McpConfigResponse) +async def update_state(request: Request, body: mcp.McpServerStateUpdateRequest): + return await _write(request, "state", body) + + +@router.delete("/servers/{server_name:path}", response_model=mcp.McpConfigResponse) +async def delete_server(request: Request, server_name: str): + return await _write(request, "delete", server_name) diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 456e27117d..3103304bdb 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -58,3 +58,6 @@ [backend/AGENTS.md](../../../../AGENTS.md)): Long-running MCP work uses a separate durable task runtime rather than keeping remote task IDs or status polling inside the Agent loop. Explicit `task_toolsets` bind raw submit/status/cancel names; only submit remains Agent-visible, and its wrapper persists the remote handle before returning a local ID. `McpTaskService` claims due rows with leases, resolves a protocol-specific `McpTaskDriver`, and writes normalized snapshots back to `mcp_tasks`; expired leases are the restart-recovery mechanism, and a result returned after expiry or after a cancel request must be discarded even when the owner token still matches. The first cancel request fences an in-flight poll lease, while repeats preserve an active cancellation lease so they cannot issue concurrent remote cancels; cancellation backoff starts when the remote attempt finishes, so a slow timeout cannot consume the retry delay. Cancellation, polling, and notification batches isolate per-task exceptions; an unexpected cancellation/poll failure leaves that record's lease to expire, while notification failures release only the affected lease for retry. Input-required and terminal event snapshots are delivered by idempotent Agent runs and marked delivered only after run success; the trusted notification instruction stays outside the input boundary while the serialized remote event is framed as untrusted data. A busy-thread conflict is normalized back to the service boundary so the queued snapshot coalesces to the latest task event. A missing dispatched run becomes a failed delivery attempt, while transient run-store hydration errors stay distinguishable and retry the same lookup. The database is the source of truth; `ThreadState` receives only a bounded current-thread projection, and display names are neutralized at that model-state boundary. The installed process-local submitter is the source of truth for management-tool exposure; hot `mcp_tasks` edits take effect only after restart, and active skills must explicitly declare the list/cancel business tools. **Task notification failure handling**: MCP notification failures use a consecutive counter separate from the idempotency-key `dispatch_attempt`, capped exponential backoff, latest-event rebuilding before a run launches, and a five-attempt budget before `dead_letter`. A permanently missing/mismatched target thread is dead-lettered immediately instead of being recreated or reclaimed. HTTP and Agent cancellation requests return after the durable cancel fence; the background loop alone owns the potentially slow remote call and retry schedule. The HTTP cancel endpoint rejects requests with 503 when the loop is not running (`mcp_tasks_available` false, e.g. `mcp_tasks.enabled=false` with SQL persistence), so a cancellation is never acknowledged without a worker to perform it. The bounded notification error/count/status join poll and cancellation diagnostics in the task detail API and expanded card. + +- Personal MCP configs live under the authenticated user root (`user_config.py`), use literal values, and never enter the deployment tool cache. `user_tools.py` guards owner and current revision before every call. Runtime server names bind owner/name/revision; durable task `driver_data.connection_scope` records personal or deployment ownership at submission, and background calls resolve a personal key only in the persisted task owner's file. Older rows without that field are deployment-owned. When a deployment name matches a personal runtime name, Agent assembly omits the personal tool rather than exposing two tools under one name; an already submitted personal task still uses its recorded personal scope. Keep catalog IDs stable across edits; never fall back by display name. +- Personal definitions without `personal_public_network: true` require current administrator authority at discovery and every tool/durable call. `personal_access.py` uses a host-installed async lookup, never a captured role or cached decision; absent/failed lookup denies privileged connections. The Gateway binds the lookup to fresh account records on its database-owning loop for its lifespan. Deployment connections are unaffected. This is admission control, not termination of operations already admitted before a role change. diff --git a/backend/packages/harness/deerflow/mcp/client.py b/backend/packages/harness/deerflow/mcp/client.py index 70e53a9ebf..f02bb32dc7 100644 --- a/backend/packages/harness/deerflow/mcp/client.py +++ b/backend/packages/harness/deerflow/mcp/client.py @@ -36,6 +36,10 @@ def build_server_params(server_name: str, config: McpServerConfig) -> dict[str, if not config.url: raise ValueError(f"MCP server '{server_name}' with {transport_type} transport requires 'url' field") params["url"] = config.url + if (config.model_extra or {}).get("personal_public_network") is True: + from deerflow.mcp.personal_network import personal_httpx_client_factory + + params["httpx_client_factory"] = personal_httpx_client_factory # Add headers if present if config.headers: # A statically configured value the transport would refuse gets the diff --git a/backend/packages/harness/deerflow/mcp/personal_access.py b/backend/packages/harness/deerflow/mcp/personal_access.py new file mode 100644 index 0000000000..277438314a --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/personal_access.py @@ -0,0 +1,46 @@ +"""Current host authority for personal connections saved with admin privileges.""" + +from collections.abc import Awaitable, Callable + +from langchain_core.tools import ToolException + +from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig + +PersonalMcpAdminChecker = Callable[[str], Awaitable[bool]] +_admin_checker: PersonalMcpAdminChecker | None = None + + +def set_personal_mcp_admin_checker(checker: PersonalMcpAdminChecker | None) -> PersonalMcpAdminChecker | None: + """Install the host's fresh authority lookup, returning the previous lookup.""" + global _admin_checker + previous, _admin_checker = _admin_checker, checker + return previous + + +def requires_admin(server: McpServerConfig) -> bool: + # Only definitions admitted under the ordinary-user policy carry True. + # Missing markers, including legacy personal files, require authority too. + return (server.model_extra or {}).get("personal_public_network") is not True + + +async def _is_current_admin(owner: str) -> bool: + checker = _admin_checker + if checker is None: + return False + try: + return await checker(owner) is True + except Exception: + # Authority lookup failures must not reuse a previous admin decision. + return False + + +async def require_personal_mcp_access(owner: str, server: McpServerConfig) -> None: + if requires_admin(server) and not await _is_current_admin(owner): + raise ToolException("Current administrator access is required for this personal MCP connection; save it again under your current permissions") + + +async def authorized_personal_config(owner: str, config: ExtensionsConfig) -> ExtensionsConfig: + """Omit privileged connections before discovery can launch or contact them.""" + if not any(server.enabled and requires_admin(server) for server in config.mcp_servers.values()) or await _is_current_admin(owner): + return config + return config.model_copy(update={"mcp_servers": {name: server for name, server in config.mcp_servers.items() if not requires_admin(server)}}) diff --git a/backend/packages/harness/deerflow/mcp/personal_network.py b/backend/packages/harness/deerflow/mcp/personal_network.py new file mode 100644 index 0000000000..5e11f46ffb --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/personal_network.py @@ -0,0 +1,62 @@ +"""Public-network policy for non-operator personal HTTP MCP connections.""" + +import asyncio +import ipaddress + +import httpx + +from deerflow.community.url_safety import resolve_host_addresses, validate_public_http_url + + +def _public_addresses(url: httpx.URL) -> list[str]: + addresses = [] + + def resolve(host): + candidates = resolve_host_addresses(host) + addresses.extend(str(address) for address in candidates) + return candidates + + if validate_public_http_url(str(url), action="connect to", resolver=resolve): + raise ValueError("Personal MCP connections require a public HTTP(S) endpoint") + # Literal IP URLs are validated without calling the resolver. + return list(dict.fromkeys(addresses)) if addresses else [str(ipaddress.ip_address(url.host))] + + +class _PersonalTransport(httpx.AsyncBaseTransport): + def __init__(self): + self._ssl_context = httpx.create_ssl_context(trust_env=False) + self._transports: dict[tuple[bytes, bytes, int | None], httpx.AsyncHTTPTransport] = {} + + async def handle_async_request(self, request: httpx.Request) -> httpx.Response: + addresses = await asyncio.to_thread(_public_addresses, request.url) + origin = (request.url.raw_scheme, request.url.raw_host, request.url.port) + # Pin TCP to the vetted IP, but isolate pools by original origin so two + # names sharing an IP cannot reuse a connection authenticated for one. + if origin not in self._transports: + self._transports[origin] = httpx.AsyncHTTPTransport(verify=self._ssl_context, trust_env=False) + transport = self._transports[origin] + for index, address in enumerate(addresses): + pinned = httpx.Request( + request.method, + request.url.copy_with(host=address), + headers=request.headers, + stream=request.stream, + extensions={**request.extensions, "sni_hostname": request.url.raw_host.decode("ascii")}, + ) + try: + return await transport.handle_async_request(pinned) + except (httpx.ConnectError, httpx.ConnectTimeout): + if index == len(addresses) - 1: + raise + raise AssertionError("Public address validation returned no addresses") + + async def aclose(self): + for transport in self._transports.values(): + await transport.aclose() + + +def personal_httpx_client_factory(headers=None, timeout=None, auth=None): + kwargs = {"headers": headers, "auth": auth, "follow_redirects": False, "trust_env": False, "transport": _PersonalTransport()} + if timeout is not None: + kwargs["timeout"] = timeout + return httpx.AsyncClient(**kwargs) diff --git a/backend/packages/harness/deerflow/mcp/task_tool_caller.py b/backend/packages/harness/deerflow/mcp/task_tool_caller.py index 6290736377..ba73a3dbd2 100644 --- a/backend/packages/harness/deerflow/mcp/task_tool_caller.py +++ b/backend/packages/harness/deerflow/mcp/task_tool_caller.py @@ -4,10 +4,12 @@ from __future__ import annotations import asyncio import logging +import threading +from collections import OrderedDict from collections.abc import Mapping from dataclasses import dataclass from datetime import timedelta -from typing import Any +from typing import Any, Literal from deerflow.config.extensions_config import ExtensionsConfig from deerflow.config.paths import get_paths @@ -17,11 +19,15 @@ from deerflow.mcp.context_headers import build_context_headers_interceptor from deerflow.mcp.headers import apply_header_overrides from deerflow.mcp.interceptors import build_mcp_tool_interceptors from deerflow.mcp.oauth import OAuthTokenManager, build_oauth_tool_interceptor +from deerflow.mcp.personal_access import require_personal_mcp_access from deerflow.mcp.session_pool import MCPSessionPool, call_pooled_session_tool, get_session_pool +from deerflow.mcp.user_config import PersonalMcpConfigSnapshot, load_user_mcp_config_if_changed from deerflow.mcp_scope import mcp_session_scope_key from deerflow.runtime.user_context import reset_current_user, set_current_user logger = logging.getLogger(__name__) +# Cached callers retain personal credentials and OAuth state in memory. +_MAX_PERSONAL_CALLERS = 128 @dataclass @@ -64,6 +70,8 @@ class McpTaskToolCaller: oauth_token_manager: OAuthTokenManager | None = None, ) -> None: self._extensions_config = extensions_config + self._personal_callers: OrderedDict[str, tuple[PersonalMcpConfigSnapshot, McpTaskToolCaller | None]] = OrderedDict() + self._personal_callers_lock = threading.Lock() self._oauth_token_manager = oauth_token_manager or OAuthTokenManager.from_extensions_config(extensions_config) context_headers_interceptor = build_context_headers_interceptor(extensions_config) # Built once so the two chains keep an identical interceptor order and a @@ -100,14 +108,68 @@ class McpTaskToolCaller: thread_id: str, thread_incarnation: str | None = None, request_scoped_headers: bool = False, + connection_scope: Literal["deployment", "personal"] = "deployment", ) -> Any: """Call a raw MCP tool. ``request_scoped_headers`` opts this call into the ``headers_from_context`` interceptor. Only the durable *submit* may set it: submit is awaited inside the Agent run that carries the secrets, while status and cancel - run after that run ended. + run after that run ended. ``connection_scope`` is captured in the task + binding, so an equal deployment server name cannot change its owner. """ + if connection_scope == "personal": + caller = await asyncio.to_thread(self._personal_caller_for, user_id, server_name) + await require_personal_mcp_access(user_id, caller._extensions_config.mcp_servers[server_name]) + return await caller._call_configured_tool( + server_name=server_name, + tool_name=tool_name, + arguments=arguments, + user_id=user_id, + thread_id=thread_id, + thread_incarnation=thread_incarnation, + request_scoped_headers=request_scoped_headers, + ) + if connection_scope != "deployment": + raise ValueError("Invalid MCP task connection scope") + return await self._call_configured_tool( + server_name=server_name, + tool_name=tool_name, + arguments=arguments, + user_id=user_id, + thread_id=thread_id, + thread_incarnation=thread_incarnation, + request_scoped_headers=request_scoped_headers, + ) + + def _personal_caller_for(self, user_id: str, server_name: str) -> McpTaskToolCaller: + with self._personal_callers_lock: + previous = self._personal_callers.get(user_id) + snapshot = load_user_mcp_config_if_changed(user_id, previous[0] if previous else None) + caller = previous[1] if previous and previous[0] is snapshot else None + self._personal_callers[user_id] = (snapshot, caller) + self._personal_callers.move_to_end(user_id) + if len(self._personal_callers) > _MAX_PERSONAL_CALLERS: + self._personal_callers.popitem(last=False) + server = snapshot.config.mcp_servers.get(server_name) + if server is None or not server.enabled: + raise LookupError("Personal MCP task connection is missing, disabled or changed") + if caller is None: + caller = McpTaskToolCaller(snapshot.config) + self._personal_callers[user_id] = (snapshot, caller) + return caller + + async def _call_configured_tool( + self, + *, + server_name: str, + tool_name: str, + arguments: dict[str, Any], + user_id: str, + thread_id: str, + thread_incarnation: str | None, + request_scoped_headers: bool, + ) -> Any: is_background_call = not request_scoped_headers interceptors = self._interceptors if is_background_call else self._submit_interceptors server_config = self._extensions_config.get_enabled_mcp_servers().get(server_name) diff --git a/backend/packages/harness/deerflow/mcp/tasks/ordinary.py b/backend/packages/harness/deerflow/mcp/tasks/ordinary.py index c80e172f7f..c743735bb2 100644 --- a/backend/packages/harness/deerflow/mcp/tasks/ordinary.py +++ b/backend/packages/harness/deerflow/mcp/tasks/ordinary.py @@ -34,6 +34,7 @@ class McpTaskToolCaller(Protocol): thread_id: str, thread_incarnation: str | None = None, request_scoped_headers: bool = False, + connection_scope: Literal["deployment", "personal"] = "deployment", ) -> Any: ... @@ -86,6 +87,14 @@ def _tool_name(data: dict[str, Any], role: str) -> str: return value +def _connection_scope(data: dict[str, Any]) -> Literal["deployment", "personal"]: + # Older task rows predate personal connections and are deployment-owned. + scope = data.get("connection_scope", "deployment") + if scope not in ("deployment", "personal"): + raise McpTaskProtocolError("Task driver_data has an invalid connection scope") + return scope + + def _first_error_text(call_result: Any) -> str | None: content = getattr(call_result, "content", None) if not isinstance(content, (list, tuple)): @@ -165,6 +174,7 @@ class OrdinaryMcpTaskDriver: # durable-task call that can carry the run's request-scoped # credentials; status and cancel run after that run ended. request_scoped_headers=True, + connection_scope=_connection_scope(request.driver_data), ) payload = _parse( _SubmitPayload, @@ -187,6 +197,7 @@ class OrdinaryMcpTaskDriver: user_id=task.user_id, thread_id=task.thread_id, thread_incarnation=task.thread_incarnation, + connection_scope=_connection_scope(task.driver_data), ) payload = _parse( _StatusPayload, @@ -206,6 +217,7 @@ class OrdinaryMcpTaskDriver: user_id=task.user_id, thread_id=task.thread_id, thread_incarnation=task.thread_incarnation, + connection_scope=_connection_scope(task.driver_data), ) payload = _parse( _CancelPayload, diff --git a/backend/packages/harness/deerflow/mcp/tools.py b/backend/packages/harness/deerflow/mcp/tools.py index 298f55a99e..524338a86e 100644 --- a/backend/packages/harness/deerflow/mcp/tools.py +++ b/backend/packages/harness/deerflow/mcp/tools.py @@ -726,6 +726,7 @@ def _make_background_submit_tool( submit_tool: str, status_tool: str, cancel_tool: str, + connection_scope: str, ) -> BaseTool: background_contract = f"Submitted as durable background task {task_name!r}; returns a DeerFlow task ID immediately and status polling is handled automatically." @@ -755,6 +756,7 @@ def _make_background_submit_tool( "submit_tool": submit_tool, "status_tool": status_tool, "cancel_tool": cancel_tool, + "connection_scope": connection_scope, }, ), ) @@ -780,6 +782,7 @@ def _configure_task_tools_for_server( server_name: str, server_config: McpServerConfig, tool_name_prefix: bool, + connection_scope: str = "deployment", ) -> list[BaseTool]: """Hide driver-only tools and replace submit with a durable wrapper.""" if not server_config.task_toolsets: @@ -829,12 +832,13 @@ def _configure_task_tools_for_server( submit_tool=toolset.submit_tool, status_tool=toolset.status_tool, cancel_tool=toolset.cancel_tool, + connection_scope=connection_scope, ) ) return configured -async def get_mcp_tools(extensions_config: ExtensionsConfig | None = None) -> list[BaseTool]: +async def get_mcp_tools(extensions_config: ExtensionsConfig | None = None, *, personal_user_id: str | None = None) -> list[BaseTool]: """Get all tools from enabled MCP servers. Tools using stdio transport are wrapped with persistent-session logic so @@ -864,7 +868,19 @@ async def get_mcp_tools(extensions_config: ExtensionsConfig | None = None) -> li # reflected when initializing MCP tools. Callers that need to prove which # revision produced these tools pass the instance they snapshotted instead. extensions_config = ExtensionsConfig.from_file() - validate_mcp_task_config_snapshot(extensions_config) + if personal_user_id is None: + validate_mcp_task_config_snapshot(extensions_config) + else: + from deerflow.mcp.personal_access import authorized_personal_config + from deerflow.mcp.tasks.runtime import is_mcp_task_runtime_available + from deerflow.mcp.user_config import load_user_mcp_config + + current = await asyncio.to_thread(load_user_mcp_config, personal_user_id) + if current != extensions_config: + raise McpTaskConfigurationError("Personal MCP configuration changed during discovery; retry the run") + extensions_config = current = await authorized_personal_config(personal_user_id, current) + if any(server.task_toolsets for server in current.mcp_servers.values()) and not is_mcp_task_runtime_available(): + raise McpTaskConfigurationError("Personal MCP task toolsets require the platform's durable task runtime") servers_config = build_servers_config(extensions_config) if not servers_config: @@ -1022,6 +1038,7 @@ async def get_mcp_tools(extensions_config: ExtensionsConfig | None = None) -> li server_name=source_name, server_config=server_cfg, tool_name_prefix=tool_name_prefix, + connection_scope="personal" if personal_user_id is not None else "deployment", ) wrapped_tools.extend(current_server_tools) diff --git a/backend/packages/harness/deerflow/mcp/user_config.py b/backend/packages/harness/deerflow/mcp/user_config.py new file mode 100644 index 0000000000..6c8ad2d990 --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/user_config.py @@ -0,0 +1,77 @@ +"""Persistent personal MCP connections, separate from deployment configuration.""" + +from __future__ import annotations + +import hashlib +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + +from deerflow.config.extensions_config import ExtensionsConfig, read_raw_extensions_config +from deerflow.config.paths import get_paths + +PERSONAL_SERVER_PREFIX = "personal_" +_FileSignature = tuple[int, int, int, int, int] | None + + +@dataclass(frozen=True) +class PersonalMcpConfigSnapshot: + path: Path + signature: _FileSignature + config: ExtensionsConfig + + +def user_mcp_config_path(user_id: str) -> Path: + """Use the same validated user root as skills and account integrations.""" + return get_paths().user_dir(user_id) / "integrations" / "mcp.json" + + +def read_user_mcp_config(user_id: str) -> dict[str, Any]: + path = user_mcp_config_path(user_id) + if not path.exists(): + return {"mcpServers": {}} + return read_raw_extensions_config(path) + + +def personal_server_name(user_id: str, name: str, configuration: dict[str, Any]) -> str: + """Bind runtime sessions and durable tasks to one owner and config revision. + + Display names remain local to each user's file. Never let a same-named + personal connection replace a deployment connection or another user's tools. + """ + identity = json.dumps([user_id, name, configuration], sort_keys=True) + return PERSONAL_SERVER_PREFIX + hashlib.sha256(identity.encode()).hexdigest()[:32] + + +def load_user_mcp_config(user_id: str) -> ExtensionsConfig: + """Read literal user values; personal files cannot resolve host $ENV secrets.""" + raw = read_user_mcp_config(user_id) + servers = {} + for name, definition in raw.get("mcpServers", {}).items(): + runtime_name = personal_server_name(user_id, name, definition) + servers[runtime_name] = {**definition, "tool_name_prefix": True} + return ExtensionsConfig.model_validate({"mcpServers": servers}) + + +def _file_signature(path: Path) -> _FileSignature: + try: + stat = path.stat() + except FileNotFoundError: + return None + # Include file identity and ctime: atomic replacement can preserve size and + # mtime, while an in-place edit can preserve the inode. + return stat.st_dev, stat.st_ino, stat.st_size, stat.st_mtime_ns, stat.st_ctime_ns + + +def load_user_mcp_config_if_changed(user_id: str, cached: PersonalMcpConfigSnapshot | None = None) -> PersonalMcpConfigSnapshot: + """Reuse validated config while the owner's on-disk file is unchanged.""" + path = user_mcp_config_path(user_id) + for _ in range(3): + signature = _file_signature(path) + if cached is not None and cached.path == path and cached.signature == signature: + return cached + config = load_user_mcp_config(user_id) + if _file_signature(path) == signature: + return PersonalMcpConfigSnapshot(path, signature, config) + raise RuntimeError("Personal MCP configuration changed during loading; retry the call") diff --git a/backend/packages/harness/deerflow/mcp/user_tools.py b/backend/packages/harness/deerflow/mcp/user_tools.py new file mode 100644 index 0000000000..b76ced9689 --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/user_tools.py @@ -0,0 +1,61 @@ +"""Load only the caller's personal tools; never publish them in the global cache.""" + +from __future__ import annotations + +import asyncio +import functools +import threading + +from langchain_core.tools import ToolException + +from deerflow.mcp.personal_access import require_personal_mcp_access +from deerflow.mcp.user_config import PersonalMcpConfigSnapshot, load_user_mcp_config, load_user_mcp_config_if_changed +from deerflow.runtime.user_context import resolve_runtime_user_id +from deerflow.tools.mcp_metadata import get_mcp_source +from deerflow.tools.sync import make_sync_tool_wrapper + + +def _guard(tool, owner: str, server_name: str): + original = tool.coroutine + snapshot: PersonalMcpConfigSnapshot | None = None + snapshot_lock = threading.Lock() + + def current_connection(): + nonlocal snapshot + with snapshot_lock: + snapshot = load_user_mcp_config_if_changed(owner, snapshot) + server = snapshot.config.mcp_servers.get(server_name) + return server if server is not None and server.enabled else None + + @functools.wraps(original) + async def invoke(*args, **kwargs): + if resolve_runtime_user_id(kwargs.get("runtime")) != owner: + raise ToolException("This MCP connection belongs to another user") + server = await asyncio.to_thread(current_connection) + if server is None: + raise ToolException("This personal MCP connection was changed, disabled or removed; start a new run") + await require_personal_mcp_access(owner, server) + return await original(*args, **kwargs) + + return tool.model_copy(update={"coroutine": invoke, "func": make_sync_tool_wrapper(invoke, tool.name)}) + + +async def _load(owner: str, config): + from deerflow.mcp.tools import get_mcp_tools + + tools = await get_mcp_tools(config, personal_user_id=owner) + return [_guard(tool, owner, source["server_name"]) for tool in tools if (source := get_mcp_source(tool))] + + +def get_user_mcp_tools(): + """Discover from one literal snapshot, with an ownership check on every call. + + Deliberately keep personal discovery out of the deployment's singleton + cache. Stdio session reuse remains owner/thread/revision-scoped. + """ + owner = resolve_runtime_user_id(None) + config = load_user_mcp_config(owner) + if not config.get_enabled_mcp_servers(): + return [], config + tools = make_sync_tool_wrapper(_load, "personal MCP discovery")(owner, config) + return tools, config diff --git a/backend/packages/harness/deerflow/tools/tools.py b/backend/packages/harness/deerflow/tools/tools.py index d894a05d56..1bd5a1d57c 100644 --- a/backend/packages/harness/deerflow/tools/tools.py +++ b/backend/packages/harness/deerflow/tools/tools.py @@ -26,7 +26,7 @@ from deerflow.tools.builtins import ( task_tool, view_image_tool, ) -from deerflow.tools.mcp_metadata import tag_mcp_tool +from deerflow.tools.mcp_metadata import get_mcp_source, tag_mcp_tool from deerflow.tools.sync import make_sync_tool_wrapper logger = logging.getLogger(__name__) @@ -262,6 +262,25 @@ def get_available_tools( # MCP pays no config-hashing or discovery cost, while one that # did still checks the existing cache for staleness. refresh_mcp_cache_if_active() + from deerflow.mcp.user_tools import get_user_mcp_tools + + try: + personal_tools, personal_config = get_user_mcp_tools() + except Exception as exc: + # Invalid personal credentials must neither be logged nor + # bypass explicit selection of the platform tools. + logger.warning("Could not load personal MCP tools (%s)", type(exc).__name__) + personal_tools, personal_config = [], ExtensionsConfig() + colliding_names = extensions_config.mcp_servers.keys() & personal_config.mcp_servers.keys() + if colliding_names: + # A deployment name can match a generated personal runtime + # name. Keep only the deployment tool and its routing metadata. + logger.warning("Skipping personal MCP tools with deployment name collisions: %s", sorted(colliding_names)) + personal_tools = [tool for tool in personal_tools if (source := get_mcp_source(tool)) and source["server_name"] not in colliding_names] + available_servers = {name: server for name, server in personal_config.mcp_servers.items() if name not in colliding_names} + personal_config = personal_config.model_copy(update={"mcp_servers": available_servers}) + mcp_tools = [*mcp_tools, *personal_tools] + extensions_config = extensions_config.model_copy(update={"mcp_servers": {**extensions_config.mcp_servers, **personal_config.mcp_servers}}) if mcp_plugins is not None: from deerflow.capabilities.runtime import filter_mcp_plugins diff --git a/backend/tests/test_gateway_lifespan_shutdown.py b/backend/tests/test_gateway_lifespan_shutdown.py index 2316bcf68e..44edb3a9b7 100644 --- a/backend/tests/test_gateway_lifespan_shutdown.py +++ b/backend/tests/test_gateway_lifespan_shutdown.py @@ -172,6 +172,30 @@ def test_lifespan_sweeps_upload_staging_files_on_startup(): stop_channel_service.assert_awaited_once() +def test_personal_mcp_authority_spans_runtime_startup_and_shutdown(): + from deerflow.mcp import personal_access + + events = [] + previous = personal_access._admin_checker + + @asynccontextmanager + async def checked_runtime(_app, _config): + checker = personal_access._admin_checker + assert checker is not None and checker is not previous + events.append("start") + try: + yield + finally: + assert personal_access._admin_checker is checker + events.append("stop") + + with patch(f"{__name__}._noop_langgraph_runtime", checked_runtime): + asyncio.run(_run_lifespan_with_upload_staging_cleanup()) + + assert events == ["start", "stop"] + assert personal_access._admin_checker is previous + + async def _run_lifespan_with_mcp_task_config_snapshot() -> None: from app.gateway.app import lifespan from deerflow.config.extensions_config import ExtensionsConfig diff --git a/backend/tests/test_mcp_task_ordinary_driver.py b/backend/tests/test_mcp_task_ordinary_driver.py index ddc3025e41..5720e1532b 100644 --- a/backend/tests/test_mcp_task_ordinary_driver.py +++ b/backend/tests/test_mcp_task_ordinary_driver.py @@ -1,3 +1,4 @@ +from dataclasses import replace from types import SimpleNamespace import pytest @@ -80,10 +81,41 @@ async def test_submit_uses_structured_content_and_keeps_remote_id_out_of_driver_ # Submit is the one durable-task call awaited inside the Agent run, # so it is the only one that may carry request-scoped credentials. "request_scoped_headers": True, + "connection_scope": "deployment", } ] +@pytest.mark.asyncio +async def test_personal_connection_scope_survives_submit_status_and_cancel() -> None: + caller = FakeCaller( + _result({"task_id": "remote-1", "status": "running"}), + _result({"task_id": "remote-1", "status": "completed"}), + _result({"task_id": "remote-1", "status": "cancelled"}), + ) + driver = OrdinaryMcpTaskDriver(caller) + request = _request() + request = replace(request, driver_data={**request.driver_data, "connection_scope": "personal"}) + submission = await driver.submit(request) + reference = replace(_reference(), driver_data=submission.driver_data) + + await driver.get_status(reference) + await driver.cancel(reference) + + assert submission.driver_data["connection_scope"] == "personal" + assert [call["connection_scope"] for call in caller.calls] == ["personal"] * 3 + + +@pytest.mark.asyncio +async def test_unknown_task_connection_scope_fails_closed() -> None: + driver = OrdinaryMcpTaskDriver(FakeCaller()) + reference = _reference() + reference = replace(reference, driver_data={**reference.driver_data, "connection_scope": "unknown"}) + + with pytest.raises(McpTaskProtocolError, match="connection scope"): + await driver.get_status(reference) + + @pytest.mark.asyncio @pytest.mark.parametrize( "structured_content,match", diff --git a/backend/tests/test_mcp_task_tool_wrapping.py b/backend/tests/test_mcp_task_tool_wrapping.py index 6cf262d6d5..d7ab4e8775 100644 --- a/backend/tests/test_mcp_task_tool_wrapping.py +++ b/backend/tests/test_mcp_task_tool_wrapping.py @@ -124,7 +124,8 @@ def test_configured_task_toolsets_fail_when_a_raw_tool_is_missing() -> None: @pytest.mark.asyncio -async def test_submit_wrapper_persists_before_returning_only_the_local_handle() -> None: +@pytest.mark.parametrize("connection_scope", ["deployment", "personal"]) +async def test_submit_wrapper_persists_before_returning_only_the_local_handle(connection_scope: str) -> None: submitter = FakeSubmitter() set_mcp_task_submitter(submitter) try: @@ -137,6 +138,7 @@ async def test_submit_wrapper_persists_before_returning_only_the_local_handle() server_name="reports", server_config=_server_config(), tool_name_prefix=False, + connection_scope=connection_scope, ) submit_tool = configured[0] runtime = SimpleNamespace( @@ -165,6 +167,7 @@ async def test_submit_wrapper_persists_before_returning_only_the_local_handle() "submit_tool": "submit_report", "status_tool": "get_report_status", "cancel_tool": "cancel_report", + "connection_scope": connection_scope, } finally: set_mcp_task_submitter(None) diff --git a/backend/tests/test_personal_mcp.py b/backend/tests/test_personal_mcp.py new file mode 100644 index 0000000000..9e8c8aa367 --- /dev/null +++ b/backend/tests/test_personal_mcp.py @@ -0,0 +1,545 @@ +"""Personal MCP persistence, HTTP ownership and real MCP credential routing.""" + +import asyncio +import ipaddress +import json +from contextlib import asynccontextmanager +from types import SimpleNamespace +from unittest.mock import AsyncMock, MagicMock, patch + +import httpx +import pytest +from fastapi import FastAPI +from fastapi.testclient import TestClient +from langchain_core.tools import ToolException + +from app.gateway.routers import personal_mcp +from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp.user_config import load_user_mcp_config, read_user_mcp_config, user_mcp_config_path +from deerflow.runtime.user_context import reset_current_user, set_current_user + + +@pytest.fixture +def personal_client(tmp_path, monkeypatch): + from deerflow.config.paths import Paths + + monkeypatch.setattr("deerflow.mcp.user_config.get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr("deerflow.mcp.personal_access._admin_checker", AsyncMock(return_value=True)) + monkeypatch.setattr("app.gateway.personal_mcp_access._is_current_admin", AsyncMock(return_value=True)) + app = FastAPI() + + @app.middleware("http") + async def identity(request, call_next): + if name := request.headers.get("test-user"): + request.state.user = SimpleNamespace(id=name, system_role=request.headers.get("test-role", "admin")) + request.state.auth_source = "session" + return await call_next(request) + + app.include_router(personal_mcp.router) + with TestClient(app) as client: + yield client + + +def create(client, user, *, name="github", token=None, role="admin"): + return client.post( + "/api/mcp/personal/config/servers", + headers={"test-user": user, "test-role": role}, + json={"mcp_servers": {name: {"type": "http", "url": "https://example.com/mcp", "headers": {"Authorization": token or f"Bearer {user}"}}}}, + ) + + +def test_persistent_same_name_connections_are_owner_only(personal_client): + client = personal_client + assert client.get("/api/mcp/personal/config").status_code == 401 + for user in ("alice", "bob"): + result = create(client, user) + assert result.status_code == 200, result.text + assert result.json()["mcp_servers"]["github"]["headers"]["Authorization"] == "***" + assert read_user_mcp_config("alice")["mcpServers"]["github"]["headers"]["Authorization"] == "Bearer alice" + assert read_user_mcp_config("bob")["mcpServers"]["github"]["headers"]["Authorization"] == "Bearer bob" + assert user_mcp_config_path("alice").stat().st_mode & 0o777 == 0o600 + assert create(client, "alice").status_code == 409 + assert create(client, "alice", name="only-alice").status_code == 200 + assert client.delete("/api/mcp/personal/config/servers/only-alice", headers={"test-user": "bob"}).status_code == 404 + result = client.get("/api/mcp/personal/config", headers={"test-user": "bob"}) + assert "only-alice" not in result.text + before = user_mcp_config_path("alice").read_bytes() + assert client.patch("/api/mcp/personal/config", headers={"test-user": "bob"}, json={"server_name": "github", "enabled": False}).status_code == 200 + assert user_mcp_config_path("alice").read_bytes() == before + # A fresh read from disk, not a request-local cache, retains each owner. + assert len(load_user_mcp_config("alice").get_enabled_mcp_servers()) == 2 + assert not load_user_mcp_config("bob").get_enabled_mcp_servers() + + +@pytest.mark.parametrize("contents", ["{", "[]"]) +def test_corrupt_personal_config_returns_client_errors_without_overwriting_it(personal_client, contents): + path = user_mcp_config_path("alice") + path.parent.mkdir(parents=True, exist_ok=True) + path.write_text(contents) + headers = {"test-user": "alice"} + requests = [ + ("GET", "/api/mcp/personal/config", None), + ("POST", "/api/mcp/personal/config/servers", {"mcp_servers": {"new": {"type": "http", "url": "https://example.com/mcp"}}}), + ("PUT", "/api/mcp/personal/config/server", {"server_name": "new", "server": {"type": "http", "url": "https://example.com/mcp"}}), + ("PATCH", "/api/mcp/personal/config", {"server_name": "new", "enabled": False}), + ("DELETE", "/api/mcp/personal/config/servers/new", None), + ] + for method, url, body in requests: + response = personal_client.request(method, url, headers=headers, json=body) + assert response.status_code == 400, (method, response.text) + assert "Extensions configuration" in response.json()["detail"] + assert path.read_text() == contents + assert personal_client.get("/api/mcp/personal/config", headers={"test-user": "bob"}).status_code == 200 + + +def test_masked_edit_and_delete_do_not_touch_platform_or_peer(personal_client, tmp_path, monkeypatch): + client = personal_client + platform = tmp_path / "platform.json" + platform.write_text(json.dumps({"mcpServers": {"github": {"type": "http", "url": "https://example.com/platform"}}})) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(platform)) + before = platform.read_bytes() + create(client, "alice") + create(client, "bob") + result = client.put("/api/mcp/personal/config/server", headers={"test-user": "alice"}, json={"server_name": "github", "server": {"type": "http", "url": "https://example.com/updated", "headers": {"Authorization": "***"}}}) + assert result.status_code == 200, result.text + assert read_user_mcp_config("alice")["mcpServers"]["github"]["headers"]["Authorization"] == "Bearer alice" + assert client.delete("/api/mcp/personal/config/servers/github", headers={"test-user": "alice"}).status_code == 200 + assert not load_user_mcp_config("alice").mcp_servers + assert read_user_mcp_config("bob")["mcpServers"]["github"]["headers"]["Authorization"] == "Bearer bob" + assert platform.read_bytes() == before + + +@pytest.mark.parametrize("contents", ["{", "[]"]) +@pytest.mark.parametrize("adapter", ["mcp", "business"]) +def test_corrupt_personal_config_in_capability_listings(personal_client, tmp_path, monkeypatch, contents, adapter): + from app.gateway.deps import get_config + from app.gateway.routers import capabilities + + app = personal_client.app + app.include_router(capabilities.router) + app.dependency_overrides[get_config] = lambda: SimpleNamespace() + deployment = tmp_path / "deployment.json" + deployment.write_text(json.dumps({"mcpServers": {"shared": {"type": "http", "url": "https://example.com/mcp"}}})) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(deployment)) + path = user_mcp_config_path("alice") + path.parent.mkdir(parents=True) + path.write_text(contents) + url = f"/api/capabilities/installations/{adapter}" + for scope in ("user", "all"): + response = personal_client.get(url, params={"scope": scope}, headers={"test-user": "alice"}) + assert response.status_code == 400 + assert "Extensions configuration" in response.json()["detail"] + assert personal_client.get(url, params={"scope": "deployment"}, headers={"test-user": "alice"}).status_code == 200 + assert personal_client.get(url, params={"scope": "all"}, headers={"test-user": "bob"}).status_code == 200 + assert path.read_text() == contents + + +def test_personal_values_do_not_resolve_platform_environment(personal_client, monkeypatch): + monkeypatch.setenv("PLATFORM_ONLY_KEY", "platform-secret") + assert create(personal_client, "alice", token="$PLATFORM_ONLY_KEY").status_code == 200 + config = load_user_mcp_config("alice") + assert next(iter(config.mcp_servers.values())).headers["Authorization"] == "$PLATFORM_ONLY_KEY" + + +def test_catalog_listing_and_installation_respect_owner(personal_client, monkeypatch, tmp_path): + from app.gateway.deps import get_config + from app.gateway.routers import capabilities + + app = personal_client.app + app.include_router(capabilities.router) + app.dependency_overrides[get_config] = lambda: SimpleNamespace() + platform = tmp_path / "deployment.json" + platform.write_text(json.dumps({"mcpServers": {"shared": {"type": "http", "url": "https://example.com/platform"}}})) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(platform)) + monkeypatch.setattr("deerflow.community.url_safety.validate_public_http_url", lambda *a, **k: None) + response = personal_client.post( + "/api/capabilities/installations", + headers={"test-user": "alice", "test-role": "user"}, + json={"plugin_id": "github", "name": "my-github", "scope": "user", "configuration": {"type": "http", "url": "https://example.com/mcp", "headers": {"Authorization": "Bearer alice"}}}, + ) + assert response.status_code == 200, response.text + assert response.json()["can_manage"] is True + assert response.json()["items"][0]["scope"] == "user" + for user, expected in (("alice", {"shared", "my-github"}), ("bob", {"shared"})): + response = personal_client.get("/api/capabilities/installations/mcp?scope=all", headers={"test-user": user}) + assert response.status_code == 200, response.text + assert {item["name"] for item in response.json()["items"]} == expected + assert "Bearer alice" not in response.text + + +def test_tool_assembly_combines_platform_and_only_current_owner(personal_client, monkeypatch, tmp_path): + from langchain_core.tools import StructuredTool + + from deerflow.tools.mcp_metadata import tag_mcp_tool + from deerflow.tools.tools import get_available_tools + + path = tmp_path / "deployment.json" + path.write_text(json.dumps({"mcpServers": {"shared": {"type": "http", "url": "https://example.com/mcp"}}})) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(path)) + shared = tag_mcp_tool(StructuredTool.from_function(lambda: "platform", name="shared_test", description="Shared tool"), server_name="shared") + monkeypatch.setattr("deerflow.mcp.cache.get_cached_mcp_tools", lambda: [shared]) + + async def discover(config, **kwargs): + name = next(iter(config.mcp_servers)) + + async def echo(): + return config.mcp_servers[name].headers["Authorization"] + + return [tag_mcp_tool(StructuredTool.from_function(coroutine=echo, name=name + "_test", description="Personal tool"), server_name=name)] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", discover) + config = SimpleNamespace(tools=[], models=[], get_model_config=lambda _: None) + for user in ("alice", "bob"): + create(personal_client, user) + identity = set_current_user(SimpleNamespace(id=user)) + try: + tools = get_available_tools(app_config=config) + assert shared in tools + personal = [tool for tool in tools if tool.name.startswith("personal_")] + assert len(personal) == 1 + assert personal[0].invoke({}) == f"Bearer {user}" + # Selecting the peer's installation cannot add that peer's tools. + peer = "bob" if user == "alice" else "alice" + peer_id = next(iter(read_user_mcp_config(peer)["mcpServers"].values()), {}).get("capability", {}).get("id", "not-owned") + assert not any(tool.name.startswith("personal_") for tool in get_available_tools(app_config=config, mcp_plugins=[peer_id])) + finally: + reset_current_user(identity) + + +def test_deployment_name_collision_does_not_publish_a_personal_tool(personal_client, monkeypatch, tmp_path): + from langchain_core.tools import StructuredTool + + from deerflow.tools.mcp_metadata import tag_mcp_tool + from deerflow.tools.tools import get_available_tools + + assert create(personal_client, "alice").status_code == 200 + name = next(iter(load_user_mcp_config("alice").mcp_servers)) + path = tmp_path / "deployment.json" + path.write_text(json.dumps({"mcpServers": {name: {"type": "http", "url": "https://example.com/platform"}}})) + monkeypatch.setenv("DEER_FLOW_EXTENSIONS_CONFIG_PATH", str(path)) + platform = tag_mcp_tool(StructuredTool.from_function(lambda: "platform", name=name + "_test", description="Shared tool"), server_name=name) + monkeypatch.setattr("deerflow.mcp.cache.get_cached_mcp_tools", lambda: [platform]) + + async def discover(config, **kwargs): + async def personal(): + return "personal" + + return [tag_mcp_tool(StructuredTool.from_function(coroutine=personal, name=name + "_test", description="Personal tool"), server_name=name)] + + monkeypatch.setattr("deerflow.mcp.tools.get_mcp_tools", discover) + config = SimpleNamespace(tools=[], models=[], get_model_config=lambda _: None) + identity = set_current_user(SimpleNamespace(id="alice")) + try: + tools = get_available_tools(app_config=config) + assert [tool for tool in tools if tool.name == name + "_test"] == [platform] + personal_id = read_user_mcp_config("alice")["mcpServers"]["github"]["capability"]["id"] + selected = get_available_tools(app_config=config, mcp_plugins=[personal_id]) + assert not any(tool.name == name + "_test" for tool in selected) + finally: + reset_current_user(identity) + + +def test_untrusted_users_cannot_launch_packages_or_connect_to_private_hosts(personal_client, monkeypatch): + client = personal_client + response = client.post("/api/mcp/personal/config/servers", headers={"test-user": "alice", "test-role": "user"}, json={"mcp_servers": {"shell": {"type": "stdio", "command": "npx", "args": ["untrusted-package"]}}}) + assert response.status_code == 403 + response = client.post("/api/mcp/personal/config/servers", headers={"test-user": "alice", "test-role": "user"}, json={"mcp_servers": {"private": {"type": "http", "url": "http://127.0.0.1/mcp", "personal_public_network": False}}}) + assert response.status_code == 400 + monkeypatch.setattr("deerflow.community.url_safety.validate_public_http_url", lambda *a, **k: None) + assert create(client, "alice", role="user").status_code == 200 + assert read_user_mcp_config("alice")["mcpServers"]["github"]["personal_public_network"] is True + + +@pytest.mark.asyncio +async def test_personal_network_rechecks_destination_before_each_request(monkeypatch): + from deerflow.mcp import personal_network + + received = [] + blocked = False + + def transport(request): + received.append(str(request.url)) + return httpx.Response(302, headers={"location": "http://127.0.0.1/private"}) + + monkeypatch.setattr(personal_network.httpx, "AsyncHTTPTransport", lambda **kwargs: httpx.MockTransport(transport)) + monkeypatch.setattr(personal_network, "resolve_host_addresses", lambda host: [ipaddress.ip_address("127.0.0.1" if blocked else "8.8.8.8")]) + async with personal_network.personal_httpx_client_factory() as client: + assert (await client.get("https://example.com/mcp")).status_code == 302 + assert received == ["https://8.8.8.8/mcp"] + blocked = True + with pytest.raises(ValueError, match="public HTTP"): + await client.get("https://example.com/mcp") + assert len(received) == 1 + + +@pytest.mark.asyncio +async def test_real_mcp_calls_keep_credentials_separate_and_reject_stale_tools(personal_client, monkeypatch): + from mcp.server.fastmcp import Context, FastMCP + from mcp.server.transport_security import TransportSecuritySettings + + from deerflow.mcp import client as mcp_client + from deerflow.mcp.user_tools import _load + + server = FastMCP("identity", stateless_http=True, json_response=True, transport_security=TransportSecuritySettings(allowed_hosts=["example.com"])) + received = [] + + @server.tool() + async def whoami(ctx: Context) -> str: + credential = ctx.request_context.request.headers.get("authorization") + received.append(credential) + return credential + + app = server.streamable_http_app() + build = mcp_client.build_server_params + + def params(name, config): + result = build(name, config) + result["httpx_client_factory"] = lambda headers=None, timeout=None, auth=None: httpx.AsyncClient(transport=httpx.ASGITransport(app=app), headers=headers, timeout=timeout or 30, auth=auth) + return result + + monkeypatch.setattr(mcp_client, "build_server_params", params) + for user in ("alice", "bob"): + assert create(personal_client, user).status_code == 200 + + async def run(user): + identity = set_current_user(SimpleNamespace(id=user)) + try: + tools = await _load(user, load_user_mcp_config(user)) + assert len(tools) == 1 + await tools[0].ainvoke({}) + return tools[0] + finally: + reset_current_user(identity) + + async with app.router.lifespan_context(app): + alice_tool, bob_tool = await asyncio.gather(run("alice"), run("bob")) + assert sorted(received) == ["Bearer alice", "Bearer bob"] + assert alice_tool.name != bob_tool.name + identity = set_current_user(SimpleNamespace(id="bob")) + try: + with pytest.raises(ToolException, match="another user"): + await alice_tool.ainvoke({}) + assert personal_client.patch("/api/mcp/personal/config", headers={"test-user": "bob"}, json={"server_name": "github", "enabled": False}).status_code == 200 + with pytest.raises(ToolException, match="changed, disabled or removed"): + await bob_tool.ainvoke({}) + assert len(received) == 2 + finally: + reset_current_user(identity) + + +@pytest.mark.asyncio +async def test_personal_tool_guard_reuses_validation_and_rejects_file_replacement(personal_client, monkeypatch): + import os + + from langchain_core.tools import StructuredTool + + import deerflow.mcp.user_config as user_config + from deerflow.config.extensions_config import atomic_write_extensions_config + from deerflow.mcp.user_tools import _guard + + assert create(personal_client, "alice").status_code == 200 + old_name = next(iter(load_user_mcp_config("alice").mcp_servers)) + load = user_config.load_user_mcp_config + reads = 0 + + def counted_load(user_id): + nonlocal reads + reads += 1 + return load(user_id) + + monkeypatch.setattr(user_config, "load_user_mcp_config", counted_load) + calls = [] + + async def invoke(): + calls.append(True) + return "ok" + + tool = _guard(StructuredTool.from_function(coroutine=invoke, name="personal_test", description="Personal test"), "alice", old_name) + identity = set_current_user(SimpleNamespace(id="alice")) + try: + assert await tool.ainvoke({}) == "ok" + assert await tool.ainvoke({}) == "ok" + assert reads == 1 + + path = user_mcp_config_path("alice") + before = path.stat() + raw = read_user_mcp_config("alice") + raw["mcpServers"]["github"]["headers"]["Authorization"] = "Bearer ALICE" + atomic_write_extensions_config(path, raw) + os.utime(path, ns=(before.st_atime_ns, before.st_mtime_ns)) + after = path.stat() + assert (after.st_size, after.st_mtime_ns) == (before.st_size, before.st_mtime_ns) + assert after.st_ino != before.st_ino + with pytest.raises(ToolException, match="changed, disabled or removed"): + await tool.ainvoke({}) + assert reads == 2 + assert len(calls) == 2 + finally: + reset_current_user(identity) + + +@pytest.mark.asyncio +async def test_background_calls_resolve_only_persisted_task_owner(personal_client, monkeypatch): + import deerflow.mcp.user_config as user_config + from deerflow.mcp.task_tool_caller import McpTaskToolCaller + + create(personal_client, "alice") + name = next(iter(load_user_mcp_config("alice").mcp_servers)) + received = [] + loaded = [] + load = user_config.load_user_mcp_config + + def counted_load(user_id): + loaded.append(user_id) + return load(user_id) + + monkeypatch.setattr(user_config, "load_user_mcp_config", counted_load) + + async def invoke(self, **kwargs): + received.append(self._extensions_config.mcp_servers[kwargs["server_name"]].headers["Authorization"]) + + monkeypatch.setattr(McpTaskToolCaller, "_call_configured_tool", invoke) + caller = McpTaskToolCaller(ExtensionsConfig()) + await caller.call_tool(server_name=name, tool_name="status", arguments={}, user_id="alice", thread_id="thread", connection_scope="personal") + first_caller = caller._personal_callers["alice"][1] + await caller.call_tool(server_name=name, tool_name="status", arguments={}, user_id="alice", thread_id="thread", connection_scope="personal") + assert caller._personal_callers["alice"][1] is first_caller + assert loaded == ["alice"] + with pytest.raises(LookupError, match="Personal MCP"): + await caller.call_tool(server_name=name, tool_name="status", arguments={}, user_id="bob", thread_id="thread", connection_scope="personal") + assert received == ["Bearer alice", "Bearer alice"] + + # The same deployment name must not steal an existing personal task after + # the Gateway restarts with that deployment entry in its startup snapshot. + deployment = McpTaskToolCaller(ExtensionsConfig.model_validate({"mcpServers": {name: {"type": "http", "url": "https://example.com/platform", "headers": {"Authorization": "platform"}}}})) + await deployment.call_tool(server_name=name, tool_name="status", arguments={}, user_id="alice", thread_id="thread", connection_scope="personal") + await deployment.call_tool(server_name=name, tool_name="status", arguments={}, user_id="bob", thread_id="thread") + assert received == ["Bearer alice", "Bearer alice", "Bearer alice", "platform"] + assert personal_client.patch("/api/mcp/personal/config", headers={"test-user": "alice"}, json={"server_name": "github", "enabled": False}).status_code == 200 + with pytest.raises(LookupError, match="Personal MCP"): + await deployment.call_tool(server_name=name, tool_name="status", arguments={}, user_id="alice", thread_id="thread", connection_scope="personal") + + +@pytest.mark.asyncio +async def test_background_caller_rebuilds_after_personal_credential_edit(personal_client, monkeypatch): + from deerflow.config.extensions_config import atomic_write_extensions_config + from deerflow.mcp.task_tool_caller import McpTaskToolCaller + + assert create(personal_client, "alice").status_code == 200 + old_name = next(iter(load_user_mcp_config("alice").mcp_servers)) + + async def credential(self, **kwargs): + return self._extensions_config.mcp_servers[kwargs["server_name"]].headers["Authorization"] + + monkeypatch.setattr(McpTaskToolCaller, "_call_configured_tool", credential) + caller = McpTaskToolCaller(ExtensionsConfig()) + request = {"tool_name": "status", "arguments": {}, "user_id": "alice", "thread_id": "thread", "connection_scope": "personal"} + assert await caller.call_tool(server_name=old_name, **request) == "Bearer alice" + previous = caller._personal_callers["alice"][1] + + path = user_mcp_config_path("alice") + raw = read_user_mcp_config("alice") + raw["mcpServers"]["github"]["headers"]["Authorization"] = "Bearer ALICE" + atomic_write_extensions_config(path, raw) + with pytest.raises(LookupError, match="Personal MCP"): + await caller.call_tool(server_name=old_name, **request) + new_name = next(iter(load_user_mcp_config("alice").mcp_servers)) + assert await caller.call_tool(server_name=new_name, **request) == "Bearer ALICE" + assert caller._personal_callers["alice"][1] is not previous + path.unlink() + with pytest.raises(LookupError, match="Personal MCP"): + await caller.call_tool(server_name=new_name, **request) + + +@pytest.mark.asyncio +async def test_background_caller_keeps_only_recent_owners(personal_client, monkeypatch): + import deerflow.mcp.task_tool_caller as task_tool_caller + + monkeypatch.setattr(task_tool_caller, "_MAX_PERSONAL_CALLERS", 2) + + async def credential(self, **kwargs): + return self._extensions_config.mcp_servers[kwargs["server_name"]].headers["Authorization"] + + monkeypatch.setattr(task_tool_caller.McpTaskToolCaller, "_call_configured_tool", credential) + caller = task_tool_caller.McpTaskToolCaller(ExtensionsConfig()) + names = {} + for user_id in ("alice", "bob", "carol"): + assert create(personal_client, user_id).status_code == 200 + names[user_id] = next(iter(load_user_mcp_config(user_id).mcp_servers)) + assert await caller.call_tool(server_name=names[user_id], tool_name="status", arguments={}, user_id=user_id, thread_id="thread", connection_scope="personal") == f"Bearer {user_id}" + assert list(caller._personal_callers) == ["bob", "carol"] + assert await caller.call_tool(server_name=names["alice"], tool_name="status", arguments={}, user_id="alice", thread_id="thread", connection_scope="personal") == "Bearer alice" + assert list(caller._personal_callers) == ["carol", "alice"] + + +@pytest.mark.asyncio +async def test_gateway_registers_driver_for_personal_only_task_toolsets(personal_client): + from langchain_core.tools import StructuredTool + + from app.gateway.app import lifespan + from deerflow.config.extensions_config import atomic_write_extensions_config + from deerflow.config.mcp_tasks_config import McpTasksConfig + from deerflow.mcp.tasks import ORDINARY_MCP_TASK_DRIVER + from deerflow.mcp.tools import get_mcp_tools + + path = user_mcp_config_path("alice") + atomic_write_extensions_config( + path, + {"mcpServers": {"reports": {"type": "http", "url": "https://example.com/mcp", "task_toolsets": [{"name": "reports", "submit_tool": "submit_report", "status_tool": "status_report", "cancel_tool": "cancel_report"}]}}}, + ) + personal = load_user_mcp_config("alice") + server_name = next(iter(personal.mcp_servers)) + deployment = ExtensionsConfig() + startup = SimpleNamespace(log_level="INFO", memory=SimpleNamespace(enabled=False, token_counting="char", shutdown_flush_timeout_seconds=5.0), mcp_tasks=McpTasksConfig(enabled=True)) + app = FastAPI() + + @asynccontextmanager + async def runtime(gateway, _config): + gateway.state.mcp_task_repo = object() + yield + + class FakeClient: + def __init__(self, _servers, *, tool_interceptors, **_kwargs): + self.tool_interceptors = tool_interceptors + self.callbacks = None + + async def get_tools(self, *, server_name): + async def call(topic: str) -> str: + return topic + + return [StructuredTool.from_function(coroutine=call, name=f"{server_name}_{name}", description=name) for name in ("submit_report", "status_report", "cancel_report")] + + channel = MagicMock() + channel.get_status.return_value = {} + with ( + patch("app.gateway.app.get_app_config", return_value=startup), + patch("app.gateway.app.get_gateway_config", return_value=MagicMock(host="x", port=0)), + patch("app.gateway.app.langgraph_runtime", runtime), + patch("app.gateway.app.auth.close_oidc_service", AsyncMock()), + patch("app.channels.service.start_channel_service", AsyncMock(return_value=channel)), + patch("app.channels.service.stop_channel_service", AsyncMock()), + patch("deerflow.skills.projection.ensure_public_skill_projection"), + patch("deerflow.agents.memory.get_memory_manager", return_value=MagicMock()), + patch("deerflow.config.extensions_config.ExtensionsConfig.from_file", return_value=deployment), + patch("app.mcp_tasks.McpTaskService.start", AsyncMock()), + patch("app.mcp_tasks.McpTaskService.stop", AsyncMock()), + patch("langchain_mcp_adapters.client.MultiServerMCPClient", FakeClient), + ): + async with lifespan(app): + assert not deployment.mcp_servers + assert app.state.mcp_task_service.drivers.get(ORDINARY_MCP_TASK_DRIVER) is not None + tools = await get_mcp_tools(personal, personal_user_id="alice") + assert [tool.name for tool in tools] == [f"{server_name}_submit_report"] + app.state.mcp_task_service.submit = AsyncMock(return_value={"id": "local-1", "status": "submitted"}) + identity = set_current_user(SimpleNamespace(id="alice")) + try: + result = await tools[0].coroutine( + runtime=SimpleNamespace(context={"thread_id": "thread-1", "thread_incarnation": "incarnation-1", "run_id": "run-1"}, config={}, tool_call_id="call-1"), + topic="MCP", + ) + finally: + reset_current_user(identity) + assert result["task_id"] == "local-1" + submitted = app.state.mcp_task_service.submit.await_args.kwargs + assert submitted["driver_name"] == ORDINARY_MCP_TASK_DRIVER + assert submitted["request"].driver_data["connection_scope"] == "personal" diff --git a/backend/tests/test_personal_mcp_access.py b/backend/tests/test_personal_mcp_access.py new file mode 100644 index 0000000000..0bfba74d13 --- /dev/null +++ b/backend/tests/test_personal_mcp_access.py @@ -0,0 +1,172 @@ +"""Privilege revocation uses current accounts even with stale tools and caches.""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock + +import pytest +import pytest_asyncio +from langchain_core.tools import StructuredTool, ToolException +from sqlalchemy import delete +from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine + +from app.gateway import personal_mcp_access as gateway_access +from app.gateway.auth.local_provider import LocalAuthProvider +from app.gateway.auth.models import User +from app.gateway.auth.repositories.sqlite import SQLiteUserRepository +from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig, atomic_write_extensions_config +from deerflow.config.paths import Paths +from deerflow.mcp import personal_access +from deerflow.mcp.task_tool_caller import McpTaskToolCaller +from deerflow.mcp.user_config import load_user_mcp_config, user_mcp_config_path +from deerflow.mcp.user_tools import _guard, _load +from deerflow.persistence.user.model import UserRow +from deerflow.runtime.user_context import reset_current_user, set_current_user + + +@pytest_asyncio.fixture +async def accounts(tmp_path, monkeypatch): + engine = create_async_engine(f"sqlite+aiosqlite:///{tmp_path}/users.db") + async with engine.begin() as connection: + await connection.run_sync(UserRow.__table__.create) + sessions = async_sessionmaker(engine, expire_on_commit=False) + repository = SQLiteUserRepository(sessions) + user = await repository.create_user(User(email="admin@example.com", system_role="admin")) + monkeypatch.setattr(gateway_access, "get_local_provider", lambda: LocalAuthProvider(repository)) + monkeypatch.delenv("DEER_FLOW_AUTH_DISABLED", raising=False) + monkeypatch.setattr("deerflow.mcp.user_config.get_paths", lambda: Paths(base_dir=tmp_path)) + monkeypatch.setattr(personal_access, "_admin_checker", None) + try: + yield repository, user, sessions + finally: + await engine.dispose() + + +def privileged_definition(kind): + if kind == "stdio": + return {"type": "stdio", "command": "npx", "args": ["example-package"], "personal_public_network": False} + definition = {"type": "sse" if kind == "sse" else "http", "url": "http://127.0.0.1/mcp", "personal_public_network": False} + if kind == "legacy": + del definition["personal_public_network"] + if kind == "oauth": + definition["oauth"] = {"token_url": "http://127.0.0.1/token", "client_id": "test-client"} + return definition + + +@pytest.mark.asyncio +@pytest.mark.parametrize("kind", ["http", "sse", "stdio", "legacy", "oauth"]) +@pytest.mark.parametrize("revocation", ["demote", "delete", "lookup_failure"]) +async def test_revocation_blocks_discovery_existing_tools_and_cached_durable_calls(accounts, monkeypatch, kind, revocation): + repository, user, sessions = accounts + owner = str(user.id) + path = user_mcp_config_path(owner) + atomic_write_extensions_config(path, {"mcpServers": {"privileged": privileged_definition(kind), "public": {"type": "http", "url": "https://example.com/mcp", "personal_public_network": True}}}) + before = path.read_bytes() + config = load_user_mcp_config(owner) + privileged = next(name for name, server in config.mcp_servers.items() if personal_access.requires_admin(server)) + public = next(name for name in config.mcp_servers if name != privileged) + observed = [] + + async def execute(): + observed.append("executed") + return "ok" + + class Client: + def __init__(self, connections, **kwargs): + observed.extend(connections) + + async def get_tools(self, server_name): + return [StructuredTool.from_function(coroutine=execute, name=server_name + "_probe", description="Probe")] + + monkeypatch.setattr("langchain_mcp_adapters.client.MultiServerMCPClient", Client) + oauth = AsyncMock(return_value={}) + monkeypatch.setattr("deerflow.mcp.tools.get_initial_oauth_headers", oauth) + invoke = AsyncMock(return_value="ok") + monkeypatch.setattr(McpTaskToolCaller, "_call_configured_tool", invoke) + caller = McpTaskToolCaller(ExtensionsConfig()) + guarded = _guard(StructuredTool.from_function(coroutine=execute, name="probe", description="Probe"), owner, privileged) + public_tool = _guard(StructuredTool.from_function(coroutine=execute, name="public", description="Public"), owner, public) + token = set_current_user(SimpleNamespace(id=owner, system_role="admin")) + try: + async with gateway_access.personal_mcp_authority(): + assert len(await _load(owner, config)) == 2 + assert await guarded.ainvoke({}) == "ok" + call = dict(server_name=privileged, tool_name="submit", arguments={}, user_id=owner, thread_id="thread", connection_scope="personal") + assert await caller.call_tool(**call) == "ok" + cached_caller = caller._personal_callers[owner][1] + observed.clear() + invoke.reset_mock() + oauth.reset_mock() + + if revocation == "demote": + user.system_role = "user" + await repository.update_user(user) + elif revocation == "delete": + async with sessions() as session: + await session.execute(delete(UserRow).where(UserRow.id == owner)) + await session.commit() + else: + monkeypatch.setattr(repository, "get_user_by_id", AsyncMock(side_effect=RuntimeError("database unavailable"))) + + tools = await _load(owner, config) + assert len(tools) == 1 + assert observed == [public] + assert set(oauth.call_args.args[0].mcp_servers) == {public} + with pytest.raises(ToolException, match="Current administrator"): + await guarded.ainvoke({}) + for operation in ("submit", "status", "cancel"): + with pytest.raises(ToolException, match="Current administrator"): + await caller.call_tool(**(call | {"tool_name": operation, "request_scoped_headers": operation == "submit"})) + assert caller._personal_callers[owner][1] is cached_caller + invoke.assert_not_awaited() + assert path.read_bytes() == before + assert await public_tool.ainvoke({}) == "ok" + # Deployment calls retain their existing policy and never query + # personal ownership or the owner's current role. + assert await caller.call_tool(**(call | {"connection_scope": "deployment"})) == "ok" + finally: + reset_current_user(token) + + +@pytest.mark.asyncio +async def test_authority_lookup_stays_on_gateway_loop_during_sync_discovery(accounts, monkeypatch): + repository, user, _ = accounts + owner = str(user.id) + loop = asyncio.get_running_loop() + original = repository.get_user_by_id + reads = [] + + async def get_user(user_id): + reads.append(asyncio.get_running_loop()) + return await original(user_id) + + monkeypatch.setattr(repository, "get_user_by_id", get_user) + server = McpServerConfig.model_validate(privileged_definition("http")) + async with gateway_access.personal_mcp_authority(): + await asyncio.wait_for(asyncio.to_thread(lambda: asyncio.run(personal_access.require_personal_mcp_access(owner, server))), timeout=5) + assert reads == [loop] + assert personal_access._admin_checker is None + with pytest.raises(ToolException, match="Current administrator"): + await personal_access.require_personal_mcp_access(owner, server) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("production", [False, True]) +async def test_auth_disabled_authority_only_grants_the_default_local_user(accounts, monkeypatch, production): + monkeypatch.setenv("DEER_FLOW_AUTH_DISABLED", "1") + monkeypatch.setenv("DEER_FLOW_ENV", "production" if production else "development") + monkeypatch.delenv("ENVIRONMENT", raising=False) + async with gateway_access.personal_mcp_authority(): + assert await personal_access._is_current_admin("default") is (not production) + assert await personal_access._is_current_admin("missing-user") is False + + +@pytest.mark.asyncio +async def test_authority_registration_is_restored_after_startup_failure(monkeypatch): + previous = AsyncMock(return_value=False) + monkeypatch.setattr(personal_access, "_admin_checker", previous) + with pytest.raises(RuntimeError, match="startup failed"): + async with gateway_access.personal_mcp_authority(): + assert personal_access._admin_checker is not previous + raise RuntimeError("startup failed") + assert personal_access._admin_checker is previous diff --git a/backend/tests/test_personal_mcp_network.py b/backend/tests/test_personal_mcp_network.py new file mode 100644 index 0000000000..dc83ba8a05 --- /dev/null +++ b/backend/tests/test_personal_mcp_network.py @@ -0,0 +1,154 @@ +"""Exercise personal MCP policy through the real HTTP transport and TLS boundary.""" + +import ipaddress +import socket + +import httpcore +import pytest +from httpcore._backends.anyio import AnyIOBackend + +from deerflow.mcp.personal_network import personal_httpx_client_factory + + +class RecordingStream(httpcore.AsyncNetworkStream): + def __init__(self): + self.writes = [] + self.tls_hosts = [] + self.closed = False + + async def read(self, max_bytes, timeout=None): + return b"HTTP/1.1 200 OK\r\nContent-Length: 2\r\n\r\nok" + + async def write(self, buffer, timeout=None): + self.writes.append(buffer) + + async def aclose(self): + self.closed = True + + async def start_tls(self, ssl_context, server_hostname=None, timeout=None): + assert ssl_context.check_hostname + self.tls_hosts.append(server_hostname) + return self + + +@pytest.mark.asyncio +@pytest.mark.parametrize("scheme", ["http", "https"]) +async def test_personal_transport_pins_dns_and_preserves_authority(monkeypatch, scheme): + lookups = [] + connections = [] + streams = [] + real_resolve = socket.getaddrinfo + + def resolve(host, port, *args, **kwargs): + if host == "mcp.example": + address = "8.8.8.8" if not lookups else "127.0.0.1" + lookups.append(address) + return real_resolve(address, port, *args, **kwargs) + return real_resolve(host, port, *args, **kwargs) + + async def connect(self, host, port, **kwargs): + addresses = socket.getaddrinfo(host, port, type=socket.SOCK_STREAM) + connections.append((host, addresses[0][4][0])) + stream = RecordingStream() + streams.append(stream) + return stream + + monkeypatch.setattr(socket, "getaddrinfo", resolve) + monkeypatch.setattr(AnyIOBackend, "connect_tcp", connect) + async with personal_httpx_client_factory(headers={"Authorization": "Bearer test"}) as client: + response = await client.post(f"{scheme}://mcp.example:8443/mcp?session=1", content=b"payload") + assert response.text == "ok" + assert str(response.request.url) == f"{scheme}://mcp.example:8443/mcp?session=1" + assert connections == [("8.8.8.8", "8.8.8.8")] + assert lookups == ["8.8.8.8"] + wire = b"".join(streams[0].writes) + assert b"POST /mcp?session=1 HTTP/1.1" in wire + assert b"Host: mcp.example:8443" in wire + assert b"Authorization: Bearer test" in wire + assert b"payload" in wire + assert streams[0].tls_hosts == (["mcp.example"] if scheme == "https" else []) + with pytest.raises(ValueError, match="public HTTP"): + await client.get(f"{scheme}://mcp.example:8443/mcp") + assert len(connections) == 1 + assert all(stream.closed for stream in streams) + + +@pytest.mark.asyncio +async def test_personal_transport_does_not_share_tls_across_hostnames(monkeypatch): + streams = [] + real_resolve = socket.getaddrinfo + monkeypatch.setattr(socket, "getaddrinfo", lambda host, port, *args, **kwargs: real_resolve("8.8.8.8", port, *args, **kwargs)) + + async def connect(self, host, port, **kwargs): + stream = RecordingStream() + streams.append(stream) + return stream + + monkeypatch.setattr(AnyIOBackend, "connect_tcp", connect) + async with personal_httpx_client_factory() as client: + for host in ("first.example", "second.example"): + assert (await client.get(f"https://{host}/mcp")).status_code == 200 + assert [stream.tls_hosts for stream in streams] == [["first.example"], ["second.example"]] + assert all(stream.closed for stream in streams) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("address", ["127.0.0.1", "10.0.0.1", "169.254.169.254", "::1"]) +async def test_personal_transport_blocks_private_addresses_before_connect(monkeypatch, address): + real_resolve = socket.getaddrinfo + monkeypatch.setattr(socket, "getaddrinfo", lambda host, port, *args, **kwargs: real_resolve(address, port, *args, **kwargs)) + + async def connect(*args, **kwargs): + pytest.fail("Private address reached TCP connect") + + monkeypatch.setattr(AnyIOBackend, "connect_tcp", connect) + async with personal_httpx_client_factory() as client: + with pytest.raises(ValueError, match="public HTTP"): + await client.get("https://mcp.example/mcp") + + +@pytest.mark.asyncio +@pytest.mark.parametrize("addresses", [[], ["8.8.8.8", "127.0.0.1"], ["2606:4700:4700::1111", "::1"]]) +async def test_personal_transport_rejects_empty_or_mixed_dns_answers(monkeypatch, addresses): + monkeypatch.setattr("deerflow.mcp.personal_network.resolve_host_addresses", lambda host: [ipaddress.ip_address(address) for address in addresses]) + + async def connect(*args, **kwargs): + pytest.fail("Rejected DNS answers reached TCP connect") + + monkeypatch.setattr(AnyIOBackend, "connect_tcp", connect) + async with personal_httpx_client_factory() as client: + with pytest.raises(ValueError, match="public HTTP"): + await client.get("https://mcp.example/mcp") + + +@pytest.mark.asyncio +async def test_personal_transport_tries_only_vetted_addresses(monkeypatch): + addresses = ["2606:4700:4700::1111", "8.8.8.8"] + monkeypatch.setattr("deerflow.mcp.personal_network.resolve_host_addresses", lambda host: [ipaddress.ip_address(address) for address in addresses]) + attempts = [] + + async def connect(self, host, port, **kwargs): + attempts.append(host) + if host == addresses[0]: + raise httpcore.ConnectError("IPv6 unavailable") + return RecordingStream() + + monkeypatch.setattr(AnyIOBackend, "connect_tcp", connect) + async with personal_httpx_client_factory() as client: + assert (await client.get("https://mcp.example/mcp")).status_code == 200 + assert attempts == addresses + + +@pytest.mark.asyncio +@pytest.mark.parametrize("address", ["8.8.8.8", "[2606:4700:4700::1111]"]) +async def test_personal_transport_accepts_public_literal_ips(monkeypatch, address): + attempts = [] + + async def connect(self, host, port, **kwargs): + attempts.append(host) + return RecordingStream() + + monkeypatch.setattr(AnyIOBackend, "connect_tcp", connect) + async with personal_httpx_client_factory() as client: + assert (await client.get(f"http://{address}/mcp")).status_code == 200 + assert attempts == [address.strip("[]")] diff --git a/docs/capability-center.md b/docs/capability-center.md index 476c11e5ab..d92b3ab8bf 100644 --- a/docs/capability-center.md +++ b/docs/capability-center.md @@ -20,7 +20,8 @@ deferred to a separate proposal and PR. | Validated manifest schema | `deerflow.capabilities.catalog.PluginManifest` | | Installation/status adapters | `backend/app/gateway/capabilities.py` | | Catalog and safe discovery HTTP API | `backend/app/gateway/routers/capabilities.py` | -| MCP settings, secrets, enable/delete, cache reload | Existing `/api/mcp/config` services and `extensions_config.json` | +| Personal MCP settings, secrets, enable/delete | `/api/mcp/personal/config` and `.deer-flow/users//integrations/mcp.json` | +| Deployment MCP settings and cache reload | Administrator-only `/api/mcp/config` and `extensions_config.json` | | Lark installation and personal account authorization | Existing `/api/integrations/lark` services | | Skill archives, enable state, user storage | Existing `/api/skills` services | | Agent selection | Agent config `mcp_plugins` and existing `skills` | @@ -35,7 +36,7 @@ loader is separate from this user-facing directory. Three entries ship working API clients with the harness, using the `business` configuration adapter and the existing stdio MCP runtime. No separate service, package download, or Dify runtime is needed. Configure them under **Capability -Center → Plugins** as an administrator: +Center → Plugins** using your own account: | Plugin | Configuration | Tools | | --- | --- | --- | @@ -51,10 +52,15 @@ DeerFlow's incoming IM channels. Existing manually configured CLI connections are not rewritten when the catalog entry changes. Configuration saves credentials without sending a message or creating a CRM -record. These deployment credentials are shared by runs allowed to use the -configured MCP server; they are not personal OAuth connections. Credentials -remain in the existing MCP `env` configuration and its masked admin editor. -Edit, toggle and delete the configured entry using the existing MCP controls. +record. Connections created in the web interface are personal: credentials, +enabled state and edits belong to the signed-in user, including administrators. +They are stored under `.deer-flow/users//integrations/mcp.json` and +remain masked in the editor. Existing deployment connections in +`extensions_config.json` stay shared and are not copied to any personal account. +The page separates **Platform provided** (deployment setup guides and shared MCP +entries) from **My plugins** (personal connections, configuration templates and +Lark account authorization). Existing search and category filters apply to both. +Edit, toggle and delete personal entries using the existing MCP controls. An Agent selects these connections through **Plugins and skills**, just like other MCP servers. New tool selection applies on the next run. @@ -118,7 +124,7 @@ only link to setup instructions; listing them never claims they are installed. The catalog version describes this manifest, not a remotely detected server version. A catalog listing does not install, enable, or authorize anything. -The first MCP form supports HTTP endpoints and a deployment Authorization header. +The MCP form saves a personal HTTP endpoint and Authorization header. The advanced JSON editor retains stdio, SSE, OAuth token configuration, and per-user credential mappings supported by the existing MCP runtime. The presence of `oauth` in a manifest is not a new interactive OAuth implementation. Lark @@ -134,8 +140,11 @@ service's authorization and validation; never add a provider branch to the galle - `GET /api/capabilities/catalog`: validated bundled manifests. - `GET /api/capabilities/installations/{adapter}`: safe installation projections for `mcp`, `business`, `lark`, and `skills`; separate requests isolate integration failures. -- `POST /api/capabilities/installations`: administrator installation dispatch. - Body: `plugin_id`, `name`, and adapter `configuration`. MCP configuration uses +- `POST /api/capabilities/installations`: installation dispatch with `plugin_id`, + `name`, adapter `configuration`, and optional `scope` (default `deployment`). + Deployment scope requires an administrator; authenticated users may choose + `scope: "user"` for MCP and bundled business connections. Other adapters + still require an administrator. MCP configuration uses the existing server definition schema. HTTP/SSE connections require a valid HTTP(S) URL without embedded credentials; stdio connections require a command. Invalid transport configuration returns 422 before saving. Duplicate server @@ -204,3 +213,42 @@ Agent persistence, delegated selection, and a real local stdio MCP invocation. Browser tests cover catalog installation, Agent selection save/reopen, existing icon editing, filtering, and desktop/mobile settings layouts. Browser fixture screenshots show sample integrations; they are not production default settings. + +## Personal and deployment MCP configuration + +`config.yaml` tools and existing `extensions_config.json` MCP servers remain +platform-provided capabilities. The administrator-only `/api/mcp/config` API +continues to manage deployment MCP configuration. + +The web editor uses `/api/mcp/personal/config`: GET reads only the authenticated +user's definitions; POST `/servers`, PUT `/server`, PATCH and DELETE +`/servers/{name}` manage that same user's file. No caller-supplied user ID selects +the owner. Catalog installation accepts `scope: "user"`; catalog discovery +accepts `scope=user` or `scope=all` (deployment plus the caller's connections). +The default API scope remains deployment for existing integrations. + +Personal files use the existing atomic file writer and cross-process lock. Keep +`DEER_FLOW_HOME` on persistent storage; multiple Gateway workers must share it. +Personal values are literal and cannot read deployment environment variables. +A missing personal connection never falls back to a same-named shared one. +Personal tools are discovered per run, outside the process-wide platform cache. +Calls check ownership and the current stored revision. Credential edits, disable +and delete invalidate already assembled personal tools; restart the run after +an edit. Durable tasks bind the same owner and connection revision, including +after recovery; changing that connection blocks its pending remote calls until +the original configuration is restored. + +Allowing personal configuration does not grant host execution privileges: +ordinary users may configure public HTTP/SSE endpoints and the fixed bundled +business launchers. Arbitrary stdio commands, internal endpoints and OAuth token +endpoints still require an administrator. Public endpoints are checked again on +HTTP requests, with redirects and environment proxies disabled. + +Connections saved with administrator privileges require the owner's current +administrator role during discovery and before every tool call, including +durable submit, poll and cancel calls. Demotion, account deletion or an authority +lookup failure blocks those connections even if tools or callers are cached. +The owner can save a compatible connection again under ordinary-user policy. +Gateway workers query current account records; standalone hosts must install +an authority lookup to use privileged personal definitions. Already admitted +operations are not forcibly terminated by a subsequent role change. diff --git a/docs/pr-evidence/connection-ownership-capability-center.png b/docs/pr-evidence/connection-ownership-capability-center.png new file mode 100644 index 0000000000..b9f08201ec Binary files /dev/null and b/docs/pr-evidence/connection-ownership-capability-center.png differ diff --git a/frontend/src/AGENTS.md b/frontend/src/AGENTS.md index 2ec508b0a2..443ebc35d4 100644 --- a/frontend/src/AGENTS.md +++ b/frontend/src/AGENTS.md @@ -39,13 +39,14 @@ against raw artifact paths before encoding. pages; `resolveThreadContext` must neither enqueue account writes nor create a fallback thread override that masks a later server preference. The - Capability Center > Plugins MCP switch calls the targeted `PATCH /api/mcp/config` + Capability Center > My plugins MCP switch calls the targeted `PATCH /api/mcp/personal/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. - Server management uses targeted `POST /api/mcp/config/servers`, - `PUT /api/mcp/config/server`, and bodyless - `DELETE /api/mcp/config/servers/{server_name}` mutations. Delete names are + Personal server management uses targeted `POST /api/mcp/personal/config/servers`, + `PUT /api/mcp/personal/config/server`, and bodyless + `DELETE /api/mcp/personal/config/servers/{server_name}` mutations. The deployment + `/api/mcp/config` endpoints remain operator-only and are not used by this editor. Delete names are percent-encoded, including legacy empty and slash-containing names; every successful mutation invalidates `["mcpConfig"]` only after the response. Current-chat MCP background tasks use `core/background-tasks`: the header @@ -196,3 +197,5 @@ are sent on save. Custom-agent chat derives its initial selection from the saved binding; explicit page-local overrides survive new-thread route replacement and reset on conversation changes. Gateway supplies defaults for clients without a selector; frontend visibility must not become a runtime enforcement boundary. + +MCP editor requests use the personal configuration API. Query keys for personal MCP config and capability installations include the authenticated user ID; remount the editor when that ID changes. Deployment installations remain read-only under Platform provided, alongside deployment setup guides. My plugins contains personal configuration templates, Lark account authorization and the personal MCP editor; a deployment installation must not mark a personal template configured. diff --git a/frontend/src/components/workspace/capabilities/plugin-directory.tsx b/frontend/src/components/workspace/capabilities/plugin-directory.tsx index d5c04bce13..6a419db06c 100644 --- a/frontend/src/components/workspace/capabilities/plugin-directory.tsx +++ b/frontend/src/components/workspace/capabilities/plugin-directory.tsx @@ -1,6 +1,6 @@ "use client"; -import { type ReactNode } from "react"; +import { type ReactNode, useId } from "react"; import { useI18n } from "@/core/i18n/hooks"; @@ -77,6 +77,7 @@ export function PluginDirectory({ }) { const { t } = useI18n(); const copy = t.capabilities.directory; + const groupId = useId(); const visible = entries.filter( (entry) => (!installedOnly || entry.installed) && @@ -98,12 +99,9 @@ export function PluginDirectory({ const items = visible.filter((entry) => entry.category === key); if (!items.length) return null; return ( -
+
-

+

{copy.categories[key]}

diff --git a/frontend/src/components/workspace/capabilities/plugin-gallery.tsx b/frontend/src/components/workspace/capabilities/plugin-gallery.tsx index 1e0479ce2f..1fbd58a40c 100644 --- a/frontend/src/components/workspace/capabilities/plugin-gallery.tsx +++ b/frontend/src/components/workspace/capabilities/plugin-gallery.tsx @@ -76,7 +76,7 @@ export function PluginGallery({ query }: { query: string }) { const copy = t.capabilities.directory; const labels = capabilityCopy(locale); const { user } = useAuth(); - const canManage = user?.system_role === "admin" && !isStaticWebsiteOnly(); + const canManage = !!user && !isStaticWebsiteOnly(); const directory = useCapabilityCatalog(); const definitions = directory.data ?? []; const adapterNames = [ @@ -86,7 +86,11 @@ export function PluginGallery({ query }: { query: string }) { .filter((name) => name !== "guide"), ), ]; - const states = useQueries({ queries: adapterNames.map(installationQuery) }); + const states = useQueries({ + queries: adapterNames.map((adapter) => + installationQuery(adapter, user?.id), + ), + }); const client = useQueryClient(); const [filter, setFilter] = useState("all"); const [category, setCategory] = useState("all"); @@ -113,7 +117,9 @@ export function PluginGallery({ query }: { query: string }) { .map((item) => [item.id, item]), ).values(), ]; - const catalog: PluginDirectoryEntry[] = definitions + const platformCatalog: PluginDirectoryEntry[] = []; + const personalCatalog: PluginDirectoryEntry[] = []; + definitions .filter( (plugin) => canManage || @@ -121,12 +127,16 @@ export function PluginGallery({ query }: { query: string }) { (item) => item.adapter === "mcp" && item.plugin_id === plugin.id, ), ) - .map((plugin) => { + .forEach((plugin) => { + const isPlatform = plugin.adapter === "guide"; const status = installations.find( - (item) => item.plugin_id === plugin.id && item.installed, + (item) => + item.plugin_id === plugin.id && + item.installed && + (isPlatform ? item.scope !== "user" : item.scope === "user"), ); const unavailable = states[adapterNames.indexOf(plugin.adapter)]?.isError; - return { + (isPlatform ? platformCatalog : personalCatalog).push({ id: plugin.id, category: plugin.category, installed: !!status, @@ -163,50 +173,49 @@ export function PluginGallery({ query }: { query: string }) { ), - }; - }); - // Non-admins see safe installation projections, not the administrator's raw config editor. - if (!canManage) - for (const item of installations.filter((item) => item.adapter === "mcp")) { - const manifest = definitions.find( - (plugin) => plugin.id === item.plugin_id, - ); - catalog.push({ - id: item.id, - category: manifest?.category ?? "custom", - search: `${item.name} ${item.description}`, - installed: true, - node: ( - - } - label={ - item.selectable === false - ? labels.unavailable - : item.enabled - ? t.capabilities.enabled - : t.capabilities.disabled - } - > - - {item.auth_status === "required" - ? labels.required - : item.auth_status === "configured" - ? labels.configured - : labels.unknown} - - - ), }); - } + }); + // Deployment entries remain read-only; the editor manages personal connections. + for (const item of installations.filter( + (item) => item.adapter === "mcp" && item.scope !== "user", + )) { + const manifest = definitions.find((plugin) => plugin.id === item.plugin_id); + platformCatalog.push({ + id: item.id, + category: manifest?.category ?? "custom", + search: `${item.name} ${item.description}`, + installed: true, + node: ( + + } + label={ + item.selectable === false + ? labels.unavailable + : item.enabled + ? t.capabilities.enabled + : t.capabilities.disabled + } + > + + {item.auth_status === "required" + ? labels.required + : item.auth_status === "configured" + ? labels.configured + : labels.unknown} + + + ), + }); + } const toolbar = ( @@ -250,26 +259,69 @@ export function PluginGallery({ query }: { query: string }) {
{directory.isError &&

{labels.catalogError}

} {directory.isLoading &&

{t.common.loading}

} - {canManage ? ( - +
+

+ {labels.platformTitle} +

+

{labels.platformHint}

+
+ - ) : ( - <> - {toolbar} - +
+ {canManage ? ( + +

+ {labels.personalTitle} +

+

+ {labels.personalHint} +

+ + } category={category} installedOnly={filter === "installed"} - entries={catalog} + catalog={personalCatalog} + definitions={definitions} /> - - )} + ) : ( + <> +
+

+ {labels.personalTitle} +

+

+ {labels.personalHint} +

+
+ + + )} +
!value && close()}> {selected && ( @@ -287,10 +339,14 @@ export function PluginGallery({ query }: { query: string }) { {Settings ? ( ) : (

diff --git a/frontend/src/core/capabilities/copy.ts b/frontend/src/core/capabilities/copy.ts index 382f776339..3c984ad549 100644 --- a/frontend/src/core/capabilities/copy.ts +++ b/frontend/src/core/capabilities/copy.ts @@ -1,4 +1,8 @@ const en = { + platformTitle: "Platform provided", + platformHint: "Shared capabilities managed through deployment configuration.", + personalTitle: "My plugins", + personalHint: "Your connections and settings are only used by your account.", install: "Configure plugin", name: "Connection name", url: "Server URL", @@ -8,7 +12,7 @@ const en = { webhook_key: "Robot webhook key (the key parameter in its URL)", save: "Save configuration", accountHint: - "These are deployment credentials managed by the administrator. Personal authorization, when supported, uses the integration's account flow.", + "This configuration is saved to your account and is only used by your runs.", saved: "Configuration saved", unknown: "Not checked", required: "Account required", @@ -31,6 +35,10 @@ const en = { invalidUrl: "Enter an HTTP or HTTPS server URL.", }; const zh: typeof en = { + platformTitle: "平台提供", + platformHint: "由平台统一配置，供所有用户使用。", + personalTitle: "我的插件", + personalHint: "连接和配置仅属于你的账号，其他用户无法使用。", install: "配置插件", name: "连接名称", url: "服务地址", @@ -39,8 +47,7 @@ const zh: typeof en = { sign_secret: "机器人加签密钥", webhook_key: "机器人 Webhook 密钥（地址中的 key 参数）", save: "保存配置", - accountHint: - "这里配置的是由管理员管理的部署凭据。支持个人授权的集成，通过其账号流程连接。", + accountHint: "这里的配置仅保存到你的账号，其他用户无法查看或使用。", saved: "配置已保存", unknown: "未检测", required: "需要连接账号", diff --git a/frontend/src/core/capabilities/hooks.ts b/frontend/src/core/capabilities/hooks.ts index 43b2e01d1d..52c92c52d1 100644 --- a/frontend/src/core/capabilities/hooks.ts +++ b/frontend/src/core/capabilities/hooks.ts @@ -1,6 +1,7 @@ import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { fetch } from "@/core/api/fetcher"; +import { useAuth } from "@/core/auth/AuthProvider"; import { getBackendBaseURL } from "@/core/config"; import type { InstallationList, PluginManifest } from "./types"; @@ -30,7 +31,8 @@ export function useCapabilityCatalog() { }); } export function useCapabilityInstallations(adapter: string) { - return useQuery(installationQuery(adapter)); + const { user } = useAuth(); + return useQuery(installationQuery(adapter, user?.id)); } export function useInstallCapability() { const client = useQueryClient(); @@ -43,7 +45,7 @@ export function useInstallCapability() { request("installations", { method: "POST", headers: { "Content-Type": "application/json" }, - body: JSON.stringify(body), + body: JSON.stringify({ ...body, scope: "user" }), }), onSuccess: async () => { await Promise.all([ @@ -55,10 +57,12 @@ export function useInstallCapability() { }); } -export function installationQuery(adapter: string) { +export function installationQuery(adapter: string, userId?: string) { return { - queryKey: ["capabilities", "installations", adapter], + queryKey: ["capabilities", "installations", adapter, userId], queryFn: () => - request(`installations/${encodeURIComponent(adapter)}`), + request( + `installations/${encodeURIComponent(adapter)}?scope=all`, + ), }; } diff --git a/frontend/src/core/mcp/api.ts b/frontend/src/core/mcp/api.ts index b07217acdc..27778000b3 100644 --- a/frontend/src/core/mcp/api.ts +++ b/frontend/src/core/mcp/api.ts @@ -3,6 +3,10 @@ import { getBackendBaseURL } from "@/core/config"; import type { MCPConfig, MCPServerConfig } from "./types"; +export type MCPScope = "deployment" | "user"; +const configPath = (scope: MCPScope) => + scope === "user" ? "/api/mcp/personal/config" : "/api/mcp/config"; + export class MCPConfigRequestError extends Error { readonly status: number; constructor(status: number, message: string) { @@ -25,8 +29,8 @@ async function readErrorDetail( return typeof error.detail === "string" ? error.detail : fallback; } -export async function loadMCPConfig() { - const response = await fetch(`${getBackendBaseURL()}/api/mcp/config`); +export async function loadMCPConfig(scope: MCPScope = "deployment") { + const response = await fetch(`${getBackendBaseURL()}${configPath(scope)}`); if (!response.ok) { throw new MCPConfigRequestError( response.status, @@ -76,27 +80,37 @@ async function mutateMCPServerConfig( return response.json() as Promise; } -export function createMCPServers(servers: Record) { +export function createMCPServers( + servers: Record, + scope: MCPScope = "deployment", +) { return mutateMCPServerConfig( - "/api/mcp/config/servers", + `${configPath(scope)}/servers`, "POST", { mcp_servers: servers }, "Failed to add MCP servers", ); } -export function updateMCPServer(serverName: string, server: MCPServerConfig) { +export function updateMCPServer( + serverName: string, + server: MCPServerConfig, + scope: MCPScope = "deployment", +) { return mutateMCPServerConfig( - "/api/mcp/config/server", + `${configPath(scope)}/server`, "PUT", { server_name: serverName, server }, "Failed to update MCP server", ); } -export function deleteMCPServer(serverName: string) { +export function deleteMCPServer( + serverName: string, + scope: MCPScope = "deployment", +) { return mutateMCPServerConfig( - `/api/mcp/config/servers/${encodeURIComponent(serverName)}`, + `${configPath(scope)}/servers/${encodeURIComponent(serverName)}`, "DELETE", undefined, "Failed to delete MCP server", @@ -106,8 +120,9 @@ export function deleteMCPServer(serverName: string) { export async function updateMCPServerState( serverName: string, enabled: boolean, + scope: MCPScope = "deployment", ) { - const response = await fetch(`${getBackendBaseURL()}/api/mcp/config`, { + const response = await fetch(`${getBackendBaseURL()}${configPath(scope)}`, { method: "PATCH", headers: { "Content-Type": "application/json", diff --git a/frontend/src/core/mcp/hooks.ts b/frontend/src/core/mcp/hooks.ts index f9124c0011..9e14e25649 100644 --- a/frontend/src/core/mcp/hooks.ts +++ b/frontend/src/core/mcp/hooks.ts @@ -6,6 +6,8 @@ import { } from "@tanstack/react-query"; import { toast } from "sonner"; +import { useAuth } from "@/core/auth/AuthProvider"; + import { createMCPServers, deleteMCPServer, @@ -14,12 +16,15 @@ import { updateMCPServer, updateMCPServerState, } from "./api"; +import type { MCPScope } from "./api"; import type { MCPServerConfig } from "./types"; export function useMCPConfig() { + const { user } = useAuth(); const { data, isLoading, error } = useQuery({ - queryKey: ["mcpConfig"], - queryFn: () => loadMCPConfig(), + queryKey: ["mcpConfig", "user", user?.id], + queryFn: () => loadMCPConfig("user"), + enabled: !!user, retry: (count, error) => !(error instanceof MCPConfigRequestError) && count < 3, }); @@ -31,10 +36,13 @@ interface EnableMCPServerVariables { enabled: boolean; } -export function getEnableMCPServerMutationOptions(queryClient: QueryClient) { +export function getEnableMCPServerMutationOptions( + queryClient: QueryClient, + scope: MCPScope = "deployment", +) { return { mutationFn: ({ serverName, enabled }: EnableMCPServerVariables) => - updateMCPServerState(serverName, enabled), + updateMCPServerState(serverName, enabled, scope), onSuccess: async () => { await Promise.all([ queryClient.invalidateQueries({ queryKey: ["mcpConfig"] }), @@ -49,7 +57,7 @@ export function getEnableMCPServerMutationOptions(queryClient: QueryClient) { export function useEnableMCPServer() { const queryClient = useQueryClient(); - return useMutation(getEnableMCPServerMutationOptions(queryClient)); + return useMutation(getEnableMCPServerMutationOptions(queryClient, "user")); } export type MCPServerMutationVariables = @@ -67,16 +75,19 @@ export type MCPServerMutationVariables = serverName: string; }; -export function getMCPServerMutationOptions(queryClient: QueryClient) { +export function getMCPServerMutationOptions( + queryClient: QueryClient, + scope: MCPScope = "deployment", +) { return { mutationFn: (variables: MCPServerMutationVariables) => { switch (variables.operation) { case "create": - return createMCPServers(variables.servers); + return createMCPServers(variables.servers, scope); case "update": - return updateMCPServer(variables.serverName, variables.server); + return updateMCPServer(variables.serverName, variables.server, scope); case "delete": - return deleteMCPServer(variables.serverName); + return deleteMCPServer(variables.serverName, scope); } }, onSuccess: async () => { @@ -93,5 +104,5 @@ export function getMCPServerMutationOptions(queryClient: QueryClient) { export function useMCPServerMutation() { const queryClient = useQueryClient(); - return useMutation(getMCPServerMutationOptions(queryClient)); + return useMutation(getMCPServerMutationOptions(queryClient, "user")); } diff --git a/frontend/tests/e2e/business-plugins.spec.ts b/frontend/tests/e2e/business-plugins.spec.ts index 716d98450c..29d37c2b5f 100644 --- a/frontend/tests/e2e/business-plugins.spec.ts +++ b/frontend/tests/e2e/business-plugins.spec.ts @@ -31,7 +31,7 @@ for (const plugin of [ let submission: Record | undefined; let configured = false; const connectionName = `team-${plugin.id}`; - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: configured @@ -82,6 +82,7 @@ for (const plugin of [ plugin_id: plugin.id, name: connectionName, configuration: plugin.fields, + scope: "user", }); await expect( page.locator("article").filter({ hasText: connectionName }), @@ -113,7 +114,7 @@ test("rejected credentials remain editable and are not marked configured", async ); }); -test("ordinary users see each shared connection once and cannot configure credentials", async ({ +test("ordinary users see shared connections read-only and configure their own credentials", async ({ page, }) => { mockLangGraphAPI(page); @@ -141,8 +142,10 @@ test("ordinary users see each shared connection once and cannot configure creden }; // Both adapters project the same MCP installation, not a second account. for (const adapter of ["mcp", "business"]) { - await page.route(`**/api/capabilities/installations/${adapter}`, (route) => - route.fulfill({ json: { items: [connection], can_manage: false } }), + await page.route( + `**/api/capabilities/installations/${adapter}?scope=all`, + (route) => + route.fulfill({ json: { items: [connection], can_manage: false } }), ); } await page.goto("/workspace/capabilities"); @@ -154,12 +157,41 @@ test("ordinary users see each shared connection once and cannot configure creden await page.evaluate(() => document.dispatchEvent(new Event("visibilitychange")), ); + const platform = page.getByRole("region", { + name: "Platform provided", + exact: true, + }); + const personal = page.getByRole("region", { + name: "My plugins", + exact: true, + }); + await expect( + platform.getByRole("button", { + name: "Setup guide Firecrawl", + exact: true, + }), + ).toBeVisible(); + await expect( + platform.locator("article").filter({ hasText: "Team CRM" }), + ).toHaveCount(1); + await expect( + personal.locator("article").filter({ hasText: "Team CRM" }), + ).toHaveCount(0); + await expect( + personal.getByRole("button", { name: "Add MCP plugin" }), + ).toBeVisible(); + await expect( + personal.locator("article").filter({ hasText: "HubSpot" }), + ).toContainText("Not configured"); await expect( page.locator("article").filter({ hasText: "Team CRM" }), ).toHaveCount(1); await expect( - page.getByRole("button", { name: "Add MCP plugin" }), + page.locator("article").filter({ hasText: "Team CRM" }).getByRole("switch"), ).toHaveCount(0); + await expect( + page.getByRole("button", { name: "Add MCP plugin" }), + ).toHaveCount(1); await page .getByRole("button", { name: "Configure WeCom group notifications", @@ -167,9 +199,9 @@ test("ordinary users see each shared connection once and cannot configure creden }) .click(); const dialog = page.getByRole("dialog"); - await expect(dialog.locator("#plugin-field-webhook_key")).toBeDisabled(); + await expect(dialog.locator("#plugin-field-webhook_key")).toBeEnabled(); await expect(dialog.locator("#plugin-field-webhook_key")).toHaveValue(""); await expect( dialog.getByRole("button", { name: "Save configuration" }), - ).toBeDisabled(); + ).toBeEnabled(); }); diff --git a/frontend/tests/e2e/capability-center.spec.ts b/frontend/tests/e2e/capability-center.spec.ts index 9c0008f3d2..7a7f9f6e57 100644 --- a/frontend/tests/e2e/capability-center.spec.ts +++ b/frontend/tests/e2e/capability-center.spec.ts @@ -53,7 +53,7 @@ const skills = [ async function mockCatalog(page: Page, locale = "en-US") { mockLangGraphAPI(page, { skills, threads: [] }); - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: { @@ -218,7 +218,7 @@ test("MCP access errors preserve the independently available Lark integration", page, }) => { await mockCatalog(page); - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ status: 403, json: { detail: "Admin only" } }), ); await page.goto("/workspace/capabilities"); @@ -268,7 +268,7 @@ test("plugin filters remain usable after an MCP refetch fails", async ({ }) => { await mockCatalog(page); let failRead = false; - await page.route("**/api/mcp/config", async (route) => { + await page.route("**/api/mcp/personal/config", async (route) => { if (route.request().method() === "PATCH") { failRead = true; return route.fulfill({ json: { mcp_servers: {} } }); @@ -323,7 +323,7 @@ test("plugin categories, setup guides, and installed state remain distinct", asy "研发与运维", ]) { await expect( - page.getByRole("heading", { name, exact: true }), + page.getByRole("heading", { name, exact: true }).first(), ).toBeVisible(); } await screenshot(page, "capability-catalog-zh.png"); @@ -356,7 +356,7 @@ test("plugin categories, setup guides, and installed state remain distinct", asy await page.getByRole("tab", { name: "已安装", exact: true }).click(); await expect(page.locator("article")).toHaveCount(0); await expect( - page.getByText("没有找到匹配的内容", { exact: true }), + page.getByText("没有找到匹配的内容", { exact: true }).first(), ).toBeVisible(); await page.getByRole("textbox", { name: "搜索插件名称或用途" }).fill(""); await expect(page.locator("article")).toHaveCount(5); @@ -385,7 +385,7 @@ test("manifest installation saves through the adapter and refreshes the catalog" mockLangGraphAPI(page); let installed: Record | null = null; let submission: Record | null = null; - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: installed ? { "team-code": installed } : {} }, }), @@ -443,7 +443,7 @@ test("agent selection saves explicit plugin IDs and an empty skill list", async agents: [{ name: "analyst", description: "Analyze reports" }], skills, }); - await page.route("**/api/capabilities/installations/mcp", (route) => + await page.route("**/api/capabilities/installations/mcp?scope=all", (route) => route.fulfill({ json: { can_manage: false, diff --git a/frontend/tests/e2e/mcp-settings.spec.ts b/frontend/tests/e2e/mcp-settings.spec.ts index cff1a67178..345fa997cf 100644 --- a/frontend/tests/e2e/mcp-settings.spec.ts +++ b/frontend/tests/e2e/mcp-settings.spec.ts @@ -15,7 +15,7 @@ for (const { viewport, name } of [ await page.setViewportSize(viewport); mockLangGraphAPI(page); const server = { enabled: false, command: "npx", args: ["example"] }; - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: { [name]: server } } }), ); await page.goto("/workspace/capabilities"); @@ -109,7 +109,7 @@ for (const viewport of [ Array.from({ length: 60 }, (_, index) => [`TEST_${index}`, "example"]), ), }; - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: { github: server } } }), ); await page.goto("/workspace/capabilities"); @@ -196,14 +196,14 @@ test.describe("MCP server settings", () => { | { server_name: string; server: (typeof servers)["remote"] } | undefined; - await page.route("**/api/mcp/config", async (route) => { + await page.route("**/api/mcp/personal/config", async (route) => { await route.fulfill({ status: 200, contentType: "application/json", body: JSON.stringify({ mcp_servers: servers }), }); }); - await page.route("**/api/mcp/config/server", async (route) => { + await page.route("**/api/mcp/personal/config/server", async (route) => { if (route.request().method() !== "PUT") { await route.fallback(); return; diff --git a/frontend/tests/e2e/plugin-icons.spec.ts b/frontend/tests/e2e/plugin-icons.spec.ts index e6fba13a6f..95aed85b93 100644 --- a/frontend/tests/e2e/plugin-icons.spec.ts +++ b/frontend/tests/e2e/plugin-icons.spec.ts @@ -48,7 +48,7 @@ test("brand icons load locally for both recommendations and configured MCP serve .context() .addCookies([{ name: "locale", value: "zh-CN", url: baseURL! }]); mockLangGraphAPI(page); - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: { @@ -137,10 +137,10 @@ test("upload previews, cancellation, save, reload, restore default, and create a }, }; let writes = 0; - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: servers } }), ); - await page.route("**/api/mcp/config/server", async (route) => { + await page.route("**/api/mcp/personal/config/server", async (route) => { writes++; const body = route.request().postDataJSON() as { server_name: string; @@ -149,7 +149,7 @@ test("upload previews, cancellation, save, reload, restore default, and create a servers[body.server_name] = body.server; await route.fulfill({ json: { mcp_servers: servers } }); }); - await page.route("**/api/mcp/config/servers", async (route) => { + await page.route("**/api/mcp/personal/config/servers", async (route) => { writes++; const body = route.request().postDataJSON() as { mcp_servers: Record; @@ -254,7 +254,7 @@ test("invalid or oversized uploads do not replace the existing icon", async ({ page, }) => { mockLangGraphAPI(page); - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: { @@ -322,7 +322,7 @@ test("a custom MCP name does not impersonate a catalog brand in rows or editing" page, }) => { mockLangGraphAPI(page); - await page.route("**/api/mcp/config", (route) => + await page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: { diff --git a/frontend/tests/e2e/utils/mock-api.ts b/frontend/tests/e2e/utils/mock-api.ts index 358a03190f..3b530b4a43 100644 --- a/frontend/tests/e2e/utils/mock-api.ts +++ b/frontend/tests/e2e/utils/mock-api.ts @@ -1859,7 +1859,7 @@ export function mockLangGraphAPI(page: Page, options?: MockAPIOptions) { return route.fallback(); }); - void page.route("**/api/mcp/config", (route) => + void page.route("**/api/mcp/personal/config", (route) => route.fulfill({ json: { mcp_servers: {} } }), ); @@ -1877,7 +1877,7 @@ export function mockLangGraphAPI(page: Page, options?: MockAPIOptions) { }), ); void page.route("**/api/capabilities/installations/*", (route) => { - const adapter = route.request().url().split("/").pop(); + const adapter = new URL(route.request().url()).pathname.split("/").pop(); const items = adapter === "lark" ? [ diff --git a/frontend/tests/unit/core/mcp/ownership.dom.test.tsx b/frontend/tests/unit/core/mcp/ownership.dom.test.tsx new file mode 100644 index 0000000000..ece9098b47 --- /dev/null +++ b/frontend/tests/unit/core/mcp/ownership.dom.test.tsx @@ -0,0 +1,66 @@ +import { afterEach, describe, expect, it, rs } from "@rstest/core"; +import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; +import { cleanup, renderHook, waitFor } from "@testing-library/react"; +import type { ReactNode } from "react"; + +const auth = rs.hoisted(() => ({ user: { id: "alice" } })); +rs.mock("@/core/auth/AuthProvider", () => ({ useAuth: () => auth })); +rs.mock("@/core/api/fetcher", () => ({ fetch: rs.fn() })); + +import { fetch } from "@/core/api/fetcher"; +import { useMCPConfig } from "@/core/mcp/hooks"; + +afterEach(cleanup); + +describe("personal MCP ownership", () => { + it("does not display a previous account's late response after switching users", async () => { + auth.user = { id: "alice" }; + let finishAlice!: (response: Response) => void; + rs.mocked(fetch).mockImplementation(async () => { + if (auth.user.id === "alice") { + return new Promise((resolve) => { + finishAlice = resolve; + }); + } + return new Response( + JSON.stringify({ mcp_servers: { "bob-only": { enabled: true } } }), + ); + }); + const client = new QueryClient({ + defaultOptions: { queries: { retry: false } }, + }); + const wrapper = ({ children }: { children: ReactNode }) => ( + {children} + ); + const hook = renderHook(() => useMCPConfig(), { wrapper }); + await waitFor(() => expect(finishAlice).toBeDefined()); + auth.user = { id: "bob" }; + hook.rerender(); + await waitFor(() => + expect(hook.result.current.config?.mcp_servers).toHaveProperty( + "bob-only", + ), + ); + finishAlice( + new Response( + JSON.stringify({ mcp_servers: { "alice-only": { enabled: true } } }), + ), + ); + await waitFor(() => + expect(client.getQueryData(["mcpConfig", "user", "alice"])).toBeDefined(), + ); + expect(hook.result.current.config?.mcp_servers).not.toHaveProperty( + "alice-only", + ); + expect(hook.result.current.config?.mcp_servers).toHaveProperty("bob-only"); + expect( + rs + .mocked(fetch) + .mock.calls.every( + ([url]) => + typeof url === "string" && url.endsWith("/api/mcp/personal/config"), + ), + ).toBe(true); + client.clear(); + }); +});