feat(mcp): per-user credential injection for shared MCP servers (#4868)

* feat(mcp): per-user credential injection for shared MCP servers A single HTTP/SSE MCP server entry can now serve several users, each authenticated to the remote service with their own credential. A server opts in with a user_auth block mapping user ids to credential header values ($ENV_VAR references supported): "user_auth": { "header": "Authorization", "users": { "<user-id>": "$SERVICE_TOKEN_ALICE" } } The built-in user-scoped auth interceptor resolves the authenticated runtime user on every tool call (request runtime -> ambient LangGraph runtime -> auth config -> request-scoped user ContextVar) and rewrites the configured header via request.override(), the same per-call mechanism as the OAuth interceptor. It registers after OAuth in the shared assembly so its per-user value wins the header when a server declares both. The entry's static headers are used only for startup tool discovery. Fail-closed by default: an unmapped user - including the anonymous default-user fallback - or a credential whose env reference resolved empty gets an actionable ToolException instead of another user's credential; on_missing: "passthrough" opts out per server. Combined with the existing per-(user, thread) MCP session scoping this gives credential isolation on shared servers. Gateway API: user_auth.users values are masked in GET responses, and PUT round-trips preserve stored credentials for masked values (same contract as env/headers/oauth secrets); a masked value for a user id not already stored is rejected. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mcp): address review — preserve stored user_auth sub-fields on partial PUT, warn-and-skip stdio, allow extras Review findings from #4868: 1. A partial user_auth payload (e.g. {"enabled": false}) merged to users={} and default on_missing, irreversibly wiping stored credentials on PUT. The merge is now sub-field-aware via model_fields_set — omitted sub-fields carry the stored values, an explicitly sent users map still replaces (so full-round-trip removal works), masked values still swap back for stored credentials. 2. user_auth on a stdio server was a silent no-op: rewritten headers go to call meta, never a transport header, while deny errors still fired. The interceptor builder now warns and skips non-sse/http servers, matching the tool_call_timeout transport-mismatch convention. 3. McpUserScopedAuthConfigResponse now allows extra keys like the harness-side model, and extras survive masking and merge, matching the server-level model_extra handling. Adds four regression tests (partial-PUT preservation, explicit-map replacement, extras round-trip, stdio warn-and-skip). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mcp): reject blank user_auth.header at the gateway A blank header passed the gateway response model, was persisted, then failed the harness-side ExtensionsConfig validator on reload — the PUT returned 500 after the write and every later config load/startup failed until the file was hand-edited. Mirror the harness non-blank validator on McpUserScopedAuthConfigResponse so the PUT fails with 422 before anything is written. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * style: ruff format extensions_config.py Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> * fix(mcp): harden the trust chain and masking around user-scoped credential selection Address review round 4: - Scrub client-supplied user_id from run context/configurable for external callers in inject_authenticated_user_context, before every early return, and restamp only from request.state.user. Now that user_id selects which user's credential user-scoped MCP auth injects, a forged value must not survive any future path that skips the restamp. Internal callers (IM channels, scheduler) keep supplying end-user identity as before (PR #3294 contract). Regression tests pin both the scrub and that a forged body.context.user_id can never resolve as another user through merge + inject ordering. - Include the caller's own resolved user id in the fail-closed deny message so operators can copy the exact users key (it differs by deployment path), and document the key formats in the mcp.mdx doc. - Mask sensitive extra keys inside user_auth on GET like server-level extras, and swap masked sentinels back for stored values on PUT via _merge_extra_value_preserving_masked. - Extract the interceptor wrap loop into compose_tool_interceptors and pin the security property functionally: an OAuth interceptor that actually sets Authorization loses the final header to the per-user credential through the same composition the session-pool path uses. Co-Authored-By: Claude Fable 5 <noreply@anthropic.com> --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
2026-09-09 21:49:37 +00:00 · 2026-08-23 08:16:04 +01:00 · 2026-08-23 08:16:04 +01:00 · 7e95bef2e7
commit 7e95bef2e7
parent 236a068e77
10 changed files with 768 additions and 11 deletions
--- a/backend/app/gateway/routers/mcp.py
+++ b/backend/app/gateway/routers/mcp.py
@ -7,7 +7,7 @@ from pathlib import Path
 from typing import Any, Literal, NamedTuple

 from fastapi import APIRouter, HTTPException, Request, status
-from pydantic import BaseModel, ConfigDict, Field, model_validator
+from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator

 from app.gateway.deps import require_admin_user
 from deerflow.config.extensions_config import (
@ -347,6 +347,29 @@ _CODE_INJECTING_ENV_VARS = frozenset(
 )


+class McpUserScopedAuthConfigResponse(BaseModel):
+    """Per-user credential injection configuration for an MCP server."""
+
+    enabled: bool = Field(default=True, description="Whether user-scoped credential injection is enabled")
+    header: str = Field(default="Authorization", description="HTTP header to set with the resolved user credential")
+    users: dict[str, str] = Field(default_factory=dict, description="Map of DeerFlow user id to credential header value")
+    on_missing: Literal["deny", "passthrough"] = Field(default="deny", description="Behavior when the calling user has no mapped credential")
+    # Mirror the harness-side McpUserScopedAuthConfig (extra="allow"): without
+    # this, an operator's unknown key inside user_auth would be silently
+    # stripped by the next admin PUT, while server-level extras are preserved.
+    model_config = ConfigDict(extra="allow")
+
+    # Mirror the harness-side non-blank check: a blank header accepted here
+    # would be persisted, then fail ExtensionsConfig validation on reload —
+    # wedging every subsequent config load until the file is hand-edited.
+    @field_validator("header")
+    @classmethod
+    def _validate_header_not_blank(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("user_auth.header must not be empty")
+        return value
+
+
 class McpOAuthConfigResponse(BaseModel):
    """OAuth configuration for an MCP server."""

@ -377,6 +400,7 @@ class McpServerConfigResponse(BaseModel):
    url: str | None = Field(default=None, description="URL of the MCP server (for sse or http type)")
    headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)")
    oauth: McpOAuthConfigResponse | None = Field(default=None, description="OAuth configuration for MCP HTTP/SSE servers")
+    user_auth: McpUserScopedAuthConfigResponse | None = Field(default=None, description="Per-user credential injection for MCP HTTP/SSE servers")
    description: str = Field(default="", description="Human-readable description of what this MCP server provides")
    routing: McpRoutingConfig = Field(default_factory=McpRoutingConfig, description="Soft routing hints for tools from this MCP server")
    tools: dict[str, McpToolOverride] = Field(default_factory=dict, description="Per-original-tool MCP configuration overrides")
@ -653,12 +677,21 @@ def _mask_server_config(server: McpServerConfigResponse) -> McpServerConfigRespo
                "refresh_token": None,
            }
        )
+    masked_user_auth = None
+    if server.user_auth is not None:
+        # Extras inside user_auth get the same sensitive-key masking as
+        # server-level extras: they round-trip through PUT (extra="allow"), so
+        # an operator-stored secret-bearing key must not come back in
+        # cleartext from GET while the identical key at server level is masked.
+        masked_ua_extra = {key: _MASKED_VALUE if _is_sensitive_extra_key(key) else _mask_sensitive_extra_value(value) for key, value in (server.user_auth.model_extra or {}).items()}
+        masked_user_auth = server.user_auth.model_copy(update={"users": {k: _MASKED_VALUE for k in server.user_auth.users}, **masked_ua_extra})
    masked_extra = {key: _MASKED_VALUE if _is_sensitive_extra_key(key) else _mask_sensitive_extra_value(value) for key, value in (server.model_extra or {}).items()}
    return server.model_copy(
        update={
            "env": masked_env,
            "headers": masked_headers,
            "oauth": masked_oauth,
+            "user_auth": masked_user_auth,
            **masked_extra,
        }
    )
@ -720,11 +753,62 @@ def _merge_preserving_secrets(
                "refresh_token": merged_refresh_token,
            }
        )
+    merged_user_auth = incoming.user_auth
+    if incoming.user_auth is not None:
+        # Sub-field-aware merge: a partial user_auth payload (e.g. just
+        # {"enabled": false}) must not wipe the stored credential map or reset
+        # other stored sub-fields. Only sub-fields the request explicitly set
+        # replace stored values; the rest carry over — the same contract the
+        # block-level `model_fields_set` check below applies one level up.
+        incoming_ua = incoming.user_auth
+        base = existing.user_auth
+        set_fields = incoming_ua.model_fields_set
+        effective: dict[str, Any] = {}
+        if base is not None:
+            effective.update({name: getattr(base, name) for name in ("enabled", "header", "users", "on_missing")})
+            effective.update(base.model_extra or {})
+        for name in ("enabled", "header", "on_missing"):
+            if name in set_fields:
+                effective[name] = getattr(incoming_ua, name)
+        # Extras are masked by GET (see _mask_server_config), so a round-trip
+        # PUT must swap masked sentinel values back for the stored ones —
+        # the same contract server-level extras get below.
+        base_extra = (base.model_extra or {}) if base is not None else {}
+        for key, value in (incoming_ua.model_extra or {}).items():
+            effective[key] = _merge_extra_value_preserving_masked(
+                key,
+                value,
+                base_extra.get(key),
+                existing_present=key in base_extra,
+            )
+        if "users" in set_fields:
+            # An explicitly sent map replaces the stored one (so a full
+            # round-trip can remove a user), with masked values swapped back
+            # for the stored credentials.
+            existing_users = base.users if base is not None else {}
+            merged_users = {}
+            for k, v in incoming_ua.users.items():
+                if v == _MASKED_VALUE:
+                    if k in existing_users:
+                        merged_users[k] = existing_users[k]
+                    else:
+                        raise HTTPException(
+                            status_code=400,
+                            detail=f"Cannot set user_auth credential for '{k}' to masked value '***'; provide a real value.",
+                        )
+                else:
+                    merged_users[k] = v
+            effective["users"] = merged_users
+        merged_user_auth = McpUserScopedAuthConfigResponse(**effective)
+
    update = {
        "env": merged_env,
        "headers": merged_headers,
        "oauth": merged_oauth,
+        "user_auth": merged_user_auth,
    }
+    if "user_auth" not in incoming.model_fields_set:
+        update["user_auth"] = existing.user_auth
    if "routing" not in incoming.model_fields_set:
        update["routing"] = existing.routing
    if "tools" not in incoming.model_fields_set:
--- a/backend/app/gateway/services.py
+++ b/backend/app/gateway/services.py
@ -499,6 +499,18 @@ def inject_authenticated_user_context(
        for key in _SERVER_OWNED_AUTHZ_CONTEXT_KEYS:
            configurable.pop(key, None)
    auth_source = getattr(getattr(request, "state", None), "auth_source", None)
+    # ``user_id`` is server-owned for EXTERNAL callers: it now selects which
+    # user's credential user-scoped MCP auth injects, so a client-forged value
+    # must never survive any early return below — scrub it here and restamp it
+    # only from ``request.state.user``. Internal callers (IM channels, the
+    # scheduler) are the deliberate exception: they authenticate their own end
+    # users and supply that identity in run context (PR #3294), which the
+    # internal-role branch below preserves.
+    user = getattr(getattr(request, "state", None), "user", None)
+    if auth_source != AUTH_SOURCE_INTERNAL and getattr(user, "system_role", None) != INTERNAL_SYSTEM_ROLE:
+        runtime_context.pop("user_id", None)
+        if isinstance(configurable, dict):
+            configurable.pop("user_id", None)
    runtime_context["is_internal"] = auth_source == AUTH_SOURCE_INTERNAL
    if auth_source == AUTH_SOURCE_INTERNAL and request_context is not None:
        channel_user_id = request_context.get("channel_user_id")
--- a/backend/packages/harness/deerflow/config/extensions_config.py
+++ b/backend/packages/harness/deerflow/config/extensions_config.py
@ -98,6 +98,40 @@ class McpTaskToolsetConfig(BaseModel):
        return value


+class McpUserScopedAuthConfig(BaseModel):
+    """Per-user credential injection for a shared MCP server (HTTP/SSE transports).
+
+    Maps DeerFlow user ids to credential header values so that one configured
+    MCP server can serve several users, each authenticated to the remote
+    service with their own credential. The credential for the authenticated
+    user is injected into every tool call by the built-in user-scoped auth
+    interceptor; the server entry's static ``headers`` are only used for
+    startup tool discovery.
+
+    Values support the same ``$ENV_VAR`` resolution as the rest of this file,
+    so raw secrets can stay in the process environment.
+    """
+
+    enabled: bool = Field(default=True, description="Whether user-scoped credential injection is enabled")
+    header: str = Field(default="Authorization", description="HTTP header to set with the resolved user credential")
+    users: dict[str, str] = Field(
+        default_factory=dict,
+        description="Map of DeerFlow user id to full credential header value (e.g. 'Bearer <token>'); values support $ENV_VAR references",
+    )
+    on_missing: Literal["deny", "passthrough"] = Field(
+        default="deny",
+        description=("Behavior when the calling user has no mapped credential (or the mapped value resolved empty): 'deny' fails the tool call with an actionable error; 'passthrough' forwards the request with the server's static headers"),
+    )
+    model_config = ConfigDict(extra="allow")
+
+    @field_validator("header")
+    @classmethod
+    def _validate_header_not_blank(cls, value: str) -> str:
+        if not value.strip():
+            raise ValueError("user_auth.header must not be empty")
+        return value
+
+
 class McpOAuthConfig(BaseModel):
    """OAuth configuration for an MCP server (HTTP/SSE transports)."""

@ -132,6 +166,10 @@ class McpServerConfig(BaseModel):
    url: str | None = Field(default=None, description="URL of the MCP server (for sse or http type)")
    headers: dict[str, str] = Field(default_factory=dict, description="HTTP headers to send (for sse or http type)")
    oauth: McpOAuthConfig | None = Field(default=None, description="OAuth configuration (for sse or http type)")
+    user_auth: McpUserScopedAuthConfig | None = Field(
+        default=None,
+        description="Per-user credential injection (for sse or http type): map DeerFlow user ids to per-user credential header values",
+    )
    description: str = Field(default="", description="Human-readable description of what this MCP server provides")
    routing: McpRoutingConfig = Field(default_factory=McpRoutingConfig, description="Soft routing hints for tools from this MCP server")
    tools: dict[str, McpToolOverride] = Field(default_factory=dict, description="Per-original-tool MCP configuration overrides")
--- a/backend/packages/harness/deerflow/mcp/AGENTS.md
+++ b/backend/packages/harness/deerflow/mcp/AGENTS.md
@ -10,6 +10,7 @@
 - **Transports**: stdio (command-based), SSE, HTTP
 - **Per-server tool-name prefixing**: `mcpServers.<server>.tool_name_prefix` defaults to `true`, preserving the collision-safe `<server_name>_` prefix. Servers whose tools already carry a stable namespace may set it to `false`; discovery then calls `langchain_mcp_adapters.tools.load_mcp_tools` with that server's flag. Source routing and stdio session-pool wrapping are based on the producing server and transport, never on whether the visible tool name starts with the server prefix.
 - **OAuth (HTTP/SSE)**: Supports token endpoint flows (`client_credentials`, `refresh_token`) with automatic token refresh + Authorization header injection
+- **Per-user credentials (HTTP/SSE)**: `mcpServers.<server>.user_auth` maps DeerFlow user ids to credential header values (`$ENV_VAR` references supported). `mcp/user_scoped_auth.py::build_user_scoped_auth_interceptor` rewrites the configured header on every tool call from the authenticated runtime user (registered after OAuth in `mcp/interceptors.py`, so its per-user value wins the header for servers declaring both). Fail-closed: an unmapped user or an empty resolved credential raises a `ToolException` unless `on_missing: "passthrough"` is set. The server's static `headers` serve startup tool discovery only. Gateway GET masks `user_auth.users` values; PUT round-trips masked values by preserving stored credentials.
 - **Routing hints**: `extensions_config.json -> mcpServers.<server>.routing` and
  `tools.<original_tool_name>.routing` are soft preference metadata. The effective
  routing is resolved while `mcp/tools.py::get_mcp_tools()` still has both
--- a/backend/packages/harness/deerflow/mcp/interceptors.py
+++ b/backend/packages/harness/deerflow/mcp/interceptors.py
@ -7,6 +7,7 @@ from typing import Any

 from deerflow.config.extensions_config import ExtensionsConfig
 from deerflow.mcp.oauth import build_oauth_tool_interceptor
+from deerflow.mcp.user_scoped_auth import build_user_scoped_auth_interceptor
 from deerflow.reflection import resolve_variable

 logger = logging.getLogger(__name__)
@ -16,15 +17,23 @@ def build_mcp_tool_interceptors(
    extensions_config: ExtensionsConfig,
    *,
    oauth_builder: Any = build_oauth_tool_interceptor,
+    user_auth_builder: Any = build_user_scoped_auth_interceptor,
    resolver: Any = resolve_variable,
    target_logger: logging.Logger = logger,
 ) -> list[Any]:
-    """Build OAuth followed by configured custom MCP interceptors."""
+    """Build OAuth, user-scoped auth, then configured custom MCP interceptors."""
    interceptors: list[Any] = []
    oauth_interceptor = oauth_builder(extensions_config)
    if oauth_interceptor is not None:
        interceptors.append(oauth_interceptor)

+    # After OAuth so a server declaring both gets the per-user credential:
+    # interceptors wrap outermost-first, so the later-registered user-scoped
+    # override runs closer to the transport and wins the final header value.
+    user_auth_interceptor = user_auth_builder(extensions_config)
+    if user_auth_interceptor is not None:
+        interceptors.append(user_auth_interceptor)
+
    raw_paths = (extensions_config.model_extra or {}).get("mcpInterceptors")
    if isinstance(raw_paths, str):
        raw_paths = [raw_paths]
@ -55,3 +64,23 @@ def build_mcp_tool_interceptors(
                exc_info=True,
            )
    return interceptors
+
+
+def compose_tool_interceptors(interceptors: list[Any], base_handler: Any) -> Any:
+    """Compose interceptors onion-style around ``base_handler``: first = outermost.
+
+    The later-registered interceptor runs closer to the transport, so its
+    header writes win over earlier ones — the property user-scoped auth relies
+    on to override an OAuth-injected credential. This is the single wrap
+    convention; the session-pool tool path composes through here so tests that
+    pin the override property exercise the production composition.
+    """
+    handler = base_handler
+    for interceptor in reversed(interceptors):
+        outer = handler
+
+        async def wrapped(req: Any, _i: Any = interceptor, _h: Any = outer) -> Any:
+            return await _i(req, _h)
+
+        handler = wrapped
+    return handler
--- a/backend/packages/harness/deerflow/mcp/tools.py
+++ b/backend/packages/harness/deerflow/mcp/tools.py
@ -18,7 +18,7 @@ from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig,
 from deerflow.config.paths import VIRTUAL_PATH_PREFIX, Paths, get_paths
 from deerflow.constants import DEFAULT_MCP_SESSION_INIT_TIMEOUT, MCP_TMP_SUBDIR
 from deerflow.mcp.client import build_servers_config
-from deerflow.mcp.interceptors import build_mcp_tool_interceptors
+from deerflow.mcp.interceptors import build_mcp_tool_interceptors, compose_tool_interceptors
 from deerflow.mcp.oauth import build_oauth_tool_interceptor, get_initial_oauth_headers
 from deerflow.mcp.session_pool import get_session_pool
 from deerflow.mcp.tasks import ORDINARY_MCP_TASK_DRIVER, TaskSubmitRequest
@ -562,14 +562,7 @@ def _make_session_pool_tool(
                    **kwargs,
                )

-            handler = base_handler
-            for interceptor in reversed(tool_interceptors):
-                outer = handler
-
-                async def wrapped(req: Any, _i: Any = interceptor, _h: Any = outer) -> Any:
-                    return await _i(req, _h)
-
-                handler = wrapped
+            handler = compose_tool_interceptors(tool_interceptors, base_handler)

            request = MCPToolCallRequest(
                name=original_name,
--- a/backend/packages/harness/deerflow/mcp/user_scoped_auth.py
+++ b/backend/packages/harness/deerflow/mcp/user_scoped_auth.py
@ -0,0 +1,117 @@
+"""Per-user credential injection for shared MCP servers.
+
+One configured HTTP/SSE MCP server can serve several DeerFlow users, each
+authenticated to the remote service with their own credential. A server opts in
+by declaring a ``user_auth`` block (:class:`McpUserScopedAuthConfig`) mapping
+DeerFlow user ids to credential header values. On every tool call the
+interceptor resolves the authenticated user and rewrites the configured header
+via ``request.override(headers=...)`` — the same per-call mechanism the OAuth
+interceptor uses.
+
+The server entry's static ``headers`` are used only for startup tool discovery
+(``tools/list``); they never authenticate a user's tool call when ``user_auth``
+is enabled for that server, except under an explicit ``on_missing:
+"passthrough"`` opt-out.
+
+Fail-closed by default: an unmapped user (including the anonymous
+``DEFAULT_USER_ID`` fallback), or a mapped credential whose ``$ENV_VAR``
+reference resolved to an empty string, gets an actionable ``ToolException``
+instead of another user's credential or the discovery credential.
+"""
+
+from __future__ import annotations
+
+import logging
+from typing import Any
+
+from langchain_core.tools import ToolException
+
+from deerflow.config.extensions_config import ExtensionsConfig, McpUserScopedAuthConfig
+from deerflow.runtime.user_context import resolve_runtime_user_id
+
+logger = logging.getLogger(__name__)
+
+
+def _current_runtime() -> Any | None:
+    """Best-effort access to the LangGraph runtime for the current tool call.
+
+    ``get_runtime()`` raises outside a runtime context (embedded clients, unit
+    tests, discovery paths); ``resolve_runtime_user_id`` accepts ``None`` and
+    falls back to LangGraph auth config and the request-scoped user
+    ContextVar, so failures here reduce accuracy but never crash the call.
+    """
+    try:
+        from langgraph.runtime import get_runtime
+
+        return get_runtime()
+    except Exception:
+        return None
+
+
+def build_user_scoped_auth_interceptor(extensions_config: ExtensionsConfig) -> Any | None:
+    """Build a tool interceptor injecting per-user credentials, or ``None``.
+
+    Returns ``None`` when no enabled server declares an enabled ``user_auth``
+    block, so callers can skip registration entirely (mirrors
+    ``build_oauth_tool_interceptor``).
+    """
+    user_auth_by_server: dict[str, McpUserScopedAuthConfig] = {}
+    for server_name, server_config in extensions_config.get_enabled_mcp_servers().items():
+        if server_config.user_auth is None or not server_config.user_auth.enabled:
+            continue
+        if server_config.type not in ("sse", "http"):
+            # A stdio server has no HTTP headers: the pooled stdio path forwards
+            # rewritten headers as call meta, never a transport header, so the
+            # credential would go nowhere while deny errors still fired for
+            # unmapped users. Warn-and-skip matches the existing convention for
+            # transport/config mismatches (e.g. tool_call_timeout on non-stdio).
+            logger.warning(
+                "MCP server '%s' declares user_auth but uses the '%s' transport; user-scoped credentials only apply to 'sse'/'http' servers — ignoring user_auth for this server",
+                server_name,
+                server_config.type,
+            )
+            continue
+        user_auth_by_server[server_name] = server_config.user_auth
+
+    if not user_auth_by_server:
+        return None
+
+    async def user_scoped_auth_interceptor(request: Any, handler: Any) -> Any:
+        user_auth = user_auth_by_server.get(request.server_name)
+        if user_auth is None:
+            return await handler(request)
+
+        # Prefer the runtime attached to the request (set by the adapter when
+        # the call originates inside a graph); fall back to the ambient
+        # LangGraph runtime, then to resolve_runtime_user_id's own chain
+        # (LangGraph auth config → request-scoped user ContextVar → default).
+        runtime = getattr(request, "runtime", None)
+        if runtime is None:
+            runtime = _current_runtime()
+        user_id = resolve_runtime_user_id(runtime)
+        # Empty string covers a `$ENV_VAR` reference whose variable was unset:
+        # ExtensionsConfig.resolve_env_variables stores "" for those, and an
+        # empty credential must fail closed rather than send an empty header.
+        credential = user_auth.users.get(user_id, "")
+        if not credential:
+            if user_auth.on_missing == "passthrough":
+                return await handler(request)
+            logger.warning(
+                "Denied MCP tool call to server '%s': no user-scoped credential for user '%s'",
+                request.server_name,
+                user_id,
+            )
+            # The resolved id is included so the operator can copy the exact
+            # ``users`` key: it differs by deployment path (a safe-slug like
+            # ``alice-example-com-ab12cd34`` via LangGraph auth, a raw user
+            # UUID via the embedded Gateway). It is the caller's own id, so
+            # surfacing it leaks nothing across users.
+            raise ToolException(
+                f"No credential is configured for your account (user id '{user_id}') on MCP server '{request.server_name}'. Ask the operator to add this exact id to that server's user_auth.users map (or set its environment variable)."
+            )
+
+        updated_headers = dict(request.headers or {})
+        updated_headers[user_auth.header] = credential
+        return await handler(request.override(headers=updated_headers))
+
+    return user_scoped_auth_interceptor
--- a/backend/tests/test_gateway_services.py
+++ b/backend/tests/test_gateway_services.py
@ -3015,3 +3015,40 @@ async def test_start_run_rejects_invalid_thread_id_before_resolving_dependencies

    assert exc_info.value.status_code == 422
    assert "Invalid thread_id" in exc_info.value.detail
+
+
+def test_client_forged_user_id_is_scrubbed_for_external_callers():
+    """user_id now selects which credential user-scoped MCP auth injects, so a
+    client-forged value must never survive merge + inject on any external path
+    — including ones that end in an early return (no authenticated user)."""
+    from types import SimpleNamespace
+
+    from app.gateway.services import build_run_config, inject_authenticated_user_context, merge_run_context_overrides
+
+    # Forged via body.config (copied verbatim) AND body.context (merged).
+    config = build_run_config("thread-1", {"context": {"user_id": "victim"}, "configurable": {"user_id": "victim"}}, None)
+    merge_run_context_overrides(config, {"user_id": "victim"})
+
+    # External caller with no authenticated user: scrub, never restamp.
+    request = SimpleNamespace(state=SimpleNamespace(user=None, auth_source=None))
+    inject_authenticated_user_context(config, request)
+    assert "user_id" not in config["context"]
+    assert "user_id" not in config["configurable"]
+
+
+def test_client_forged_user_id_never_selects_another_users_credential():
+    """End-to-end pin through merge + inject ordering: the id user-scoped MCP
+    auth resolves from runtime context is the authenticated user, regardless of
+    what the client put in body.context/config."""
+    from types import SimpleNamespace
+
+    from app.gateway.services import build_run_config, inject_authenticated_user_context, merge_run_context_overrides
+    from deerflow.runtime.user_context import resolve_runtime_user_id
+
+    config = build_run_config("thread-1", {"context": {"user_id": "victim"}}, None)
+    merge_run_context_overrides(config, {"user_id": "victim"})
+    request = SimpleNamespace(state=SimpleNamespace(user=SimpleNamespace(id="attacker-own-id", system_role=None, oauth_provider=None, oauth_id=None), auth_source=None))
+    inject_authenticated_user_context(config, request)
+
+    runtime = SimpleNamespace(server_info=None, context=config["context"])
+    assert resolve_runtime_user_id(runtime) == "attacker-own-id"
--- a/backend/tests/test_mcp_user_scoped_auth.py
+++ b/backend/tests/test_mcp_user_scoped_auth.py
@ -0,0 +1,397 @@
+"""Tests for per-user credential injection on shared MCP servers."""
+
+import asyncio
+from types import SimpleNamespace
+from unittest.mock import AsyncMock, patch
+
+import pytest
+from langchain_core.tools import ToolException
+from langchain_mcp_adapters.interceptors import MCPToolCallRequest
+
+from deerflow.config.extensions_config import (
+    ExtensionsConfig,
+    McpServerConfig,
+    McpUserScopedAuthConfig,
+)
+from deerflow.mcp.interceptors import build_mcp_tool_interceptors
+from deerflow.mcp.user_scoped_auth import build_user_scoped_auth_interceptor
+
+
+def _config(**user_auth_kwargs) -> ExtensionsConfig:
+    return ExtensionsConfig(
+        mcp_servers={
+            "shared-http": McpServerConfig(
+                enabled=True,
+                type="http",
+                url="https://mcp.example.com/mcp",
+                headers={"Authorization": "Bearer discovery-token"},
+                user_auth=McpUserScopedAuthConfig(**user_auth_kwargs),
+            ),
+            "other": McpServerConfig(enabled=True, type="http", url="https://other.example.com/mcp"),
+        },
+        skills={},
+    )
+
+
+def _request(server_name: str = "shared-http", headers: dict | None = None, runtime: object | None = None) -> MCPToolCallRequest:
+    return MCPToolCallRequest(
+        name="act",
+        args={},
+        server_name=server_name,
+        headers=headers,
+        runtime=runtime,
+    )
+
+
+def _runtime_for_user(user_id: str) -> object:
+    return SimpleNamespace(server_info=None, context={"user_id": user_id})
+
+
+async def _echo_handler(request: MCPToolCallRequest) -> MCPToolCallRequest:
+    return request
+
+
+def test_no_user_auth_servers_returns_none():
+    config = ExtensionsConfig(
+        mcp_servers={"plain": McpServerConfig(enabled=True, type="http", url="https://x.example.com")},
+        skills={},
+    )
+    assert build_user_scoped_auth_interceptor(config) is None
+
+
+def test_disabled_user_auth_returns_none():
+    config = _config(users={"u1": "Bearer t1"}, enabled=False)
+    assert build_user_scoped_auth_interceptor(config) is None
+
+
+def test_disabled_server_is_ignored():
+    config = _config(users={"u1": "Bearer t1"})
+    config.mcp_servers["shared-http"].enabled = False
+    assert build_user_scoped_auth_interceptor(config) is None
+
+
+def test_mapped_user_gets_own_credential():
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1", "u2": "Bearer t2"}))
+    result = asyncio.run(interceptor(_request(headers={"Authorization": "Bearer discovery-token"}, runtime=_runtime_for_user("u2")), _echo_handler))
+    assert result.headers["Authorization"] == "Bearer t2"
+
+
+def test_custom_header_and_other_headers_preserved():
+    interceptor = build_user_scoped_auth_interceptor(_config(header="X-Api-Key", users={"u1": "k1"}))
+    result = asyncio.run(interceptor(_request(headers={"Accept": "application/json"}, runtime=_runtime_for_user("u1")), _echo_handler))
+    assert result.headers == {"Accept": "application/json", "X-Api-Key": "k1"}
+
+
+def test_other_server_passes_through_untouched():
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"}))
+    request = _request(server_name="other", headers={"Authorization": "Bearer static"}, runtime=_runtime_for_user("u1"))
+    result = asyncio.run(interceptor(request, _echo_handler))
+    assert result is request
+
+
+def test_unmapped_user_denied_without_calling_handler():
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"}))
+    handler = AsyncMock()
+    with pytest.raises(ToolException, match="No credential is configured"):
+        asyncio.run(interceptor(_request(runtime=_runtime_for_user("stranger")), handler))
+    handler.assert_not_awaited()
+
+
+def test_empty_resolved_credential_is_denied():
+    """An unset $ENV_VAR reference resolves to "" and must fail closed."""
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": ""}))
+    with pytest.raises(ToolException, match="No credential is configured"):
+        asyncio.run(interceptor(_request(runtime=_runtime_for_user("u1")), AsyncMock()))
+
+
+def test_on_missing_passthrough_keeps_static_headers():
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"}, on_missing="passthrough"))
+    request = _request(headers={"Authorization": "Bearer discovery-token"}, runtime=_runtime_for_user("stranger"))
+    result = asyncio.run(interceptor(request, _echo_handler))
+    assert result.headers["Authorization"] == "Bearer discovery-token"
+
+
+def test_default_user_fallback_is_denied_when_unmapped():
+    """Without any resolvable identity the DEFAULT_USER_ID fallback must not inherit a credential."""
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"}))
+    with patch("deerflow.mcp.user_scoped_auth._current_runtime", return_value=None), pytest.raises(ToolException):
+        asyncio.run(interceptor(_request(runtime=None), AsyncMock()))
+
+
+def test_env_var_reference_resolution(tmp_path, monkeypatch):
+    monkeypatch.setenv("TEST_USER_CRED", "Bearer from-env")
+    config_file = tmp_path / "extensions_config.json"
+    config_file.write_text(
+        """
+        {
+          "mcpServers": {
+            "shared-http": {
+              "enabled": true,
+              "type": "http",
+              "url": "https://mcp.example.com/mcp",
+              "user_auth": {"users": {"u1": "$TEST_USER_CRED", "u2": "$TEST_USER_CRED_UNSET"}}
+            }
+          }
+        }
+        """
+    )
+    config = ExtensionsConfig.from_file(str(config_file))
+    user_auth = config.mcp_servers["shared-http"].user_auth
+    assert user_auth.users["u1"] == "Bearer from-env"
+    assert user_auth.users["u2"] == ""
+
+
+def test_registered_after_oauth_in_shared_assembly():
+    config = _config(users={"u1": "Bearer t1"})
+
+    async def oauth(request, handler):  # pragma: no cover - identity only
+        return await handler(request)
+
+    interceptors = build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: oauth)
+    assert len(interceptors) == 2
+    assert interceptors[0] is oauth
+    assert interceptors[1].__name__ == "user_scoped_auth_interceptor"
+
+
+def test_shared_assembly_skips_when_no_user_auth():
+    config = ExtensionsConfig(
+        mcp_servers={"plain": McpServerConfig(enabled=True, type="http", url="https://x.example.com")},
+        skills={},
+    )
+    interceptors = build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: None)
+    assert interceptors == []
+
+
+def test_gateway_masks_user_auth_credentials():
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _mask_server_config,
+    )
+
+    server = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}),
+    )
+    masked = _mask_server_config(server)
+    assert masked.user_auth.users == {"u1": "***"}
+    assert masked.user_auth.header == "Authorization"
+
+
+def test_gateway_merge_preserves_masked_user_auth_values():
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _merge_preserving_secrets,
+    )
+
+    existing = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}),
+    )
+    incoming = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "***", "u2": "Bearer new-secret"}),
+    )
+    merged = _merge_preserving_secrets(incoming, existing)
+    assert merged.user_auth.users == {"u1": "Bearer real-secret", "u2": "Bearer new-secret"}
+
+
+def test_gateway_merge_rejects_masked_value_for_new_user():
+    from fastapi import HTTPException
+
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _merge_preserving_secrets,
+    )
+
+    existing = McpServerConfigResponse(type="http", url="https://mcp.example.com/mcp")
+    incoming = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"new-user": "***"}),
+    )
+    with pytest.raises(HTTPException):
+        _merge_preserving_secrets(incoming, existing)
+
+
+def test_gateway_merge_preserves_user_auth_when_field_omitted():
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _merge_preserving_secrets,
+    )
+
+    existing = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}),
+    )
+    incoming = McpServerConfigResponse(type="http", url="https://mcp.example.com/mcp")
+    merged = _merge_preserving_secrets(incoming, existing)
+    assert merged.user_auth is not None
+    assert merged.user_auth.users == {"u1": "Bearer real-secret"}
+
+
+def test_partial_user_auth_put_preserves_stored_subfields():
+    """A payload like {"enabled": false} must not wipe users or reset on_missing."""
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _merge_preserving_secrets,
+    )
+
+    existing = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer real-secret"}, on_missing="passthrough", header="X-Api-Key"),
+    )
+    incoming = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(enabled=False),
+    )
+    merged = _merge_preserving_secrets(incoming, existing)
+    assert merged.user_auth.enabled is False
+    assert merged.user_auth.users == {"u1": "Bearer real-secret"}
+    assert merged.user_auth.on_missing == "passthrough"
+    assert merged.user_auth.header == "X-Api-Key"
+
+
+def test_explicit_users_map_still_replaces_and_can_remove():
+    """An explicitly sent map replaces the stored one, so removal via full round-trip works."""
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _merge_preserving_secrets,
+    )
+
+    existing = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer s1", "u2": "Bearer s2"}),
+    )
+    incoming = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "***"}),
+    )
+    merged = _merge_preserving_secrets(incoming, existing)
+    assert merged.user_auth.users == {"u1": "Bearer s1"}  # u2 removed, u1 preserved through mask
+
+
+def test_user_auth_extra_keys_survive_parse_mask_and_merge():
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _mask_server_config,
+        _merge_preserving_secrets,
+    )
+
+    ua = McpUserScopedAuthConfigResponse(**{"users": {"u1": "Bearer s"}, "custom_note": "keep-me"})
+    assert (ua.model_extra or {}).get("custom_note") == "keep-me"
+    server = McpServerConfigResponse(type="http", url="https://x", user_auth=ua)
+    masked = _mask_server_config(server)
+    assert (masked.user_auth.model_extra or {}).get("custom_note") == "keep-me"
+    merged = _merge_preserving_secrets(
+        McpServerConfigResponse(type="http", url="https://x", user_auth=McpUserScopedAuthConfigResponse(enabled=False)),
+        server,
+    )
+    assert (merged.user_auth.model_extra or {}).get("custom_note") == "keep-me"
+
+
+def test_stdio_server_user_auth_is_skipped_with_warning(caplog):
+    import logging
+
+    config = ExtensionsConfig(
+        mcp_servers={
+            "local-stdio": McpServerConfig(
+                enabled=True,
+                type="stdio",
+                command="npx",
+                args=["-y", "some-server"],
+                user_auth=McpUserScopedAuthConfig(users={"u1": "Bearer t1"}),
+            ),
+        },
+        skills={},
+    )
+    with caplog.at_level(logging.WARNING, logger="deerflow.mcp.user_scoped_auth"):
+        interceptor = build_user_scoped_auth_interceptor(config)
+    assert interceptor is None  # no eligible servers -> nothing registered, no deny errors
+    assert any("user_auth" in r.message and "stdio" in r.message for r in caplog.records)
+
+
+def test_gateway_rejects_blank_user_auth_header():
+    """A blank header must be rejected at the gateway, not persisted and left to
+    wedge extensions_config.json on reload (harness-side validator would raise)."""
+    import pydantic
+    import pytest
+
+    from app.gateway.routers.mcp import McpUserScopedAuthConfigResponse
+
+    for blank in ("", "   ", "\t"):
+        with pytest.raises(pydantic.ValidationError, match="must not be empty"):
+            McpUserScopedAuthConfigResponse(header=blank)
+    # Non-blank still fine, and default untouched.
+    assert McpUserScopedAuthConfigResponse(header="X-Api-Key").header == "X-Api-Key"
+    assert McpUserScopedAuthConfigResponse().header == "Authorization"
+
+
+def test_deny_error_includes_the_callers_resolved_user_id():
+    """The users key format differs by deployment path; the fail-closed error
+    must show the caller's own resolved id so the operator can copy the exact key."""
+    interceptor = build_user_scoped_auth_interceptor(_config(users={"u1": "Bearer t1"}))
+    with pytest.raises(ToolException, match="user id 'stranger-uuid'"):
+        asyncio.run(interceptor(_request(runtime=_runtime_for_user("stranger-uuid")), AsyncMock()))
+
+
+def test_user_credential_wins_over_oauth_set_header_through_real_composition():
+    """Pin the wrap-order property functionally, not just list order: an OAuth
+    interceptor that actually sets Authorization must lose the final header to
+    the per-user credential, through the same composition the session-pool
+    tool path uses."""
+    from deerflow.mcp.interceptors import compose_tool_interceptors
+
+    config = _config(users={"u1": "Bearer user-cred"})
+
+    async def oauth(request, handler):
+        headers = dict(request.headers or {})
+        headers["Authorization"] = "Bearer oauth-token"
+        return await handler(request.override(headers=headers))
+
+    interceptors = build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: oauth)
+    handler = compose_tool_interceptors(interceptors, _echo_handler)
+    final = asyncio.run(handler(_request(runtime=_runtime_for_user("u1"))))
+    assert final.headers["Authorization"] == "Bearer user-cred"
+    # And on a server without user_auth the OAuth header must survive untouched.
+    final_other = asyncio.run(handler(_request(server_name="other", runtime=_runtime_for_user("u1"))))
+    assert final_other.headers["Authorization"] == "Bearer oauth-token"
+
+
+def test_gateway_masks_sensitive_user_auth_extra_keys():
+    """Secret-bearing extras inside user_auth must be masked by GET like the
+    identical keys at server level, and a masked round-trip must preserve them."""
+    from app.gateway.routers.mcp import (
+        McpServerConfigResponse,
+        McpUserScopedAuthConfigResponse,
+        _mask_server_config,
+        _merge_preserving_secrets,
+    )
+
+    server = McpServerConfigResponse(
+        type="http",
+        url="https://mcp.example.com/mcp",
+        user_auth=McpUserScopedAuthConfigResponse(users={"u1": "Bearer s1"}, client_secret="super-secret", custom_note="keep-me"),
+    )
+    masked = _mask_server_config(server)
+    assert masked.user_auth.model_extra["client_secret"] == "***"
+    assert masked.user_auth.model_extra["custom_note"] == "keep-me"
+
+    # Round-trip: PUT of the masked GET payload keeps the stored secret.
+    merged = _merge_preserving_secrets(masked, server)
+    assert merged.user_auth.model_extra["client_secret"] == "super-secret"
+    assert merged.user_auth.users == {"u1": "Bearer s1"}
--- a/frontend/src/content/en/harness/mcp.mdx
+++ b/frontend/src/content/en/harness/mcp.mdx
@ -105,6 +105,55 @@ When an OAuth-protected MCP server is connected, DeerFlow will:

 The OAuth flow is transparent to the Lead Agent — it simply calls the tool, and DeerFlow handles the authentication.

