Python: scope Secure MCP URL headers to origin

python/packages/core/AGENTS.md

@@ -164,7 +164,7 @@ The vector store API is experimental under the shared `VECTOR_STORES` feature ID
 - **`allowed_tools`** (constructor arg on all `MCPTool` subclasses) - Restricts exposed MCP tools by raw remote MCP tool identity. Prefixed local names remain accepted only when the raw remote name already matches its normalized form; normalized/local aliases do not authorize a different raw remote name. If multiple raw remote tool names map to the same local function name, tool loading raises `ToolExecutionException` instead of first-one-wins shadowing.
 - **Progressive MCP disclosure** (`use_progressive_disclosure`, `always_load`) - When enabled on any `MCPTool` subclass, the initial model-facing surface is loader tools (`list_mcp_tools` / `load_tool` / `unload_tool`, prefixed by `tool_name_prefix` when configured) plus allowed tools selected by `always_load` and tools loaded earlier on the same `MCPTool` instance. `list_mcp_tools` only reports tools that pass `allowed_tools`; filtered tools are not listed or loadable. Loader tool names are reserved in progressive mode: remote MCP tools whose local generated name collides with a loader name are omitted from the initial/listed surface, and explicit `load_tool` calls return a model-visible message pointing callers to `tool_name_prefix` or excluding the colliding tool. `load_tool` accepts one tool name or a list of tool names and uses `FunctionInvocationContext.add_tools(...)` so the selected generated MCP `FunctionTool`s become available on the next function-calling iteration while keeping existing approval mode, argument filtering, header-provider runtime kwargs, result parsing, OTel, and task behavior. `unload_tool` accepts one dynamically loaded tool name or a list of names and removes them from the live tool list and persisted progressive surface, but it does not remove tools configured in `always_load`. Invalid `always_load` entries are ignored like unmatched `allowed_tools` entries.
 - **`additional_tool_argument_names`** (constructor arg on all `MCPTool` subclasses) - Opt extra argument names back into the allowlist. Accepts a `Sequence[str]` (applied to every tool) or a `Mapping[str, Sequence[str]]` keyed by **remote tool name**, where the reserved key `"*"` denotes global extras. It is configured only in user code at construction; there is **no per-call/runtime override**, so a model-issued tool call cannot change which names pass through — but note this constrains the *model*, not the *server*, which still widens the effective allowlist through its schema. To use a server that accepts `additionalProperties: true`, list the extra names here and then either (1) manually extend that tool's `inputSchema` (via the `.functions` list after connecting) so the model is prompted to supply them, or (2) supply the values yourself via `function_invocation_kwargs`. If a normal forwarded argument name is supplied by both the model and `function_invocation_kwargs`, the model-supplied value wins; `_meta` is the exception and only trusted runtime/caller metadata is used.
-- **`header_provider` request scoping** - When sharing an `http_client`, keep provider processing scoped to the originating `MCPStreamableHTTPTool` and strip injected headers from cross-origin redirects. The session exit stack removes its request hook after transport shutdown, including failed initialization or discovery, and closes framework-created HTTP clients. Failed discovery resets the connection and discovery flags and rolls back partial function/metadata additions. Framework-created sessions are discarded; constructor-supplied sessions remain caller-owned and reusable across cleanup, close, and reset.
+- **MCP HTTP header request scoping** - `static_headers` supplies fixed, origin-scoped headers without serializing concurrent calls; `header_provider` resolves dynamic per-call headers under the instance call lock. When both are configured, dynamic values override fixed values with the same name. With a shared `http_client`, processing stays scoped to the originating `MCPStreamableHTTPTool`, and every injected header is stripped from cross-origin redirects. The session exit stack removes its request hook after transport shutdown, including failed initialization or discovery, and closes framework-created HTTP clients. Failed discovery resets the connection and discovery flags and rolls back partial function/metadata additions. Framework-created sessions are discarded; constructor-supplied sessions remain caller-owned and reusable across cleanup, close, and reset.
 - **MCP lifecycle caller cancellation** - The lifecycle owner skips cancelled queued connect requests and waits for the caller to acknowledge successful setup. If the caller cancels before accepting a newly established connection, the owner tears it down before processing the next request. Cancelling a close waiter does not interrupt teardown, and cancelling a redundant connect does not discard a previously established session.
 - **`function_invocation_kwargs` and MCP servers** - That dict is shared across every tool in the run, including every attached `MCPTool`, and any name in it reaches a server that declares a matching `inputSchema` property. `header_provider` does not mitigate this — it reads the kwargs without consuming them. To keep a credential out of tool arguments, source it outside `function_invocation_kwargs`: read a `ContextVar` inside the provider (this still allows a different value per request), configure a custom `http_client`, or use `env` for `MCPStdioTool`.
 - **Sampling guardrails** (`sampling_callback`) - Passing `client=` advertises `SamplingCapability` so the server can send `sampling/createMessage`. Because remote servers are untrusted (confused-deputy risk), the default `sampling_callback` is **deny-by-default** and applies, in order: a per-session rate limit (`sampling_max_requests`, default `_DEFAULT_SAMPLING_MAX_REQUESTS`), an approval gate (`sampling_approval_callback`), and a `maxTokens` cap (`sampling_max_tokens`, default `_DEFAULT_SAMPLING_MAX_TOKENS`). The approval callback (constructor arg on all subclasses; exported type alias `SamplingApprovalCallback`) receives the raw `CreateMessageRequestParams`, may be sync or async, and must return truthy to approve. When it is `None` (the default) every sampling request is denied; pass `lambda params: True` to restore legacy auto-approve as an explicit opt-in. Requests and denials are logged at WARNING (content is not logged). The per-session counter resets in `_reset_session_state`.

python/packages/core/agent_framework/_mcp.py

@@ -642,11 +642,13 @@ def _inject_otel_into_mcp_meta(
     return meta
 
 
-def _url_origin(url: Any) -> tuple[str, str, int | None]:
+def _url_origin(url: Any) -> tuple[str, str, int]:
+    if url.scheme not in {"http", "https"} or not url.host:
+        raise ValueError("MCP URL must be an absolute HTTP(S) URL with a host.")
     port = url.port
     if port is None:
-        port = 443 if url.scheme == "https" else 80 if url.scheme == "http" else None
-    return (url.scheme, url.host or "", port)
+        port = 443 if url.scheme == "https" else 80
+    return (url.scheme, url.host, port)
 
 
 # Internal polling bounds for MCP long-running tasks. Not user-tunable today;
@@ -3479,6 +3481,7 @@ def __init__(
         sampling_max_requests: int | None = _DEFAULT_SAMPLING_MAX_REQUESTS,
         additional_properties: dict[str, Any] | None = None,
         http_client: AsyncClient | None = None,
+        static_headers: Mapping[str, str] | None = None,
         header_provider: Callable[[dict[str, Any]], dict[str, str]] | None = None,
         task_options: MCPTaskOptions | None = None,
         additional_tool_argument_names: Sequence[str] | Mapping[str, Sequence[str]] | None = None,
@@ -3550,8 +3553,8 @@ def __init__(
                 requests are rejected. Resets on reconnect. ``None`` disables it.
             http_client: Optional asyncClient to use. If not provided, the
                 ``streamable_http_client`` API will create and manage a default client.
-                To configure headers, timeouts, or other HTTP client settings, create
-                and pass your own ``asyncClient`` instance.
+                Use ``static_headers`` for fixed headers. To configure timeouts or other
+                HTTP client settings, create and pass your own ``asyncClient`` instance.
                 Security: when you attach sensitive headers (e.g. authentication tokens)
                 via a custom ``http_client``, you are responsible for enforcing the same
                 origin-scoped header policy that the built-in ``header_provider`` hook
@@ -3561,6 +3564,13 @@ def __init__(
                 client that sets headers unconditionally (e.g. via ``AsyncClient(headers=...)``
                 or ``follow_redirects=True`` without an origin check) can leak those headers
                 to other origins; scope them to the target origin yourself.
+            static_headers: Optional fixed HTTP headers to inject into requests to the
+                configured ``url`` origin. The headers are copied at construction, included
+                on connection-lifetime requests and tool calls, retained across same-origin
+                redirects, and removed on cross-origin redirects. Unlike ``header_provider``,
+                fixed headers do not serialize concurrent tool calls. Use ``header_provider``
+                instead when header values depend on runtime invocation arguments. When both
+                are provided, dynamic headers override fixed headers with the same name.
             header_provider: Optional callable that receives the runtime keyword arguments
                 (from ``FunctionInvocationContext.kwargs``) and returns a ``dict[str, str]``
                 of HTTP headers to inject into every outbound request to the MCP server.
@@ -3647,6 +3657,7 @@ def __init__(
         self.url = url
         self.terminate_on_close = terminate_on_close
         self._httpx_client: AsyncClient | None = http_client
+        self._static_headers = dict(static_headers or {})
         self._header_provider = header_provider
         # Headers for the in-flight call_tool invocation. The streamable HTTP transport
         # sends requests from tasks spawned at connect time, whose contexts never observe
@@ -3689,7 +3700,7 @@ def get_mcp_client(self) -> _AsyncGeneratorContextManager[Any, None]:
         from httpx import URL, AsyncClient, Timeout
 
         http_client = self._httpx_client
-        if self._header_provider is not None:
+        if self._static_headers or self._header_provider is not None:
             target_origin = _url_origin(URL(self.url))
             if http_client is None:
                 http_client = AsyncClient(
@@ -3709,26 +3720,28 @@ async def _inject_headers(request: Request) -> None:  # ruff:ignore[unused-async
                         for key in request.extensions.pop(_MCP_INJECTED_HEADER_KEYS_EXTENSION, ()):
                             request.headers.pop(key, None)
                         return
-                    # The transport may send this request from a task whose context was
-                    # captured before call_tool set the ContextVar; fall back to the
-                    # instance-level snapshot of the active call's headers. Both are None
-                    # only when this is an ambient request outside call_tool; an active
-                    # call that legitimately produced no headers yields an empty dict and
-                    # must not trigger the ambient fallback below.
-                    headers = _mcp_call_headers.get(None)
-                    if headers is None:
-                        headers = self._active_call_headers
-                    if headers is None:
+                    headers = self._static_headers.copy()
+                    if self._header_provider is not None:
+                        # The transport may send this request from a task whose context was
+                        # captured before call_tool set the ContextVar; fall back to the
+                        # instance-level snapshot of the active call's headers. Both are None
+                        # only when this is an ambient request outside call_tool; an active
+                        # call that legitimately produced no headers yields an empty dict and
+                        # must not trigger the ambient fallback below.
+                        dynamic_headers = _mcp_call_headers.get(None)
+                        if dynamic_headers is None:
+                            dynamic_headers = self._active_call_headers
+                    else:
+                        dynamic_headers = None
+                    if dynamic_headers is None and self._header_provider is not None:
                         # Ambient request made outside call_tool (the initialize handshake,
                         # load_tools/load_prompts discovery, or background pings). Invoke the
                         # provider with the kwargs seeded by the run that established this
                         # connection, so static providers and run-supplied credentials both
                         # authenticate these requests. Provider failures propagate, matching the
                         # call_tool path, except the one case below that no caller can avoid.
-                        if self._header_provider is None:
-                            raise RuntimeError("Header injection hook invoked without a header_provider.")
                         try:
-                            headers = self._header_provider(self._connection_kwargs or {})
+                            dynamic_headers = self._header_provider(self._connection_kwargs or {})
                         except KeyError:
                             # Unavoidable only when no run seeded this connection: the provider
                             # wants per-call values a connection-lifetime request cannot have. Once
@@ -3742,7 +3755,9 @@ async def _inject_headers(request: Request) -> None:  # ruff:ignore[unused-async
                                 self.name,
                                 exc_info=True,
                             )
-                            headers = {}
+                            dynamic_headers = {}
+                    if dynamic_headers is not None:
+                        headers.update(dynamic_headers)
                     for key in request.extensions.pop(_MCP_INJECTED_HEADER_KEYS_EXTENSION, ()):
                         request.headers.pop(key, None)
                     for key, value in headers.items():

python/packages/core/agent_framework/security.py

@@ -3968,8 +3968,8 @@ class SecureMCPToolProxy:
         url: URL of a remote MCP server.  When provided, the proxy creates
             an ``MCPStreamableHTTPTool`` internally.  Mutually exclusive with
             *mcp_tool*.
-        headers: HTTP headers (e.g. auth tokens) sent with every request
-            when using *url* mode.
+        headers: HTTP headers (e.g. auth tokens) sent with requests to the
+            configured origin when using *url* mode.
         name: Tool name used when creating the internal
             ``MCPStreamableHTTPTool`` (defaults to ``"mcp"``).
         description: Tool description for the internal tool.
@@ -4007,8 +4007,8 @@ def __init__(
         Keyword Args:
             url: URL of a remote MCP server. When provided, the proxy creates an
                 ``MCPStreamableHTTPTool`` internally. Mutually exclusive with ``mcp_tool``.
-            headers: HTTP headers (e.g. auth tokens) sent with every request when using
-                ``url`` mode.
+            headers: HTTP headers (e.g. auth tokens) sent with requests to the configured
+                origin when using ``url`` mode.
             name: Tool name used when creating the internal ``MCPStreamableHTTPTool``
                 (defaults to ``"mcp"``).
             description: Tool description for the internal tool.
@@ -4031,30 +4031,14 @@ def __init__(
             raise ValueError("Provide either 'mcp_tool' (an MCPTool instance) or 'url' (a remote MCP server URL).")
 
         if url is not None:
-            from httpx import AsyncClient, Timeout
-
-            from ._mcp import MCP_DEFAULT_SSE_READ_TIMEOUT, MCP_DEFAULT_TIMEOUT, MCPStreamableHTTPTool
+            from ._mcp import MCPStreamableHTTPTool
 
             static_headers = dict(headers or {})
-            # Pass headers via an AsyncClient so they are included on ALL requests
-            # (including session.initialize()), not just tool calls. Using
-            # header_provider alone only sets headers via a ContextVar that is
-            # populated during call_tool() and would be empty during initialization,
-            # causing 401s that silently manifest as anyio cancel-scope errors.
-            http_client = (
-                AsyncClient(
-                    headers=static_headers,
-                    follow_redirects=True,
-                    timeout=Timeout(MCP_DEFAULT_TIMEOUT, read=MCP_DEFAULT_SSE_READ_TIMEOUT),
-                )
-                if static_headers
-                else None
-            )
             mcp_tool = MCPStreamableHTTPTool(
                 name=name or "mcp",
                 url=url,
-                http_client=http_client,
                 description=description,
+                static_headers=static_headers,
             )
 
         # The validation above guarantees a tool is set (passed directly or built