mirror of
https://github.com/bytedance/deer-flow.git
synced 2026-09-10 05:58:36 +00:00
feat(mcp): map request-scoped secrets to MCP HTTP/SSE headers (#5010)
* feat(mcp): map request-scoped secrets to HTTP/SSE headers `user_auth` binds a credential to a configured DeerFlow user, so a caller that picks the credential per request — a multi-tenant gateway, a per-run API key, one shared MCP server fronting several environments — had to register one MCP server entry per credential. Add a declarative `mcpServers.<server>.headers_from_context` block mapping HTTP header names to keys of the run request's `config.context.secrets` carrier. A new built-in interceptor resolves the mapping on every tool call and rewrites those headers, mirroring `user_scoped_auth`. The config file stores names only, never a credential, so the Gateway returns the block unmasked. Registered after OAuth and `user_auth` in the interceptor chain: the later interceptor runs closer to the transport, and the value chosen for this one request is the most specific, so it wins. Fail-closed by default — a mapped key missing from the request raises a `ToolException` naming only that key, because falling back to the server's discovery credential would send one tenant's call under another tenant's authority. `on_missing: "passthrough"` opts out. Durable background tasks are excluded: `McpTaskToolCaller` drives status and cancel polls after the Agent run ends, where no run context exists, so the fail-closed interceptor would deny every poll. Those calls keep using server-level credentials, and a server declaring both `headers_from_context` and `task_toolsets` now logs a warning. Also corrects the custom-interceptor example in docs/MCP_SERVER.md (and the matching claim in skills/AGENTS.md), which read request secrets from `langgraph.config.get_config()["context"]`. That key is `None` inside a tool call — the run context rides the LangGraph runtime, not the RunnableConfig propagated to child runnables — so interceptors written from that example never saw a value. The example now reads `request.runtime`, and tests/test_mcp_context_headers.py pins LangGraph's runtime-injection rule by driving a real langchain-mcp-adapters tool through a real graph with the ambient-runtime fallback disabled. Closes #5005 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(mcp): resolve credential headers case-insensitively, carry them on durable submit Review follow-ups on `headers_from_context`. HTTP field names are case-insensitive, but every dict on the path to the wire is not: `build_server_params` copies the operator's static `headers` spelling verbatim, and langchain-mcp-adapters merges interceptor overrides into the connection with a plain `{**connection_headers, **override_headers}` splat. A static `authorization` and an injected `Authorization` therefore both reached httpx as separate field lines, and a server reading the field with a single-value accessor got the static discovery credential — inverting the documented `headers` < `oauth` < `user_auth` < `headers_from_context` precedence and running a per-request call under the shared credential. Normalizing inside the interceptor cannot fix that on its own: the adapter builds the request with `headers=None`, so an interceptor never sees the connection's static headers and cannot displace them however it spells its own key. A new `mcp/headers.py::apply_header_overrides` therefore drops any key differing only in case and emits the spelling the connection already uses. Applied to `headers_from_context`, `user_auth`, the OAuth interceptor, the OAuth discovery-header write, and the durable-task connection merge, which all carried the same collision. `headers_from_context.headers` now also rejects one header mapped under two spellings at config load, in both the harness model and the Gateway mirror. Durable submit now carries the mapped headers, as docs/MCP_SERVER.md already promised. `McpTaskToolCaller` disabled the interceptor for the whole caller, but that caller serves submit as well as the polls, and submit is awaited inline inside the Agent's tool call — where the run's LangGraph runtime is still the ambient contextvar, so no secret has to be threaded through `TaskSubmitRequest` or reach durable storage. The caller builds one chain and keeps a second view of it without the context-headers interceptor; `call_tool` takes `request_scoped_headers`, set only by `OrdinaryMcpTaskDriver.submit`. Status and cancel keep server-level credentials, so background polls still cannot fail closed, and the startup warning now describes the half it actually covers. `_merge_preserving_secrets` restores masked extras inside `headers_from_context` instead of writing the `***` sentinel back over the stored value, matching the treatment `user_auth` extras and server-level extras already get; extras a PUT omits carry over as well, while the declared mapping still replaces verbatim so a round trip can remove an entry. `extra="allow"` plus name-based sensitivity detection means the usual casualty is a name-valued key such as `tokenHeader`, not only a credential. The existing override test seeded the static header onto `request.headers`, which production never does, so it modelled a merge that really happens one layer down; the new tests drive a real adapter tool through a real connection and assert on the headers the session is opened with, and the durable-submit test runs through a real tool node with no runtime patching. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com> * fix(mcp): reject case-insensitive duplicate static header names * fix(mcp): preserve omitted headers_from_context fields on partial updates --------- Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
This commit is contained in:
青榆牧
GitHub
parent
cb24bc2699
commit
a94b2d8897
@ -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:
|
||||
|
||||
@ -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 <request-scoped credential>"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- 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
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -11,6 +11,9 @@
|
||||
- **Per-server tool-name prefixing**: `mcpServers.<server>.tool_name_prefix` defaults to `true`, preserving the collision-safe `<server_name>_` prefix. Servers whose tools already carry a stable namespace may set it to `false`; discovery then calls `langchain_mcp_adapters.tools.load_mcp_tools` with that server's flag. Source routing and stdio session-pool wrapping are based on the producing server and transport, never on whether the visible tool name starts with the server prefix.
|
||||
- **OAuth (HTTP/SSE)**: Supports token endpoint flows (`client_credentials`, `refresh_token`) with automatic token refresh + Authorization header injection
|
||||
- **Per-user credentials (HTTP/SSE)**: `mcpServers.<server>.user_auth` maps DeerFlow user ids to credential header values (`$ENV_VAR` references supported). `mcp/user_scoped_auth.py::build_user_scoped_auth_interceptor` rewrites the configured header on every tool call from the authenticated runtime user (registered after OAuth in `mcp/interceptors.py`, so its per-user value wins the header for servers declaring both). Fail-closed: an unmapped user or an empty resolved credential raises a `ToolException` unless `on_missing: "passthrough"` is set. The server's static `headers` serve startup tool discovery only. Gateway GET masks `user_auth.users` values; PUT round-trips masked values by preserving stored credentials.
|
||||
- **Per-request credentials (HTTP/SSE)**: `mcpServers.<server>.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.<server>.routing` and
|
||||
`tools.<original_tool_name>.routing` are soft preference metadata. The effective
|
||||
routing is resolved while `mcp/tools.py::get_mcp_tools()` still has both
|
||||
|
||||
162
backend/packages/harness/deerflow/mcp/context_headers.py
Normal file
162
backend/packages/harness/deerflow/mcp/context_headers.py
Normal file
@ -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
|
||||
61
backend/packages/harness/deerflow/mcp/headers.py
Normal file
61
backend/packages/harness/deerflow/mcp/headers.py
Normal file
@ -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
|
||||
@ -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]
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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:
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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,
|
||||
|
||||
@ -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
|
||||
|
||||
@ -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.<server>.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.
|
||||
|
||||
896
backend/tests/test_mcp_context_headers.py
Normal file
896
backend/tests/test_mcp_context_headers.py
Normal file
@ -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)
|
||||
202
backend/tests/test_mcp_header_names.py
Normal file
202
backend/tests/test_mcp_header_names.py
Normal file
@ -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"}
|
||||
@ -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,
|
||||
}
|
||||
]
|
||||
|
||||
|
||||
@ -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 <request-scoped credential>"
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
- 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:
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user