+## Per-user credentials
+
+A single HTTP/SSE MCP server entry can serve several DeerFlow users, each
+authenticated to the remote service with their own credential. Declare a
+`user_auth` block mapping DeerFlow user ids to credential header values:
+
+```json
+{
+  "mcpServers": {
+    "shared-service": {
+      "type": "http",
+      "url": "https://api.example.com/mcp",
+      "headers": { "Authorization": "$SERVICE_DISCOVERY_TOKEN" },
+      "user_auth": {
+        "header": "Authorization",
+        "users": {
+          "alice-user-id": "$SERVICE_TOKEN_ALICE",
+          "bob-user-id": "$SERVICE_TOKEN_BOB"
+        }
+      }
+    }
+  }
+}
+```
+
+On every tool call, the built-in user-scoped auth interceptor resolves the
+authenticated DeerFlow user and injects that user's credential into the
+configured header. The entry's static `headers` are used only for startup tool
+discovery.
+
+- Values support the same `$ENV_VAR` resolution as the rest of the file, so raw
+  secrets can stay in the process environment.
+- **Fail-closed by default**: a user with no mapped credential — or a mapped
+  `$ENV_VAR` that is unset — gets a clear error instead of another user's
+  credential. Set `"on_missing": "passthrough"` to instead forward such calls
+  with the server's static headers.
+- Combined with per-`(user, thread)` MCP session scoping, users cannot reach
+  each other's authenticated sessions or credentials.
+- The Gateway API masks `user_auth.users` values in GET responses and preserves
+  stored credentials when masked values are round-tripped through PUT.
+
+**Finding the right `users` key.** The resolved user id depends on how the
+deployment authenticates: with LangGraph Server auth it is the safe-slug form
+of the authenticated identity (e.g. `alice-example-com-ab12cd34`), while the
+embedded Gateway resolves the DeerFlow user's raw id (a UUID). The simplest way
+to get the exact key is to have the user attempt a call before being mapped:
+the fail-closed error message includes their resolved id verbatim, ready to
+copy into `user_auth.users`.
+
 ## Managing MCP servers

 MCP servers can be managed in several ways: