diff --git a/backend/app/gateway/routers/mcp.py b/backend/app/gateway/routers/mcp.py index 3308c980a..8d87934d0 100644 --- a/backend/app/gateway/routers/mcp.py +++ b/backend/app/gateway/routers/mcp.py @@ -370,6 +370,43 @@ class McpUserScopedAuthConfigResponse(BaseModel): return value +class McpContextHeadersConfigResponse(BaseModel): + """Per-request credential injection configuration for an MCP server. + + Holds header names and run-context key names only — never a credential — + so unlike ``user_auth`` its declared fields are returned unmasked by GET. + """ + + enabled: bool = Field(default=True, description="Whether request-scoped header injection is enabled") + headers: dict[str, str] = Field( + default_factory=dict, + description="Map of HTTP header name to the key read from the run request's config.context.secrets", + ) + on_missing: Literal["deny", "passthrough"] = Field(default="deny", description="Behavior when a mapped key is absent from the request secrets") + # Mirror the harness-side McpContextHeadersConfig (extra="allow"): without + # this, an operator's unknown key inside headers_from_context would be + # silently stripped by the next admin PUT. + model_config = ConfigDict(extra="allow") + + # Mirror the harness-side entry check: a blank name accepted here would be + # persisted, then fail ExtensionsConfig validation on reload — wedging every + # subsequent config load until the file is hand-edited. + @field_validator("headers") + @classmethod + def _validate_mapping_entries(cls, value: dict[str, str]) -> dict[str, str]: + seen: dict[str, str] = {} + for header_name, secret_key in value.items(): + if not header_name.strip(): + raise ValueError("headers_from_context.headers must not contain a blank header name") + if not isinstance(secret_key, str) or not secret_key.strip(): + raise ValueError(f"headers_from_context.headers[{header_name!r}] must name a non-blank secret key from config.context.secrets") + lowered = header_name.lower() + if lowered in seen: + raise ValueError(f"headers_from_context.headers maps the same HTTP header under two spellings ({seen[lowered]!r} and {header_name!r}); header names are case-insensitive, so keep only one") + seen[lowered] = header_name + return value + + class McpOAuthConfigResponse(BaseModel): """OAuth configuration for an MCP server.""" @@ -401,6 +438,7 @@ class McpServerConfigResponse(BaseModel): 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") + headers_from_context: McpContextHeadersConfigResponse | None = Field(default=None, description="Per-request credential injection for MCP HTTP/SSE servers: map header names to config.context.secrets keys") 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") @@ -422,6 +460,22 @@ class McpServerConfigResponse(BaseModel): ) model_config = ConfigDict(extra="allow") + @field_validator("headers") + @classmethod + def _validate_header_names(cls, value: dict[str, str]) -> dict[str, str]: + # Mirror the harness-side McpServerConfig check: HTTP field names are + # case-insensitive, so a config carrying one header under two spellings + # would persist, reload into a connection with both fields, and let a + # later override replace only one of them. Reject at the API boundary + # instead of wedging the next ExtensionsConfig reload. + seen: dict[str, str] = {} + for header_name in value: + lowered = header_name.lower() + if lowered in seen: + raise ValueError(f"headers maps the same HTTP header under two spellings ({seen[lowered]!r} and {header_name!r}); header names are case-insensitive, so keep only one") + seen[lowered] = header_name + return value + @model_validator(mode="before") @classmethod def _accept_transport_alias(cls, data: Any) -> Any: @@ -685,6 +739,14 @@ def _mask_server_config(server: McpServerConfigResponse) -> McpServerConfigRespo # 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_headers_from_context = None + if server.headers_from_context is not None: + # The declared fields hold names only and stay in cleartext — masking + # them would show operators `***` where a header name belongs. Extras + # get the same treatment as everywhere else, since `extra="allow"` lets + # an operator store a secret-bearing key that round-trips through PUT. + masked_ch_extra = {key: _MASKED_VALUE if _is_sensitive_extra_key(key) else _mask_sensitive_extra_value(value) for key, value in (server.headers_from_context.model_extra or {}).items()} + masked_headers_from_context = server.headers_from_context.model_copy(update=masked_ch_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={ @@ -692,6 +754,7 @@ def _mask_server_config(server: McpServerConfigResponse) -> McpServerConfigRespo "headers": masked_headers, "oauth": masked_oauth, "user_auth": masked_user_auth, + "headers_from_context": masked_headers_from_context, **masked_extra, } ) @@ -801,14 +864,47 @@ def _merge_preserving_secrets( effective["users"] = merged_users merged_user_auth = McpUserScopedAuthConfigResponse(**effective) + merged_context_headers = incoming.headers_from_context + if incoming.headers_from_context is not None: + # Sub-field-aware merge, mirroring user_auth above: an explicit partial + # block (e.g. {"enabled": false}) must not wipe the stored mapping or + # reset on_missing back to its default. Only fields the request set are + # replaced, so an explicitly supplied ``headers`` (even {}) replaces the + # map while omitted fields carry over from the stored block. + incoming_ch = incoming.headers_from_context + base_ch = existing.headers_from_context + set_fields = incoming_ch.model_fields_set + effective: dict[str, Any] = {} + if base_ch is not None: + effective.update({name: getattr(base_ch, name) for name in ("enabled", "headers", "on_missing")}) + effective.update(base_ch.model_extra or {}) + for name in ("enabled", "headers", "on_missing"): + if name in set_fields: + effective[name] = getattr(incoming_ch, 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 user_auth extras and server-level extras get. + base_ch_extra = (base_ch.model_extra or {}) if base_ch is not None else {} + for key, value in (incoming_ch.model_extra or {}).items(): + effective[key] = _merge_extra_value_preserving_masked( + key, + value, + base_ch_extra.get(key), + existing_present=key in base_ch_extra, + ) + merged_context_headers = McpContextHeadersConfigResponse(**effective) + update = { "env": merged_env, "headers": merged_headers, "oauth": merged_oauth, "user_auth": merged_user_auth, + "headers_from_context": merged_context_headers, } if "user_auth" not in incoming.model_fields_set: update["user_auth"] = existing.user_auth + if "headers_from_context" not in incoming.model_fields_set: + update["headers_from_context"] = existing.headers_from_context if "routing" not in incoming.model_fields_set: update["routing"] = existing.routing if "tools" not in incoming.model_fields_set: diff --git a/backend/docs/MCP_SERVER.md b/backend/docs/MCP_SERVER.md index 914dcd8b9..61f18e5ef 100644 --- a/backend/docs/MCP_SERVER.md +++ b/backend/docs/MCP_SERVER.md @@ -305,7 +305,11 @@ normally use an independently running HTTP/SSE service. Server-level OAuth works during background polling and refreshes normally. Request-scoped secrets from a particular Agent run are not durable task credentials and are unavailable to later background polls; use server-level -authentication for a task toolset. Restart DeerFlow after changing +authentication for a task toolset. `headers_from_context` follows the same +rule: submit is awaited inside the Agent run and carries the mapped headers, +while status and cancel polls skip them and authenticate with the server's +static or OAuth credentials — so `on_missing: "deny"` guards the submit but not +those polls. Declaring both on one server logs a warning at startup. Restart DeerFlow after changing `mcp_tasks`, `task_toolsets`, `mcpInterceptors`, or any connection, authentication, transport, or timeout setting on a task-enabled server. DeerFlow rejects task-tool reloads that no longer match the Gateway's startup @@ -344,6 +348,81 @@ Example: } ``` +## Request-Scoped Headers (HTTP/SSE MCP Servers) + +When the credential is chosen by the *caller* rather than by the operator — +multi-tenant gateways, per-run API keys, one shared MCP server fronting several +environments — declare a `headers_from_context` block instead of registering one +MCP server per credential. + +Each entry maps an HTTP header name to a key of the run request's +`config.context.secrets` carrier: + +```json +{ + "mcpServers": { + "shared-api": { + "enabled": true, + "type": "http", + "url": "https://mcp.example.com/mcp", + "headers": { "Authorization": "Bearer $MCP_DISCOVERY_TOKEN" }, + "headers_from_context": { + "enabled": true, + "headers": { + "X-Tenant-Id": "tenant_id", + "Authorization": "tenant_token" + }, + "on_missing": "deny" + } + } + } +} +``` + +The caller supplies the values on each run request: + +```json +{ + "config": { + "context": { + "secrets": { + "tenant_id": "acme", + "tenant_token": "Bearer " + } + } + } +} +``` + +- The config file stores **names only**, never a credential, so the block is + returned unmasked by `GET /api/mcp/config`. The values travel out-of-band with + each run and are stripped from persisted run configuration, API responses, and + trace payloads. +- The server's static `headers` are used for startup tool discovery. A mapped + header replaces the static one for that tool call, as shown above for + `Authorization`. Header names are matched case-insensitively, so a mapped + `Authorization` still replaces a static `authorization` instead of putting a + second copy of the field on the wire. Mapping one header under two spellings + is rejected at config load. +- `on_missing` defaults to `"deny"`: if the run carries no value for a mapped + key, the tool call fails with an actionable error rather than falling back to + the discovery credential — which in a multi-tenant deployment would send one + tenant's request under another tenant's authority. Set `"passthrough"` to opt + out and forward the static headers instead. +- Precedence for a server declaring several sources: static `headers` < + `oauth` < `user_auth` < `headers_from_context`. The value chosen for this one + request is the most specific, so it wins. +- `sse`/`http` only. A stdio server has no HTTP headers; declaring the block + there logs a warning and is ignored. +- Durable background tasks are the one exception, and only half of one: a + `task_toolsets` submit is awaited inside the Agent run and carries these + headers, but the status and cancel polls run after that run ends, so they skip + them and use the server's static/OAuth credentials. See *Durable Background + Tasks* above. + +Use `user_auth` instead when the credential belongs to a configured DeerFlow +user rather than to the individual request. + ## Custom Tool Interceptors You can register custom interceptors that run before every MCP tool call. This is useful for injecting per-request headers (e.g., user auth tokens from the LangGraph execution context), logging, or metrics. @@ -362,16 +441,19 @@ Declare interceptors in `extensions_config.json` using the `mcpInterceptors` fie Each entry is a Python import path in `module:variable` format (resolved via `resolve_variable`). The variable must be a **no-arg builder function** that returns an async interceptor compatible with `MultiServerMCPClient`’s `tool_interceptors` interface, or `None` to skip. Example interceptor that injects an authorization header from the request-scoped -LangGraph secret context: +LangGraph secret context. For a plain header mapping prefer the declarative +`headers_from_context` block above; write an interceptor when the header value +needs logic (signing, exchanging the secret for another token, routing on the +tool name): ```python -from langgraph.config import get_config +from deerflow.runtime.secret_context import extract_request_secrets def build_auth_interceptor(): async def interceptor(request, handler): - config = get_config() - secrets = (config.get("context") or {}).get("secrets") or {} + runtime = getattr(request, "runtime", None) + secrets = extract_request_secrets(getattr(runtime, "context", None)) token = secrets.get("MCP_AUTH_TOKEN") if token: request = request.override( @@ -382,6 +464,16 @@ def build_auth_interceptor(): return interceptor ``` +Read the run context from `request.runtime`, not from +`langgraph.config.get_config()`. The context is carried on the LangGraph +runtime, not on the `RunnableConfig` propagated to child runnables, so +`get_config().get("context")` is `None` inside a tool call. LangGraph's tool +node injects the runtime into any tool parameter named `runtime`, which is how +both the pooled stdio wrapper and `langchain-mcp-adapters`' HTTP/SSE tool +receive it. When the call originates outside a tool node, fall back to +`langgraph.runtime.get_runtime()` (see +`deerflow/mcp/context_headers.py::_current_runtime`). + Supply the credential on each run request through `config.context.secrets`: ```json diff --git a/backend/packages/harness/deerflow/config/extensions_config.py b/backend/packages/harness/deerflow/config/extensions_config.py index ae1987ed1..198a5b568 100644 --- a/backend/packages/harness/deerflow/config/extensions_config.py +++ b/backend/packages/harness/deerflow/config/extensions_config.py @@ -132,6 +132,51 @@ class McpUserScopedAuthConfig(BaseModel): return value +class McpContextHeadersConfig(BaseModel): + """Per-request credential injection for an MCP server (HTTP/SSE transports). + + Maps HTTP header names to keys of the run request's ``config.context.secrets`` + carrier, so one configured MCP server can serve callers that each supply their + own credential *per request* rather than per configured user. The built-in + context-headers interceptor resolves the mapping on every tool call; the + server entry's static ``headers`` are only used for startup tool discovery. + + Unlike ``user_auth``, this block stores **no credential** — only header names + and run-context key names — so it is safe to return unmasked from the config + API. The values arrive out-of-band with each run and never enter the prompt, + tool arguments, or trace payloads (see ``runtime/secret_context.py``). + """ + + enabled: bool = Field(default=True, description="Whether request-scoped header injection is enabled") + headers: dict[str, str] = Field( + default_factory=dict, + description="Map of HTTP header name to the key to read from the run request's config.context.secrets (e.g. {'X-Tenant-Id': 'tenant_id'})", + ) + on_missing: Literal["deny", "passthrough"] = Field( + default="deny", + description=("Behavior when a mapped key is absent from the request secrets (or 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("headers") + @classmethod + def _validate_mapping_entries(cls, value: dict[str, str]) -> dict[str, str]: + seen: dict[str, str] = {} + for header_name, secret_key in value.items(): + if not header_name.strip(): + raise ValueError("headers_from_context.headers must not contain a blank header name") + if not isinstance(secret_key, str) or not secret_key.strip(): + raise ValueError(f"headers_from_context.headers[{header_name!r}] must name a non-blank secret key from config.context.secrets") + # HTTP field names are case-insensitive, so two spellings of one + # header are one header with two candidate values, and which one + # reaches the remote would depend on dict ordering. + lowered = header_name.lower() + if lowered in seen: + raise ValueError(f"headers_from_context.headers maps the same HTTP header under two spellings ({seen[lowered]!r} and {header_name!r}); header names are case-insensitive, so keep only one") + seen[lowered] = header_name + return value + + class McpOAuthConfig(BaseModel): """OAuth configuration for an MCP server (HTTP/SSE transports).""" @@ -170,6 +215,10 @@ class McpServerConfig(BaseModel): default=None, description="Per-user credential injection (for sse or http type): map DeerFlow user ids to per-user credential header values", ) + headers_from_context: McpContextHeadersConfig | None = Field( + default=None, + description="Per-request credential injection (for sse or http type): map HTTP header names to keys of the run request's config.context.secrets", + ) 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") @@ -196,6 +245,23 @@ class McpServerConfig(BaseModel): ) model_config = ConfigDict(extra="allow") + @field_validator("headers") + @classmethod + def _validate_header_names(cls, value: dict[str, str]) -> dict[str, str]: + # HTTP field names are case-insensitive, so two spellings of one header + # are one field with two candidate values. The adapter copies the static + # mapping verbatim, so both would reach the wire; a later per-request or + # OAuth override only replaces one spelling, leaving the other to leak a + # shared credential across tenant authority. Reject at config time so a + # bad mapping cannot reach the connection. + seen: dict[str, str] = {} + for header_name in value: + lowered = header_name.lower() + if lowered in seen: + raise ValueError(f"headers maps the same HTTP header under two spellings ({seen[lowered]!r} and {header_name!r}); header names are case-insensitive, so keep only one") + seen[lowered] = header_name + return value + @model_validator(mode="before") @classmethod def _accept_transport_alias(cls, data: Any) -> Any: diff --git a/backend/packages/harness/deerflow/mcp/AGENTS.md b/backend/packages/harness/deerflow/mcp/AGENTS.md index 876c9172e..896c299a2 100644 --- a/backend/packages/harness/deerflow/mcp/AGENTS.md +++ b/backend/packages/harness/deerflow/mcp/AGENTS.md @@ -11,6 +11,9 @@ - **Per-server tool-name prefixing**: `mcpServers..tool_name_prefix` defaults to `true`, preserving the collision-safe `_` 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..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. +- **Per-request credentials (HTTP/SSE)**: `mcpServers..headers_from_context` maps HTTP header names to keys of the run request's `config.context.secrets` carrier, for credentials the caller chooses per request (multi-tenant gateways, per-run API keys) rather than per configured user. `mcp/context_headers.py::build_context_headers_interceptor` resolves the mapping on every tool call and rewrites those headers. Registered **after** `user_auth` in `mcp/interceptors.py`, so for a server declaring several sources the per-request value wins the final header: precedence is static `headers` < `oauth` < `user_auth` < `headers_from_context`. Fail-closed: a mapped key absent from the request secrets (or resolved empty) raises a `ToolException` naming only the missing key, unless `on_missing: "passthrough"` is set — a silent fallback would send one tenant's call under the discovery credential's authority. `sse`/`http` only; a stdio server warns and is skipped, as with `user_auth`. The block stores names, never a credential, so the Gateway returns it unmasked and a `PUT` replaces the declared mapping verbatim; only `extra="allow"` keys inside it get sensitive-key masking, and those are restored from the stored block on a round-trip like every other masked extra. Durable `task_toolsets` calls split: submit is awaited inside the Agent run and carries the mapped headers (`McpTaskToolCaller.call_tool(request_scoped_headers=True)`, set only by `OrdinaryMcpTaskDriver.submit`), while status and cancel run after that run ended and keep the server-level credentials — so `on_missing: "deny"` covers submit but not those polls, which is what the startup warning is about. +- **Header names are case-insensitive** (`mcp/headers.py`): every credential interceptor writes through `apply_header_overrides`, which drops a key differing only in case and emits the spelling the connection already uses. Without it a static `authorization` and an injected `Authorization` both reach httpx — the adapter merges connection and override headers with a plain `{**static, **override}` splat — and a server reading the field with a single-value accessor gets the static entry, silently inverting the precedence above. `headers_from_context.headers` also rejects two spellings of one header at config load. +- **Reading the run context from an interceptor**: use `request.runtime` (LangGraph's tool node injects a `ToolRuntime` into any tool parameter named `runtime`, which covers both the pooled stdio wrapper and `langchain-mcp-adapters`' own HTTP/SSE tool), falling back to ambient `langgraph.runtime.get_runtime()`. Do **not** use `langgraph.config.get_config()["context"]`: the run context rides the runtime, not the `RunnableConfig` propagated to child runnables, so that key is `None` inside a tool call. `tests/test_mcp_context_headers.py::test_adapter_tool_receives_the_runtime_langgraph_injects` pins the injection rule against an upstream rename by disabling the ambient fallback and driving a real adapter tool through a real graph. - **Routing hints**: `extensions_config.json -> mcpServers..routing` and `tools..routing` are soft preference metadata. The effective routing is resolved while `mcp/tools.py::get_mcp_tools()` still has both diff --git a/backend/packages/harness/deerflow/mcp/context_headers.py b/backend/packages/harness/deerflow/mcp/context_headers.py new file mode 100644 index 000000000..7777e5edd --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/context_headers.py @@ -0,0 +1,162 @@ +"""Per-request credential injection for shared MCP servers. + +``user_auth`` binds a credential to a *configured* DeerFlow user, which forces +one MCP server entry per credential when the credential is chosen by the caller +at request time (multi-tenant gateways, per-run API keys). This module closes +that gap: a server opts in by declaring a ``headers_from_context`` block +(:class:`McpContextHeadersConfig`) mapping HTTP header names to keys of the run +request's ``config.context.secrets`` carrier. On every tool call the interceptor +resolves the mapping from the live run context and rewrites those headers via +``request.override(headers=...)`` — the same per-call mechanism the OAuth and +user-scoped auth interceptors use. + +The secret values arrive out-of-band with the run request and stay there: they +are never rendered into the prompt, the tool arguments, or trace payloads (see +``runtime/secret_context.py``). Only the *names* live in the config file, so no +credential is written to disk or returned by the config API. + +Registered last in ``mcp/interceptors.py``, so for a server declaring several +credential sources the per-request value wins the final header — interceptors +wrap outermost-first, and the later-registered one runs closer to the transport. +Header names are written case-insensitively through ``mcp/headers.py``, so a +mapped ``Authorization`` replaces a static ``authorization`` rather than putting +a second copy of the field on the wire ahead of it. + +Fail-closed by default: a mapped key that is absent from the request secrets +(or resolved empty) gets an actionable ``ToolException`` rather than silently +falling back to the server's static discovery credential, which in a +multi-tenant deployment would send one tenant's request under another +tenant's authority. ``on_missing: "passthrough"`` is the explicit opt-out. +""" + +from __future__ import annotations + +import logging +from typing import Any + +from langchain_core.tools import ToolException + +from deerflow.config.extensions_config import ExtensionsConfig, McpContextHeadersConfig +from deerflow.mcp.headers import apply_header_overrides, header_spellings +from deerflow.runtime.secret_context import extract_request_secrets + +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). Mirrors ``mcp/user_scoped_auth.py``: a failure here + only means the request carries no resolvable secrets, which the caller then + handles through ``on_missing``. + """ + try: + from langgraph.runtime import get_runtime + + return get_runtime() + except Exception: + return None + + +def _request_secrets(request: Any) -> dict[str, str]: + """Return the run request's ``config.context.secrets``, or ``{}``. + + Prefer the runtime attached to the request: LangGraph's tool node injects it + into any tool parameter named ``runtime``, which covers both the pooled + stdio wrapper and ``langchain_mcp_adapters``' own HTTP/SSE tool. Fall back to + the ambient runtime for call paths outside a tool node. + + Deliberately not read from ``langgraph.config.get_config()``: the run context + is carried on the runtime, not on the ``RunnableConfig`` propagated to child + runnables, so ``get_config().get("context")`` is ``None`` inside a tool call. + """ + runtime = getattr(request, "runtime", None) + if runtime is None: + runtime = _current_runtime() + return extract_request_secrets(getattr(runtime, "context", None)) + + +def build_context_headers_interceptor(extensions_config: ExtensionsConfig) -> Any | None: + """Build a tool interceptor injecting per-request headers, or ``None``. + + Returns ``None`` when no enabled server declares a usable + ``headers_from_context`` block, so callers can skip registration entirely + (mirrors ``build_oauth_tool_interceptor`` / ``build_user_scoped_auth_interceptor``). + """ + mapping_by_server: dict[str, McpContextHeadersConfig] = {} + # The server's static header spellings, so a mapped name that differs from + # the configured one only in case still *replaces* it at the adapter's + # case-sensitive connection merge instead of riding alongside it. + spellings_by_server: dict[str, dict[str, str]] = {} + for server_name, server_config in extensions_config.get_enabled_mcp_servers().items(): + context_headers = server_config.headers_from_context + if context_headers is None or not context_headers.enabled or not context_headers.headers: + 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 + # runs that carry no secrets. Warn-and-skip matches user_auth. + logger.warning( + "MCP server '%s' declares headers_from_context but uses the '%s' transport; request-scoped headers only apply to 'sse'/'http' servers — ignoring headers_from_context for this server", + server_name, + server_config.type, + ) + continue + if server_config.task_toolsets: + # Submitting a durable task happens inside the Agent run and carries + # the request secrets; the later status/cancel polls do not, because + # the task runtime drives them long after that run ended. Those calls + # deliberately skip this interceptor (see McpTaskToolCaller), so the + # background half authenticates with the server's own credentials. + logger.warning( + "MCP server '%s' declares both headers_from_context and task_toolsets; background task status/cancel polls run outside an Agent run and will use this server's static/OAuth credentials instead of the per-request headers", + server_name, + ) + mapping_by_server[server_name] = context_headers + spellings_by_server[server_name] = header_spellings(server_config.headers) + + if not mapping_by_server: + return None + + async def context_headers_interceptor(request: Any, handler: Any) -> Any: + context_headers = mapping_by_server.get(request.server_name) + if context_headers is None: + return await handler(request) + + secrets = _request_secrets(request) + resolved: dict[str, str] = {} + missing: list[str] = [] + for header_name, secret_key in context_headers.headers.items(): + # Empty string covers a caller-side `$ENV_VAR` that was unset: an + # empty credential must fail closed rather than send an empty header. + value = secrets.get(secret_key, "") + if value: + resolved[header_name] = value + else: + missing.append(secret_key) + + if missing and context_headers.on_missing == "deny": + missing_keys = ", ".join(sorted(missing)) + logger.warning( + "Denied MCP tool call to server '%s': request context is missing secret(s) %s", + request.server_name, + missing_keys, + ) + # Only the configured *key names* are surfaced — they already live in + # the config file, so this leaks nothing the operator has not written + # down, while telling the caller exactly what to send. + raise ToolException(f"MCP server '{request.server_name}' needs request-scoped credential(s) {missing_keys}. Send them in config.context.secrets, or set this server's headers_from_context.on_missing to 'passthrough'.") + + if not resolved: + return await handler(request) + + updated_headers = apply_header_overrides( + request.headers, + resolved, + spellings=spellings_by_server.get(request.server_name), + ) + return await handler(request.override(headers=updated_headers)) + + return context_headers_interceptor diff --git a/backend/packages/harness/deerflow/mcp/headers.py b/backend/packages/harness/deerflow/mcp/headers.py new file mode 100644 index 000000000..67b6089fc --- /dev/null +++ b/backend/packages/harness/deerflow/mcp/headers.py @@ -0,0 +1,61 @@ +"""Case-insensitive header writes for MCP tool-call interceptors. + +HTTP field names are case-insensitive (RFC 9110 §5.1), but every dictionary on +the path from config to the wire is case-*sensitive*: ``build_server_params`` +copies the operator's static ``headers`` spelling verbatim, and +``langchain_mcp_adapters`` merges interceptor overrides with a plain +``{**connection_headers, **override_headers}`` splat. So a static +``authorization`` and an interceptor-written ``Authorization`` do not collide — +both survive, httpx puts both on the wire, and a server reading the field with a +single-value accessor sees the *first* one, which is the static entry the +override was supposed to replace. + +The credential interceptors therefore write header names through +:func:`apply_header_overrides`, which drops any key differing only in case and +emits the spelling the connection already uses, so the adapter's merge replaces +the static entry instead of duplicating it. +""" + +from __future__ import annotations + +from collections.abc import Iterable, Mapping + + +def header_spellings(names: Iterable[str] | None) -> dict[str, str]: + """Index header names by their lowercased form. + + Used to pin the spelling an interceptor should emit: the connection's own + static ``headers`` keys, which the adapter merges the override into. A + server that declares none passes ``None`` here rather than being special- + cased at each call site. + """ + return {name.lower(): name for name in (names or ())} + + +def apply_header_overrides( + base: Mapping[str, str] | None, + overrides: Mapping[str, str], + *, + spellings: Mapping[str, str] | None = None, +) -> dict[str, str]: + """Return ``base`` with ``overrides`` applied, case-insensitively. + + ``spellings`` maps a lowercased header name to the spelling to emit and + takes priority over ``base``'s own keys, so an override lands on the static + connection header it is meant to replace even when an earlier interceptor + already wrote a differently-cased variant. Any key of ``base`` that differs + from the emitted name only in case is removed, so the result never carries + one header under two spellings. + """ + merged = dict(base or {}) + lookup = dict(spellings or {}) + for key in merged: + lookup.setdefault(key.lower(), key) + + for name, value in overrides.items(): + lowered = name.lower() + canonical = lookup.get(lowered, name) + for existing in [key for key in merged if key != canonical and key.lower() == lowered]: + del merged[existing] + merged[canonical] = value + return merged diff --git a/backend/packages/harness/deerflow/mcp/interceptors.py b/backend/packages/harness/deerflow/mcp/interceptors.py index 65578ed75..05a52fa11 100644 --- a/backend/packages/harness/deerflow/mcp/interceptors.py +++ b/backend/packages/harness/deerflow/mcp/interceptors.py @@ -6,6 +6,7 @@ import logging from typing import Any from deerflow.config.extensions_config import ExtensionsConfig +from deerflow.mcp.context_headers import build_context_headers_interceptor 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 @@ -18,10 +19,11 @@ def build_mcp_tool_interceptors( *, oauth_builder: Any = build_oauth_tool_interceptor, user_auth_builder: Any = build_user_scoped_auth_interceptor, + context_headers_builder: Any = build_context_headers_interceptor, resolver: Any = resolve_variable, target_logger: logging.Logger = logger, ) -> list[Any]: - """Build OAuth, user-scoped auth, then configured custom MCP interceptors.""" + """Build OAuth, user-scoped auth, context headers, then custom MCP interceptors.""" interceptors: list[Any] = [] oauth_interceptor = oauth_builder(extensions_config) if oauth_interceptor is not None: @@ -34,6 +36,14 @@ def build_mcp_tool_interceptors( if user_auth_interceptor is not None: interceptors.append(user_auth_interceptor) + # Last of the built-ins, by the same rule: a credential the caller chose for + # this one request is more specific than a configured per-user or per-server + # credential, so it must win the final header value for a server declaring + # more than one source. + context_headers_interceptor = context_headers_builder(extensions_config) + if context_headers_interceptor is not None: + interceptors.append(context_headers_interceptor) + raw_paths = (extensions_config.model_extra or {}).get("mcpInterceptors") if isinstance(raw_paths, str): raw_paths = [raw_paths] diff --git a/backend/packages/harness/deerflow/mcp/oauth.py b/backend/packages/harness/deerflow/mcp/oauth.py index b757b6455..4c7672148 100644 --- a/backend/packages/harness/deerflow/mcp/oauth.py +++ b/backend/packages/harness/deerflow/mcp/oauth.py @@ -10,6 +10,7 @@ from datetime import UTC, datetime, timedelta from typing import Any from deerflow.config.extensions_config import ExtensionsConfig, McpOAuthConfig +from deerflow.mcp.headers import apply_header_overrides, header_spellings logger = logging.getLogger(__name__) @@ -188,13 +189,21 @@ def build_oauth_tool_interceptor( if not token_manager.has_oauth_servers(): return None + # The servers' static header spellings, so the injected token replaces a + # static header spelled 'authorization' at the adapter's case-sensitive + # connection merge instead of riding alongside it (see ``mcp/headers.py``). + spellings_by_server = {server_name: header_spellings(server_config.headers) for server_name, server_config in extensions_config.get_enabled_mcp_servers().items()} + async def oauth_interceptor(request: Any, handler: Any) -> Any: header = await token_manager.get_authorization_header(request.server_name) if not header: return await handler(request) - updated_headers = dict(request.headers or {}) - updated_headers["Authorization"] = header + updated_headers = apply_header_overrides( + request.headers, + {"Authorization": header}, + spellings=spellings_by_server.get(request.server_name), + ) return await handler(request.override(headers=updated_headers)) return oauth_interceptor diff --git a/backend/packages/harness/deerflow/mcp/task_tool_caller.py b/backend/packages/harness/deerflow/mcp/task_tool_caller.py index 8dadb0440..3643061d0 100644 --- a/backend/packages/harness/deerflow/mcp/task_tool_caller.py +++ b/backend/packages/harness/deerflow/mcp/task_tool_caller.py @@ -12,6 +12,8 @@ from deerflow.config.extensions_config import ExtensionsConfig from deerflow.config.paths import get_paths from deerflow.constants import MCP_TMP_SUBDIR from deerflow.mcp.client import build_server_params +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.session_pool import get_session_pool @@ -58,13 +60,29 @@ class McpTaskToolCaller: ) -> None: self._extensions_config = extensions_config self._oauth_token_manager = oauth_token_manager or OAuthTokenManager.from_extensions_config(extensions_config) - self._interceptors = build_mcp_tool_interceptors( + context_headers_interceptor = build_context_headers_interceptor(extensions_config) + # Built once so the two chains keep an identical interceptor order and a + # custom ``mcpInterceptors`` builder is invoked exactly once. + self._submit_interceptors = build_mcp_tool_interceptors( extensions_config, oauth_builder=lambda config: build_oauth_tool_interceptor( config, token_manager=self._oauth_token_manager, ), + context_headers_builder=lambda _config: context_headers_interceptor, ) + # Submitting a durable task is awaited inline inside the Agent's tool + # call, so ``config.context.secrets`` is still reachable through the + # ambient LangGraph runtime and the submit goes out under the caller's + # own credential. The later status/cancel polls are driven by the task + # runtime long after that run ended: there is no run context to read, so + # the fail-closed interceptor would deny every poll. Those keep using + # server-level credentials (see docs/MCP_SERVER.md), which is what + # ``build_context_headers_interceptor`` warns about at startup. + if context_headers_interceptor is None: + self._interceptors = self._submit_interceptors + else: + self._interceptors = [interceptor for interceptor in self._submit_interceptors if interceptor is not context_headers_interceptor] async def call_tool( self, @@ -74,7 +92,16 @@ class McpTaskToolCaller: arguments: dict[str, Any], user_id: str, thread_id: str, + request_scoped_headers: bool = False, ) -> 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. + """ + interceptors = self._submit_interceptors if request_scoped_headers else self._interceptors server_config = self._extensions_config.get_enabled_mcp_servers().get(server_name) if server_config is None: raise LookupError(f"MCP task server {server_name!r} is missing or disabled in the startup configuration") @@ -116,6 +143,7 @@ class McpTaskToolCaller: timeout_seconds=server_config.tool_call_timeout, session_init_timeout_seconds=None, persistent_session=True, + interceptors=interceptors, ) except Exception: # A dead pooled subprocess must not poison every later status @@ -125,9 +153,10 @@ class McpTaskToolCaller: authorization = await self._oauth_token_manager.get_authorization_header(server_name) if authorization: - headers = dict(connection.get("headers") or {}) - headers["Authorization"] = authorization - connection["headers"] = headers + connection["headers"] = apply_header_overrides( + connection.get("headers") or {}, + {"Authorization": authorization}, + ) return await self._invoke( session=None, connection=connection, @@ -137,6 +166,7 @@ class McpTaskToolCaller: timeout_seconds=server_config.tool_call_timeout, session_init_timeout_seconds=server_config.session_init_timeout, persistent_session=False, + interceptors=interceptors, ) async def _invoke( @@ -150,6 +180,7 @@ class McpTaskToolCaller: timeout_seconds: float | None, session_init_timeout_seconds: float | None, persistent_session: bool, + interceptors: list[Any], ) -> Any: from langchain_mcp_adapters.interceptors import MCPToolCallRequest from langchain_mcp_adapters.sessions import create_session @@ -173,9 +204,10 @@ class McpTaskToolCaller: effective_connection = dict(connection) if request.headers: - headers = dict(effective_connection.get("headers") or {}) - headers.update(dict(request.headers)) - effective_connection["headers"] = headers + effective_connection["headers"] = apply_header_overrides( + effective_connection.get("headers") or {}, + dict(request.headers), + ) captured: BaseException | None = None call_result: Any | None = None async with create_session(effective_connection) as remote_session: @@ -209,7 +241,7 @@ class McpTaskToolCaller: return call_result handler = execute - for interceptor in reversed(self._interceptors): + for interceptor in reversed(interceptors): inner = handler async def wrapped(request: Any, _interceptor: Any = interceptor, _inner: Any = inner) -> Any: diff --git a/backend/packages/harness/deerflow/mcp/tasks/ordinary.py b/backend/packages/harness/deerflow/mcp/tasks/ordinary.py index 4d8a84fc8..6faea18a8 100644 --- a/backend/packages/harness/deerflow/mcp/tasks/ordinary.py +++ b/backend/packages/harness/deerflow/mcp/tasks/ordinary.py @@ -32,6 +32,7 @@ class McpTaskToolCaller(Protocol): arguments: dict[str, Any], user_id: str, thread_id: str, + request_scoped_headers: bool = False, ) -> Any: ... @@ -158,6 +159,10 @@ class OrdinaryMcpTaskDriver: arguments=request.arguments, user_id=request.user_id, thread_id=request.thread_id, + # Submit alone is awaited inside the Agent run, so it is the one + # durable-task call that can carry the run's request-scoped + # credentials; status and cancel run after that run ended. + request_scoped_headers=True, ) payload = _parse( _SubmitPayload, diff --git a/backend/packages/harness/deerflow/mcp/tools.py b/backend/packages/harness/deerflow/mcp/tools.py index 30d0de18e..b35eb7f83 100644 --- a/backend/packages/harness/deerflow/mcp/tools.py +++ b/backend/packages/harness/deerflow/mcp/tools.py @@ -22,6 +22,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.headers import apply_header_overrides 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 MCPSessionPool, get_session_pool @@ -820,9 +821,12 @@ async def get_mcp_tools() -> list[BaseTool]: if server_name not in servers_config: continue if servers_config[server_name].get("transport") in ("sse", "http"): - existing_headers = dict(servers_config[server_name].get("headers", {})) - existing_headers["Authorization"] = auth_header - servers_config[server_name]["headers"] = existing_headers + # Case-insensitive write: a static header spelled 'authorization' + # must be replaced, not joined on the wire by a second field. + servers_config[server_name]["headers"] = apply_header_overrides( + servers_config[server_name].get("headers", {}), + {"Authorization": auth_header}, + ) tool_interceptors = build_mcp_tool_interceptors( extensions_config, diff --git a/backend/packages/harness/deerflow/mcp/user_scoped_auth.py b/backend/packages/harness/deerflow/mcp/user_scoped_auth.py index 5c2ba4f62..f2f08e110 100644 --- a/backend/packages/harness/deerflow/mcp/user_scoped_auth.py +++ b/backend/packages/harness/deerflow/mcp/user_scoped_auth.py @@ -27,6 +27,7 @@ from typing import Any from langchain_core.tools import ToolException from deerflow.config.extensions_config import ExtensionsConfig, McpUserScopedAuthConfig +from deerflow.mcp.headers import apply_header_overrides, header_spellings from deerflow.runtime.user_context import resolve_runtime_user_id logger = logging.getLogger(__name__) @@ -56,6 +57,10 @@ def build_user_scoped_auth_interceptor(extensions_config: ExtensionsConfig) -> A ``build_oauth_tool_interceptor``). """ user_auth_by_server: dict[str, McpUserScopedAuthConfig] = {} + # The server's static header spellings, so a configured ``header`` that + # differs from the static one only in case still *replaces* it at the + # adapter's case-sensitive connection merge (see ``mcp/headers.py``). + spellings_by_server: dict[str, dict[str, str]] = {} 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 @@ -72,6 +77,7 @@ def build_user_scoped_auth_interceptor(extensions_config: ExtensionsConfig) -> A ) continue user_auth_by_server[server_name] = server_config.user_auth + spellings_by_server[server_name] = header_spellings(server_config.headers) if not user_auth_by_server: return None @@ -110,8 +116,11 @@ def build_user_scoped_auth_interceptor(extensions_config: ExtensionsConfig) -> A 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 + updated_headers = apply_header_overrides( + request.headers, + {user_auth.header: credential}, + spellings=spellings_by_server.get(request.server_name), + ) return await handler(request.override(headers=updated_headers)) return user_scoped_auth_interceptor diff --git a/backend/packages/harness/deerflow/skills/AGENTS.md b/backend/packages/harness/deerflow/skills/AGENTS.md index 4e8766dcc..661a7fee8 100644 --- a/backend/packages/harness/deerflow/skills/AGENTS.md +++ b/backend/packages/harness/deerflow/skills/AGENTS.md @@ -21,7 +21,7 @@ Lets a caller pass per-request, short-lived end-user credentials (e.g. an ERP token) to a skill's sandbox scripts without the value entering the prompt, tool arguments, the executed command string, or traces (issue #3861). - **Declare**: a skill lists the secrets it needs in `SKILL.md` frontmatter — `required-secrets:` as a string list or `{name, optional}` mappings. `name` is both the lookup key and the env var name exposed to scripts. Parsed by `skills/parser.py::parse_required_secrets` into `Skill.required_secrets` (`SecretRequirement`); malformed entries are dropped with a warning. -- **Carry**: the caller sends values out-of-band in the run request's `context.secrets` mapping (never a message). `runtime/secret_context.py` owns the contract (`SECRETS_CONTEXT_KEY`, `extract_request_secrets`). The existing `context` passthrough carries it to `runtime.context` without mirroring into `configurable`. `build_run_config` still sets `configurable.thread_id` on the context path — the checkpointer requires it. MCP tool interceptors can read the same live carrier from `langgraph.config.get_config()["context"]["secrets"]`; see `docs/MCP_SERVER.md`. +- **Carry**: the caller sends values out-of-band in the run request's `context.secrets` mapping (never a message). `runtime/secret_context.py` owns the contract (`SECRETS_CONTEXT_KEY`, `extract_request_secrets`). The existing `context` passthrough carries it to `runtime.context` without mirroring into `configurable`. `build_run_config` still sets `configurable.thread_id` on the context path — the checkpointer requires it. MCP servers can read the same live carrier declaratively through `mcpServers..headers_from_context`, and custom MCP interceptors through `extract_request_secrets(request.runtime.context)` — **not** `langgraph.config.get_config()["context"]`, which is `None` inside a tool call because the run context rides the LangGraph runtime rather than the propagated `RunnableConfig`; see `docs/MCP_SERVER.md`. - **Admission and redaction ownership**: `services.py::start_run()` validates both legacy request mappings, `metadata.auth_token` and `config.metadata.auth_token`, before any run or thread persistence. `runtime/secret_context.py::redact_config_secrets()` also removes nested config metadata secrets from observable and persisted config copies; historical `RunResponse.kwargs` applies the same redaction non-mutatively, leaving stored `RunRecord` data unchanged. Keep callers on `config.context.secrets` rather than adding another credential carrier. Scheduled task definitions have no durable credential carrier: `ScheduledTaskService` supplies only `scheduled_task_id`, `scheduled_task_run_id`, and `scheduled_trigger` as run metadata. - **Bind (point A+)**: `SkillActivationMiddleware._resolve_secret_bindings` recomputes the injection set (`runtime.context[__active_skill_secrets]`) on every model call from two unioned sources, then REPLACES the key. (1) *Slash*: the run's most recent `/skill` activation, persisted as a source on the run context (only the activated skill's **canonical container path**, never its declared secrets) so the whole tool loop after the activation call keeps the binding; a new activation replaces it. Slash reads the genuine user text via `get_original_user_content_text`; `InputSanitizationMiddleware` preserves it (`ORIGINAL_USER_CONTENT_KEY`), so activation fires even after sanitization. (2) *In-context* (autonomous invocation): skills the model actually loaded in this thread — `ThreadState.skill_context` entries. **Both sources resolve the live registry skill by normalized container path on every call** (`_resolve_registry_skill`) and bind only that skill's own declared secrets — enabled + allowlist checked for both; the `secrets-autonomous: false` opt-out (malformed values fail closed to `false`) additionally gates the in-context path but exempts explicit slash. Resolving by registry — not by trusting the source's stored data — is what makes a caller-forged `__slash_skill_secret_source` harmless (`runtime.context` is caller-mergeable; the gateway also strips caller `__`-keys in `build_run_config`), #3938. Authorization is three-gated regardless of activation style: skill **enabled** by the operator × values **supplied per-request** by the caller (`context.secrets`) × names **declared** in frontmatter (∩ semantics). Because the set is recomputed per call, a skill evicted from `skill_context` (capacity) or a caller that stops supplying a value loses injection on the next call. The injected value always comes from the caller's request, never the host environment (scrubbed first — see below), so a declared name that also exists in the host env is safe: the caller's value wins and the host value is dropped (the #3861 per-user-key-overrides-shared-key case). Missing required secrets are logged once per binding change, not injected; binding changes are recorded as a `middleware:skill_secrets` journal event (skill and secret names only, never values). - **Inject**: `bash_tool` reads the injection set and passes it as `execute_command(env=...)`. Scope is the activation turn/run only — a run without `/skill` activation injects nothing. diff --git a/backend/tests/test_mcp_context_headers.py b/backend/tests/test_mcp_context_headers.py new file mode 100644 index 000000000..ac447be56 --- /dev/null +++ b/backend/tests/test_mcp_context_headers.py @@ -0,0 +1,896 @@ +"""Tests for request-scoped secret injection into MCP HTTP/SSE headers.""" + +import asyncio +import logging +from types import SimpleNamespace +from typing import Any +from unittest.mock import AsyncMock, patch + +import pytest +from langchain.agents import AgentState as _AgentState +from langchain_core.tools import ToolException +from langchain_mcp_adapters.interceptors import MCPToolCallRequest + +from deerflow.config.extensions_config import ( + ExtensionsConfig, + McpContextHeadersConfig, + McpServerConfig, + McpTaskToolsetConfig, + McpUserScopedAuthConfig, +) +from deerflow.mcp.context_headers import build_context_headers_interceptor +from deerflow.mcp.interceptors import build_mcp_tool_interceptors + +TENANT_TOKEN = "Bearer tenant-scoped-token" + + +def _config(**context_headers_kwargs) -> ExtensionsConfig: + return ExtensionsConfig( + mcp_servers={ + "shared-http": McpServerConfig( + enabled=True, + type="http", + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer discovery-token"}, + headers_from_context=McpContextHeadersConfig(**context_headers_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_with_secrets(**secrets: str) -> object: + return SimpleNamespace(context={"secrets": dict(secrets), "thread_id": "th-1"}) + + +async def _echo_handler(request: MCPToolCallRequest) -> MCPToolCallRequest: + return request + + +# --------------------------------------------------------------------------- +# Registration +# --------------------------------------------------------------------------- + + +def test_no_declaring_server_returns_none(): + config = ExtensionsConfig( + mcp_servers={"plain": McpServerConfig(enabled=True, type="http", url="https://x.example.com")}, + skills={}, + ) + assert build_context_headers_interceptor(config) is None + + +def test_disabled_block_returns_none(): + config = _config(headers={"X-Tenant-Token": "tenant_token"}, enabled=False) + assert build_context_headers_interceptor(config) is None + + +def test_empty_mapping_returns_none(): + """An enabled block with no mappings has nothing to inject.""" + assert build_context_headers_interceptor(_config(headers={})) is None + + +def test_disabled_server_is_ignored(): + config = _config(headers={"X-Tenant-Token": "tenant_token"}) + config.mcp_servers["shared-http"].enabled = False + assert build_context_headers_interceptor(config) is None + + +def test_stdio_server_is_skipped_with_warning(caplog): + """A stdio server has no HTTP headers; warn and skip rather than deny its calls.""" + config = ExtensionsConfig( + mcp_servers={ + "local": McpServerConfig( + enabled=True, + type="stdio", + command="npx", + headers_from_context=McpContextHeadersConfig(headers={"X-Tenant-Token": "tenant_token"}), + ) + }, + skills={}, + ) + with caplog.at_level(logging.WARNING, logger="deerflow.mcp.context_headers"): + assert build_context_headers_interceptor(config) is None + assert "stdio" in caplog.text + + +# --------------------------------------------------------------------------- +# Header injection +# --------------------------------------------------------------------------- + + +def test_request_secret_is_injected_as_header(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + result = asyncio.run(interceptor(_request(runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)), _echo_handler)) + assert result.headers["X-Tenant-Token"] == TENANT_TOKEN + + +def test_static_headers_are_preserved(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + request = _request(headers={"Accept": "application/json"}, runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result.headers == {"Accept": "application/json", "X-Tenant-Token": TENANT_TOKEN} + + +def test_context_mapping_overrides_a_static_header(): + """The per-request credential must win over the discovery credential.""" + interceptor = build_context_headers_interceptor(_config(headers={"Authorization": "tenant_token"})) + request = _request(headers={"Authorization": "Bearer discovery-token"}, runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result.headers["Authorization"] == TENANT_TOKEN + + +def test_multiple_headers_are_mapped(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Id": "tenant_id", "X-Org": "org"})) + runtime = _runtime_with_secrets(tenant_id="acme", org="engineering") + result = asyncio.run(interceptor(_request(runtime=runtime), _echo_handler)) + assert result.headers == {"X-Tenant-Id": "acme", "X-Org": "engineering"} + + +def test_request_headers_are_not_mutated_in_place(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + original = {"Accept": "application/json"} + asyncio.run(interceptor(_request(headers=original, runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)), _echo_handler)) + assert original == {"Accept": "application/json"} + + +def test_other_server_passes_through_untouched(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + request = _request(server_name="other", headers={"Authorization": "Bearer static"}, runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result is request + + +def test_falls_back_to_ambient_runtime_when_request_runtime_is_missing(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + with patch( + "deerflow.mcp.context_headers._current_runtime", + return_value=_runtime_with_secrets(tenant_token=TENANT_TOKEN), + ): + result = asyncio.run(interceptor(_request(runtime=None), _echo_handler)) + assert result.headers["X-Tenant-Token"] == TENANT_TOKEN + + +# --------------------------------------------------------------------------- +# Header-name casing +# +# HTTP field names are case-insensitive, but every dict on the path to the wire +# is case-sensitive — including the adapter's ``{**connection_headers, +# **override_headers}`` merge. A mapped name spelled differently from the static +# one would therefore travel *alongside* it rather than replacing it, and a +# server reading the field with a single-value accessor would see the static +# discovery credential first. +# --------------------------------------------------------------------------- + + +def _config_with_static_headers(static: dict[str, str], **context_headers_kwargs) -> ExtensionsConfig: + config = _config(**context_headers_kwargs) + config.mcp_servers["shared-http"].headers = static + return config + + +def test_mapped_name_is_emitted_in_the_servers_static_spelling(): + config = _config_with_static_headers({"authorization": "Bearer discovery-token"}, headers={"Authorization": "tenant_token"}) + interceptor = build_context_headers_interceptor(config) + result = asyncio.run(interceptor(_request(runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)), _echo_handler)) + assert result.headers == {"authorization": TENANT_TOKEN} + + +def test_mapped_name_replaces_a_differently_cased_header_from_an_earlier_interceptor(): + """OAuth and user_auth write before this interceptor; their value must not survive. + + The surviving spelling is whichever one is already on the request, so the + write lands on the existing field rather than adding a second one. + """ + config = _config_with_static_headers({}, headers={"authorization": "tenant_token"}) + interceptor = build_context_headers_interceptor(config) + request = _request(headers={"Authorization": "Bearer per-user"}, runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)) + result = asyncio.run(interceptor(request, _echo_handler)) + assert list(result.headers.values()) == [TENANT_TOKEN] + + +def test_unrelated_headers_keep_their_own_spelling(): + config = _config_with_static_headers({"Authorization": "Bearer discovery-token"}, headers={"X-Tenant-Token": "tenant_token"}) + interceptor = build_context_headers_interceptor(config) + request = _request(headers={"Accept": "application/json"}, runtime=_runtime_with_secrets(tenant_token=TENANT_TOKEN)) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result.headers == {"Accept": "application/json", "X-Tenant-Token": TENANT_TOKEN} + + +def _connection_headers_for_adapter_call(config: ExtensionsConfig) -> dict[str, str]: + """Return the headers the adapter would open the remote session with. + + Goes through the real connection merge rather than seeding static headers + onto ``request.headers``: the adapter builds the request with + ``headers=None``, so an interceptor never sees the connection's static + headers and the collision can only be observed here. + """ + from langchain_core.messages import AIMessage + from langchain_mcp_adapters.tools import convert_mcp_tool_to_langchain_tool + from langgraph.graph import END, START, StateGraph + from langgraph.prebuilt import ToolNode + from mcp.types import CallToolResult, TextContent + from mcp.types import Tool as MCPTool + + from deerflow.mcp.client import build_server_params + + opened: dict[str, str] = {} + + class _Session: + async def initialize(self) -> None: + return None + + async def call_tool(self, *_args, **_kwargs): + return CallToolResult(content=[TextContent(type="text", text="done")], isError=False) + + class _SessionContext: + def __init__(self, connection, **_kwargs): + opened.update(connection.get("headers") or {}) + + async def __aenter__(self): + return _Session() + + async def __aexit__(self, *_exc): + return False + + tool = convert_mcp_tool_to_langchain_tool( + None, + MCPTool(name="act", description="act", inputSchema={"type": "object", "properties": {}}), + connection=build_server_params("shared-http", config.mcp_servers["shared-http"]), + server_name="shared-http", + tool_interceptors=build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: None), + ) + + builder = StateGraph(_AgentState, context_schema=dict) + builder.add_node("tools", ToolNode([tool])) + builder.add_edge(START, "tools") + builder.add_edge("tools", END) + graph = builder.compile() + + with patch("langchain_mcp_adapters.tools.create_session", _SessionContext): + asyncio.run( + graph.ainvoke( + {"messages": [AIMessage(content="", tool_calls=[{"name": "act", "args": {}, "id": "call_1", "type": "tool_call"}])]}, + context={"secrets": {"tenant_token": TENANT_TOKEN}, "thread_id": "th-1"}, + ) + ) + return opened + + +def test_connection_carries_one_authorization_header_despite_a_casing_mismatch(): + """The reviewed failure: two spellings both reach httpx, static one first.""" + config = _config_with_static_headers({"authorization": "Bearer discovery-token"}, headers={"Authorization": "tenant_token"}) + opened = _connection_headers_for_adapter_call(config) + assert [name for name in opened if name.lower() == "authorization"] == ["authorization"] + assert opened["authorization"] == TENANT_TOKEN + + +def test_connection_keeps_static_headers_the_mapping_does_not_touch(): + config = _config_with_static_headers({"Authorization": "Bearer discovery-token", "X-Api-Version": "2"}, headers={"X-Tenant-Token": "tenant_token"}) + opened = _connection_headers_for_adapter_call(config) + assert opened == {"Authorization": "Bearer discovery-token", "X-Api-Version": "2", "X-Tenant-Token": TENANT_TOKEN} + + +def test_mapping_the_same_header_under_two_spellings_is_rejected(): + with pytest.raises(ValueError, match="two spellings"): + McpContextHeadersConfig(headers={"Authorization": "tenant_token", "authorization": "other_token"}) + + +def test_gateway_rejects_the_same_header_under_two_spellings(): + from pydantic import ValidationError + + from app.gateway.routers.mcp import McpContextHeadersConfigResponse + + with pytest.raises(ValidationError, match="two spellings"): + McpContextHeadersConfigResponse(headers={"Authorization": "tenant_token", "AUTHORIZATION": "other_token"}) + + +# --------------------------------------------------------------------------- +# Fail-closed behaviour +# --------------------------------------------------------------------------- + + +def test_missing_secret_denies_without_calling_handler(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + handler = AsyncMock() + with pytest.raises(ToolException, match="tenant_token"): + asyncio.run(interceptor(_request(runtime=_runtime_with_secrets(unrelated="x")), handler)) + handler.assert_not_awaited() + + +def test_empty_secret_value_is_denied(): + """An unset $ENV_VAR on the caller side arrives as "" and must fail closed.""" + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + with pytest.raises(ToolException): + asyncio.run(interceptor(_request(runtime=_runtime_with_secrets(tenant_token="")), AsyncMock())) + + +def test_absent_run_context_is_denied(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + with patch("deerflow.mcp.context_headers._current_runtime", return_value=None), pytest.raises(ToolException): + asyncio.run(interceptor(_request(runtime=None), AsyncMock())) + + +def test_deny_message_does_not_leak_other_secret_values(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})) + runtime = _runtime_with_secrets(other_secret="super-secret-value") + with pytest.raises(ToolException) as excinfo: + asyncio.run(interceptor(_request(runtime=runtime), AsyncMock())) + assert "super-secret-value" not in str(excinfo.value) + + +def test_on_missing_passthrough_keeps_static_headers(): + interceptor = build_context_headers_interceptor(_config(headers={"Authorization": "tenant_token"}, on_missing="passthrough")) + request = _request(headers={"Authorization": "Bearer discovery-token"}, runtime=_runtime_with_secrets()) + result = asyncio.run(interceptor(request, _echo_handler)) + assert result.headers["Authorization"] == "Bearer discovery-token" + + +def test_passthrough_still_injects_the_secrets_that_are_present(): + interceptor = build_context_headers_interceptor(_config(headers={"X-Tenant-Id": "tenant_id", "X-Org": "org"}, on_missing="passthrough")) + result = asyncio.run(interceptor(_request(runtime=_runtime_with_secrets(tenant_id="acme")), _echo_handler)) + assert result.headers == {"X-Tenant-Id": "acme"} + + +# --------------------------------------------------------------------------- +# Config model +# --------------------------------------------------------------------------- + + +def test_blank_header_name_is_rejected(): + with pytest.raises(ValueError, match="header name"): + McpContextHeadersConfig(headers={" ": "tenant_token"}) + + +def test_blank_secret_key_is_rejected(): + with pytest.raises(ValueError, match="secret key"): + McpContextHeadersConfig(headers={"X-Tenant-Token": ""}) + + +def test_config_round_trips_from_file(tmp_path): + config_file = tmp_path / "extensions_config.json" + config_file.write_text( + """ + { + "mcpServers": { + "shared-http": { + "enabled": true, + "transport": "http", + "url": "https://mcp.example.com/mcp", + "headers_from_context": {"headers": {"X-Tenant-Token": "tenant_token"}} + } + } + } + """ + ) + config = ExtensionsConfig.from_file(str(config_file)) + block = config.mcp_servers["shared-http"].headers_from_context + assert block is not None + assert block.enabled is True + assert block.on_missing == "deny" + assert block.headers == {"X-Tenant-Token": "tenant_token"} + + +def test_mapping_values_are_not_env_resolved(tmp_path, monkeypatch): + """The right-hand side names a run-context key, not an environment variable.""" + monkeypatch.setenv("tenant_token", "must-not-be-substituted") + config_file = tmp_path / "extensions_config.json" + config_file.write_text( + """ + { + "mcpServers": { + "shared-http": { + "enabled": true, + "transport": "http", + "url": "https://mcp.example.com/mcp", + "headers_from_context": {"headers": {"X-Tenant-Token": "tenant_token"}} + } + } + } + """ + ) + config = ExtensionsConfig.from_file(str(config_file)) + assert config.mcp_servers["shared-http"].headers_from_context.headers == {"X-Tenant-Token": "tenant_token"} + + +# --------------------------------------------------------------------------- +# Interceptor chain assembly +# --------------------------------------------------------------------------- + + +def test_registered_last_so_request_secrets_win(): + """Later interceptors run closer to the transport, so per-request values win.""" + config = _config(headers={"Authorization": "tenant_token"}) + config.mcp_servers["shared-http"].user_auth = McpUserScopedAuthConfig(users={"u1": "Bearer per-user"}) + + 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 [getattr(i, "__name__", type(i).__name__) for i in interceptors] == [ + "oauth", + "user_scoped_auth_interceptor", + "context_headers_interceptor", + ] + + +def test_shared_assembly_skips_when_not_configured(): + config = ExtensionsConfig( + mcp_servers={"plain": McpServerConfig(enabled=True, type="http", url="https://x.example.com")}, + skills={}, + ) + assert build_mcp_tool_interceptors(config, oauth_builder=lambda _cfg: None) == [] + + +# --------------------------------------------------------------------------- +# End-to-end contract with LangGraph + langchain-mcp-adapters +# --------------------------------------------------------------------------- + + +def _run_adapter_tool_in_graph(*, isolate_request_runtime: bool = False) -> dict[str, Any]: + """Drive a real adapter tool through a real graph; return the headers it sent. + + DeerFlow does not wrap HTTP/SSE MCP tools, so the tool under test here is the + one ``langchain_mcp_adapters`` builds, invoked by LangGraph's own tool node. + + With *isolate_request_runtime* the ambient-runtime fallback is disabled, so + the secrets can only arrive through the runtime LangGraph injected into the + adapter tool's ``runtime`` parameter. + """ + from langchain_core.messages import AIMessage + from langchain_mcp_adapters.tools import convert_mcp_tool_to_langchain_tool + from langgraph.graph import END, START, StateGraph + from langgraph.prebuilt import ToolNode + from mcp.types import CallToolResult, TextContent + from mcp.types import Tool as MCPTool + + seen_headers: dict[str, Any] = {} + + class _FakeSession: + async def call_tool(self, name, args, **kwargs): + return CallToolResult(content=[TextContent(type="text", text="done")], isError=False) + + async def _capture_headers(request, handler): + seen_headers.update(request.headers or {}) + return await handler(request) + + tool = convert_mcp_tool_to_langchain_tool( + _FakeSession(), + MCPTool(name="act", description="act", inputSchema={"type": "object", "properties": {}}), + server_name="shared-http", + tool_interceptors=[ + build_context_headers_interceptor(_config(headers={"X-Tenant-Token": "tenant_token"})), + _capture_headers, + ], + ) + + builder = StateGraph(_AgentState, context_schema=dict) + builder.add_node("tools", ToolNode([tool])) + builder.add_edge(START, "tools") + builder.add_edge("tools", END) + graph = builder.compile() + + def _invoke() -> None: + asyncio.run( + graph.ainvoke( + {"messages": [AIMessage(content="", tool_calls=[{"name": "act", "args": {}, "id": "call_1", "type": "tool_call"}])]}, + context={"secrets": {"tenant_token": TENANT_TOKEN}, "thread_id": "th-1"}, + ) + ) + + if isolate_request_runtime: + with patch("deerflow.mcp.context_headers._current_runtime", return_value=None): + _invoke() + else: + _invoke() + return seen_headers + + +def test_request_secret_reaches_a_real_adapter_tool_call(): + """The user-facing contract: a per-request secret lands on the outgoing call.""" + assert _run_adapter_tool_in_graph().get("X-Tenant-Token") == TENANT_TOKEN + + +def test_adapter_tool_receives_the_runtime_langgraph_injects(): + """Pin the injection rule the HTTP/SSE path depends on. + + ``langchain_mcp_adapters`` names its tool parameter ``runtime``, and + LangGraph's tool node injects a ``ToolRuntime`` into any parameter with that + name. With the ambient-runtime fallback disabled, that channel is the only + way the secrets can arrive — so an upstream rename or a change to the + injection rule fails here instead of silently dropping every header. + """ + assert _run_adapter_tool_in_graph(isolate_request_runtime=True).get("X-Tenant-Token") == TENANT_TOKEN + + +# --------------------------------------------------------------------------- +# Durable background tasks +# --------------------------------------------------------------------------- + + +def _task_config() -> ExtensionsConfig: + return ExtensionsConfig.model_validate( + { + "mcpServers": { + "reports": { + "type": "http", + "url": "https://reports.example.com/mcp", + "headers": {"Authorization": "Bearer discovery-token"}, + "headers_from_context": {"headers": {"Authorization": "tenant_token"}}, + "task_toolsets": [{"name": "reports", "submit_tool": "submit", "status_tool": "status", "cancel_tool": "cancel"}], + } + } + } + ) + + +def _task_caller(config: ExtensionsConfig) -> tuple[Any, dict[str, str], Any]: + """Build a task caller whose remote session records the headers it opened with.""" + from deerflow.mcp.task_tool_caller import McpTaskToolCaller + + opened: dict[str, str] = {} + result = SimpleNamespace(structuredContent={"task_id": "remote-1", "status": "running"}, isError=False) + + class _SessionContext: + def __init__(self, connection, **_kwargs): + opened.clear() + opened.update(connection.get("headers") or {}) + + async def __aenter__(self): + return SimpleNamespace(initialize=AsyncMock(), call_tool=AsyncMock(return_value=result)) + + async def __aexit__(self, *_exc): + return False + + caller = McpTaskToolCaller( + config, + oauth_token_manager=SimpleNamespace(has_oauth_servers=lambda: False, get_authorization_header=AsyncMock(return_value=None)), + ) + return caller, opened, _SessionContext + + +_DRIVER_DATA = {"submit_tool": "submit", "status_tool": "status", "cancel_tool": "cancel"} + + +def test_durable_submit_carries_the_request_scoped_headers(): + """Submit is awaited inside the Agent run, so it can — and must — carry them. + + Driven through a real tool node with no ambient-runtime patching: the run + context reaches the driver through the contextvar LangGraph sets around the + tool coroutine, several awaits below it. + """ + from langchain_core.messages import AIMessage + from langchain_core.tools import tool as make_tool + from langgraph.graph import END, START, StateGraph + from langgraph.prebuilt import ToolNode + + from deerflow.mcp.tasks import TaskSubmitRequest + from deerflow.mcp.tasks.ordinary import OrdinaryMcpTaskDriver + + caller, opened, session_context = _task_caller(_task_config()) + driver = OrdinaryMcpTaskDriver(caller) + + @make_tool + async def submit_report() -> str: + """Submit a durable report task.""" + await driver.submit( + TaskSubmitRequest( + user_id="user-1", + thread_id="thread-1", + run_id=None, + tool_call_id=None, + server_name="reports", + task_name="reports", + arguments={}, + driver_data=dict(_DRIVER_DATA), + ) + ) + return "submitted" + + builder = StateGraph(_AgentState, context_schema=dict) + builder.add_node("tools", ToolNode([submit_report])) + builder.add_edge(START, "tools") + builder.add_edge("tools", END) + graph = builder.compile() + + with patch("langchain_mcp_adapters.sessions.create_session", session_context): + asyncio.run( + graph.ainvoke( + {"messages": [AIMessage(content="", tool_calls=[{"name": "submit_report", "args": {}, "id": "call_1", "type": "tool_call"}])]}, + context={"secrets": {"tenant_token": TENANT_TOKEN}, "thread_id": "thread-1"}, + ) + ) + + assert opened == {"Authorization": TENANT_TOKEN} + + +@pytest.mark.asyncio +async def test_durable_status_poll_keeps_the_server_credential(): + """The poller runs after the Agent run ended: no run context, no deny.""" + from deerflow.mcp.tasks.models import TaskReference + from deerflow.mcp.tasks.ordinary import OrdinaryMcpTaskDriver + + caller, opened, session_context = _task_caller(_task_config()) + driver = OrdinaryMcpTaskDriver(caller) + + with patch("langchain_mcp_adapters.sessions.create_session", session_context): + snapshot = await driver.get_status( + TaskReference( + local_task_id="local-1", + user_id="user-1", + thread_id="thread-1", + server_name="reports", + remote_task_id="remote-1", + driver_data=dict(_DRIVER_DATA), + ) + ) + + assert snapshot is not None + assert opened == {"Authorization": "Bearer discovery-token"} + + +def test_declaring_both_request_headers_and_task_toolsets_warns(caplog): + """Background polls run outside the Agent run that carried the secrets.""" + config = _config(headers={"X-Tenant-Token": "tenant_token"}) + config.mcp_servers["shared-http"].task_toolsets = [McpTaskToolsetConfig(name="reports", submit_tool="submit", status_tool="status", cancel_tool="cancel")] + with caplog.at_level(logging.WARNING, logger="deerflow.mcp.context_headers"): + assert build_context_headers_interceptor(config) is not None + assert "task_toolsets" in caplog.text + + +@pytest.mark.asyncio +async def test_durable_task_calls_are_not_denied_for_a_missing_run_context(): + """The task runtime must keep polling on server-level auth, not fail closed.""" + from unittest.mock import MagicMock + + from deerflow.mcp.task_tool_caller import McpTaskToolCaller + + config = ExtensionsConfig.model_validate( + { + "mcpServers": { + "reports": { + "type": "http", + "url": "https://reports.example.com/mcp", + "headers": {"X-Static": "configured"}, + "headers_from_context": {"headers": {"X-Tenant-Token": "tenant_token"}}, + } + } + } + ) + result = SimpleNamespace(structuredContent={"task_id": "remote-1", "status": "running"}, isError=False) + session = SimpleNamespace(initialize=AsyncMock(), call_tool=AsyncMock(return_value=result)) + + class _SessionContext: + async def __aenter__(self): + return session + + async def __aexit__(self, *_exc): + return False + + caller = McpTaskToolCaller( + config, + oauth_token_manager=SimpleNamespace(has_oauth_servers=lambda: False, get_authorization_header=AsyncMock(return_value=None)), + ) + + with patch("langchain_mcp_adapters.sessions.create_session", MagicMock(return_value=_SessionContext())): + actual = await caller.call_tool( + server_name="reports", + tool_name="status", + arguments={"task_id": "remote-1"}, + user_id="user-1", + thread_id="thread-1", + ) + + assert actual is result + + +# --------------------------------------------------------------------------- +# Gateway API surface +# --------------------------------------------------------------------------- + + +def test_gateway_exposes_mapping_without_masking(): + """The block holds header names and run-context key names, never a credential.""" + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _mask_server_config, + ) + + server = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}), + ) + masked = _mask_server_config(server) + assert masked.headers_from_context.headers == {"X-Tenant-Token": "tenant_token"} + + +def test_gateway_masks_sensitive_extras_inside_the_block(): + """``extra="allow"`` means an operator can still store a secret-bearing key here.""" + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _mask_server_config, + ) + + server = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}, api_key="real-secret"), + ) + masked = _mask_server_config(server) + assert masked.headers_from_context.model_extra["api_key"] == "***" + assert masked.headers_from_context.headers == {"X-Tenant-Token": "tenant_token"} + + +def test_gateway_merge_preserves_block_when_field_omitted(): + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}), + ) + incoming = McpServerConfigResponse(type="http", url="https://mcp.example.com/mcp") + merged = _merge_preserving_secrets(incoming, existing) + assert merged.headers_from_context is not None + assert merged.headers_from_context.headers == {"X-Tenant-Token": "tenant_token"} + + +def test_gateway_put_can_replace_the_mapping(): + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Org": "org"}, on_missing="passthrough"), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.headers_from_context.headers == {"X-Org": "org"} + assert merged.headers_from_context.on_missing == "passthrough" + + +def test_gateway_partial_block_preserves_stored_mapping_and_policy(): + """A partial headers_from_context PUT must not wipe omitted declared fields.""" + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse( + headers={"X-Tenant-Token": "tenant_token"}, + on_missing="passthrough", + ), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(enabled=False), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.headers_from_context is not None + assert merged.headers_from_context.enabled is False + assert merged.headers_from_context.headers == {"X-Tenant-Token": "tenant_token"} + assert merged.headers_from_context.on_missing == "passthrough" + + +def test_gateway_partial_block_explicit_empty_mapping_still_clears(): + """An explicitly supplied empty mapping must clear the stored mapping, not preserve it.""" + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse( + headers={"X-Tenant-Token": "tenant_token"}, + on_missing="passthrough", + ), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={}), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.headers_from_context is not None + assert merged.headers_from_context.headers == {} + assert merged.headers_from_context.on_missing == "passthrough" + + +def test_gateway_round_trip_restores_masked_extras_inside_the_block(): + """GET masks the block's extras, so PUT must swap the sentinel back.""" + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _mask_server_config, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}, api_key="real-secret", note="kept"), + ) + merged = _merge_preserving_secrets(_mask_server_config(existing), existing) + assert merged.headers_from_context.model_extra["api_key"] == "real-secret" + assert merged.headers_from_context.model_extra["note"] == "kept" + + +def test_gateway_keeps_block_extras_a_put_does_not_mention(): + """Matches how user_auth and server-level extras survive a partial PUT.""" + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}, vendor_note="keep-me"), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Org": "org"}), + ) + merged = _merge_preserving_secrets(incoming, existing) + assert merged.headers_from_context.headers == {"X-Org": "org"} + assert merged.headers_from_context.model_extra["vendor_note"] == "keep-me" + + +def test_gateway_rejects_a_masked_value_for_an_unknown_block_extra(): + """A sentinel with nothing stored behind it must not be written to disk.""" + from fastapi import HTTPException + + from app.gateway.routers.mcp import ( + McpContextHeadersConfigResponse, + McpServerConfigResponse, + _merge_preserving_secrets, + ) + + existing = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}), + ) + incoming = McpServerConfigResponse( + type="http", + url="https://mcp.example.com/mcp", + headers_from_context=McpContextHeadersConfigResponse(headers={"X-Tenant-Token": "tenant_token"}, api_key="***"), + ) + with pytest.raises(HTTPException): + _merge_preserving_secrets(incoming, existing) diff --git a/backend/tests/test_mcp_header_names.py b/backend/tests/test_mcp_header_names.py new file mode 100644 index 000000000..ca7a5a788 --- /dev/null +++ b/backend/tests/test_mcp_header_names.py @@ -0,0 +1,202 @@ +"""Tests for case-insensitive header writes shared by the MCP interceptors. + +HTTP field names are case-insensitive, but the dicts carrying them are not: a +static ``authorization`` and an injected ``Authorization`` are two keys, both +reach httpx, and a server reading the field with a single-value accessor gets +the *static* one — the credential the injection was meant to replace. +""" + +import asyncio +from types import SimpleNamespace +from unittest.mock import AsyncMock, patch + +import pytest +from langchain_mcp_adapters.interceptors import MCPToolCallRequest +from pydantic import ValidationError + +from deerflow.config.extensions_config import ExtensionsConfig, McpServerConfig, McpUserScopedAuthConfig +from deerflow.mcp.headers import apply_header_overrides, header_spellings +from deerflow.mcp.oauth import build_oauth_tool_interceptor +from deerflow.mcp.user_scoped_auth import build_user_scoped_auth_interceptor + +DISCOVERY = "Bearer discovery-token" + + +def _request(headers: dict | None = None, runtime: object | None = None, server_name: str = "shared-http") -> MCPToolCallRequest: + return MCPToolCallRequest(name="act", args={}, server_name=server_name, headers=headers, runtime=runtime) + + +async def _echo_handler(request: MCPToolCallRequest) -> MCPToolCallRequest: + return request + + +# --------------------------------------------------------------------------- +# apply_header_overrides +# --------------------------------------------------------------------------- + + +def test_override_replaces_a_differently_cased_key(): + assert apply_header_overrides({"Authorization": DISCOVERY}, {"authorization": "Bearer new"}) == {"Authorization": "Bearer new"} + + +def test_override_prefers_the_connection_spelling(): + """The emitted name must match the connection's, since the adapter merges by key.""" + merged = apply_header_overrides( + {"Authorization": "Bearer from-an-earlier-interceptor"}, + {"AUTHORIZATION": "Bearer new"}, + spellings=header_spellings(["authorization"]), + ) + assert merged == {"authorization": "Bearer new"} + + +def test_override_keeps_unrelated_headers(): + merged = apply_header_overrides({"Accept": "application/json"}, {"X-Tenant-Id": "acme"}) + assert merged == {"Accept": "application/json", "X-Tenant-Id": "acme"} + + +def test_override_accepts_no_base(): + assert apply_header_overrides(None, {"X-Tenant-Id": "acme"}) == {"X-Tenant-Id": "acme"} + + +def test_override_does_not_mutate_the_base(): + base = {"Authorization": DISCOVERY} + apply_header_overrides(base, {"authorization": "Bearer new"}) + assert base == {"Authorization": DISCOVERY} + + +# --------------------------------------------------------------------------- +# Static header spelling validation +# --------------------------------------------------------------------------- + + +def test_static_headers_reject_case_insensitive_duplicates(): + """Two spellings of one header in the static map must be rejected at config time.""" + with pytest.raises(ValueError, match="two spellings"): + McpServerConfig( + type="http", + url="https://mcp.example.com/mcp", + headers={"Authorization": "Bearer a", "authorization": "Bearer b"}, + ) + + +def test_static_headers_allow_distinct_names(): + config = McpServerConfig( + type="http", + url="https://mcp.example.com/mcp", + headers={"X-Tenant": "acme", "X-Org": "engineering"}, + ) + assert config.headers == {"X-Tenant": "acme", "X-Org": "engineering"} + + +def test_extensions_config_rejects_static_header_duplicates(): + with pytest.raises(ValidationError, match="two spellings"): + ExtensionsConfig.model_validate( + { + "mcpServers": { + "shared-http": { + "type": "http", + "url": "https://mcp.example.com/mcp", + "headers": {"Authorization": "Bearer a", "authorization": "Bearer b"}, + } + } + } + ) + + +def test_gateway_rejects_static_header_case_insensitive_duplicates(): + from app.gateway.routers.mcp import McpServerConfigResponse + + with pytest.raises(ValidationError, match="two spellings"): + McpServerConfigResponse(headers={"Authorization": "Bearer a", "AUTHORIZATION": "Bearer b"}) + + +# --------------------------------------------------------------------------- +# The credential interceptors +# --------------------------------------------------------------------------- + + +def _server(**overrides) -> ExtensionsConfig: + return ExtensionsConfig( + mcp_servers={ + "shared-http": McpServerConfig( + enabled=True, + type="http", + url="https://mcp.example.com/mcp", + headers={"authorization": DISCOVERY}, + **overrides, + ) + }, + skills={}, + ) + + +def test_user_auth_credential_replaces_a_differently_cased_static_header(): + config = _server(user_auth=McpUserScopedAuthConfig(header="Authorization", users={"u1": "Bearer per-user"})) + interceptor = build_user_scoped_auth_interceptor(config) + runtime = SimpleNamespace(server_info=None, context={"user_id": "u1"}) + result = asyncio.run(interceptor(_request(runtime=runtime), _echo_handler)) + assert result.headers == {"authorization": "Bearer per-user"} + + +def test_oauth_token_replaces_a_differently_cased_static_header(): + config = _server( + oauth={ + "enabled": True, + "token_url": "https://auth.example.com/oauth/token", + "client_id": "id", + "client_secret": "secret", + } + ) + token_manager = SimpleNamespace( + has_oauth_servers=lambda: True, + get_authorization_header=AsyncMock(return_value="Bearer oauth-token"), + ) + interceptor = build_oauth_tool_interceptor(config, token_manager=token_manager) + result = asyncio.run(interceptor(_request(), _echo_handler)) + assert result.headers == {"authorization": "Bearer oauth-token"} + + +@pytest.mark.asyncio +async def test_durable_task_call_sends_one_authorization_header(): + """The task caller merges OAuth and interceptor headers into the connection itself.""" + from deerflow.mcp.task_tool_caller import McpTaskToolCaller + + config = ExtensionsConfig.model_validate( + { + "mcpServers": { + "reports": { + "type": "http", + "url": "https://reports.example.com/mcp", + "headers": {"authorization": DISCOVERY}, + } + } + } + ) + opened: dict[str, str] = {} + result = SimpleNamespace(structuredContent={"task_id": "remote-1", "status": "running"}, isError=False) + + class _SessionContext: + def __init__(self, connection, **_kwargs): + opened.update(connection.get("headers") or {}) + + async def __aenter__(self): + return SimpleNamespace(initialize=AsyncMock(), call_tool=AsyncMock(return_value=result)) + + async def __aexit__(self, *_exc): + return False + + caller = McpTaskToolCaller( + config, + oauth_token_manager=SimpleNamespace(has_oauth_servers=lambda: False, get_authorization_header=AsyncMock(return_value="Bearer oauth-token")), + ) + + with patch("langchain_mcp_adapters.sessions.create_session", _SessionContext): + await caller.call_tool( + server_name="reports", + tool_name="status", + arguments={"task_id": "remote-1"}, + user_id="user-1", + thread_id="thread-1", + ) + + assert opened == {"authorization": "Bearer oauth-token"} diff --git a/backend/tests/test_mcp_task_ordinary_driver.py b/backend/tests/test_mcp_task_ordinary_driver.py index 53be3fd20..25ec5532b 100644 --- a/backend/tests/test_mcp_task_ordinary_driver.py +++ b/backend/tests/test_mcp_task_ordinary_driver.py @@ -74,6 +74,9 @@ async def test_submit_uses_structured_content_and_keeps_remote_id_out_of_driver_ "arguments": {"topic": "MCP"}, "user_id": "user-1", "thread_id": "thread-1", + # 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, } ] diff --git a/frontend/src/content/en/harness/mcp.mdx b/frontend/src/content/en/harness/mcp.mdx index c51f34e41..a262047b2 100644 --- a/frontend/src/content/en/harness/mcp.mdx +++ b/frontend/src/content/en/harness/mcp.mdx @@ -154,6 +154,68 @@ 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`. +## Per-request credentials + +`user_auth` binds a credential to a *configured* DeerFlow user. When the caller +picks the credential at request time instead — a multi-tenant gateway, a per-run +API key, one shared MCP server fronting several environments — declare a +`headers_from_context` block. Each entry maps an HTTP header name to a key of +the run request's `config.context.secrets` carrier: + +```json +{ + "mcpServers": { + "shared-api": { + "type": "http", + "url": "https://api.example.com/mcp", + "headers": { "Authorization": "$SERVICE_DISCOVERY_TOKEN" }, + "headers_from_context": { + "headers": { + "X-Tenant-Id": "tenant_id", + "Authorization": "tenant_token" + }, + "on_missing": "deny" + } + } + } +} +``` + +The caller supplies the values with each run request, out of band from the +conversation: + +```json +{ + "config": { + "context": { + "secrets": { + "tenant_id": "acme", + "tenant_token": "Bearer " + } + } + } +} +``` + +- The config file stores **names only**, never a credential, so the block is + returned unmasked by the config API. The values travel with each run and are + stripped from persisted run configuration, API responses, and traces. +- The entry's static `headers` are used only for startup tool discovery; a + mapped header replaces the static one for the tool call. Header names are + matched case-insensitively, so a mapped `Authorization` replaces a static + `authorization` rather than travelling alongside it. +- **Fail-closed by default**: if the run carries no value for a mapped key, the + call fails with an error naming the missing key rather than falling back to + the discovery credential — which would send one tenant's request under + another tenant's authority. Set `"on_missing": "passthrough"` to opt out. +- For a server declaring several sources, precedence is static `headers` < + `oauth` < `user_auth` < `headers_from_context`: the value chosen for this one + request is the most specific, so it wins. +- On a server that also declares `task_toolsets`, the durable submit is awaited + inside the run and carries these headers, but the later status and cancel + polls run after the run ends and fall back to the server's own credentials — + so the fail-closed guarantee covers the submit, not those polls. + ## Managing MCP servers MCP servers can be managed in several ways: