From 05728b7e6f56c53a46de9d264c96dbc132781932 Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Tue, 8 Sep 2026 14:03:04 +0200
Subject: [PATCH 1/8] Added failing test

---
 python/packages/core/tests/core/test_mcp.py   | 76 +++++++++++++++++++
 .../tests/workflow/test_agent_executor.py     |  1 -
 2 files changed, 76 insertions(+), 1 deletion(-)

diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py
index f93727edb28..b1f79ce4a19 100644
--- a/python/packages/core/tests/core/test_mcp.py
+++ b/python/packages/core/tests/core/test_mcp.py
@@ -20,6 +20,7 @@
 from pydantic import AnyUrl, BaseModel
 
 from agent_framework import (
+    Agent,
     Content,
     FunctionInvocationContext,
     FunctionMiddleware,
@@ -28,6 +29,7 @@
     MCPStreamableHTTPTool,
     MCPWebsocketTool,
     Message,
+    SupportsChatGetResponse,
 )
 from agent_framework._feature_stage import _WARNED_FEATURES, ExperimentalFeature, ExperimentalWarning
 from agent_framework._mcp import (
@@ -6794,6 +6796,80 @@ def provider(kwargs):
         assert call_args.kwargs.get("arguments", {}).get("name") == "Alice"
 
 
+async def test_agent_run_supplies_mcp_connect_headers(
+    client: SupportsChatGetResponse,
+) -> None:
+    """Run-time credentials should authenticate implicit MCP initialization.
+
+    The agent receives function_invocation_kwargs before connecting the MCP tool
+    supplied to run(). This test exercises the real MCP transport against an
+    in-process mock HTTP endpoint and asserts that header_provider can use those
+    credentials on the initialize request, before any tool invocation occurs.
+    """
+    import httpx
+
+    captured_requests: list[tuple[str, str, dict[str, str]]] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "DELETE":
+            return httpx.Response(200)
+        if request.method == "GET":
+            return httpx.Response(405)
+        body = json.loads(request.content.decode())
+        method = body.get("method", "")
+        captured_requests.append((request.method, method, {k.lower(): v for k, v in request.headers.items()}))
+        if method == "initialize":
+            result = {
+                "protocolVersion": body["params"]["protocolVersion"],
+                "capabilities": {"tools": {}},
+                "serverInfo": {"name": "mock-server", "version": "1.0.0"},
+            }
+            return httpx.Response(
+                200,
+                headers={"mcp-session-id": "test-session"},
+                json={"jsonrpc": "2.0", "id": body["id"], "result": result},
+            )
+        if method == "tools/list":
+            result = {
+                "tools": [
+                    {
+                        "name": "greet",
+                        "description": "Says hello",
+                        "inputSchema": {"type": "object", "properties": {"name": {"type": "string"}}},
+                    }
+                ]
+            }
+            return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": result})
+        if method == "tools/call":
+            result = {"content": [{"type": "text", "text": "Hello!"}], "isError": False}
+            return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": result})
+        if "id" in body:
+            # Any other request (e.g. ping) gets an empty result so the session doesn't block on it.
+            return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}})
+        # Notifications (e.g. notifications/initialized)
+        return httpx.Response(202)
+
+    http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    tool = MCPStreamableHTTPTool(
+        name="test",
+        url="http://127.0.0.1:8000/mcp",
+        load_prompts=False,
+        http_client=http_client,
+        header_provider=lambda kw: {"x-api-key": kw["api_key"]},  # failing scenario
+        # header_provider=lambda _: {"x-api-key": "connect-token"} # working scenario (no lambda)
+    )
+    try:
+        async with Agent(client=client) as agent:
+            # placement of tools, matters as we defer the resolution until the agent calls `run`
+            await agent.run("Hello", tools=[tool], function_invocation_kwargs={"api_key": "connect-token"})
+    finally:
+        await http_client.aclose()
+
+    initialize_headers = [headers for _, method, headers in captured_requests if method == "initialize"]
+    assert len(initialize_headers) == 1
+    assert initialize_headers[0].get("x-api-key") == "connect-token"
+
+
 async def test_mcp_streamable_http_tool_header_provider_applies_across_transport_tasks():
     """Regression test for #7161: header_provider headers must reach tools/call requests.
 
diff --git a/python/packages/core/tests/workflow/test_agent_executor.py b/python/packages/core/tests/workflow/test_agent_executor.py
index 2cc2ed2ce6e..4b067ab7c1c 100644
--- a/python/packages/core/tests/workflow/test_agent_executor.py
+++ b/python/packages/core/tests/workflow/test_agent_executor.py
@@ -1,7 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
 
 import pickle
-
 from collections.abc import AsyncIterable, Awaitable
 from typing import Any, Literal, overload
 

From a1e05bc02c4cbc73ccc339617e93745628f5e91b Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Tue, 8 Sep 2026 15:20:30 +0200
Subject: [PATCH 2/8] seeding run header value for auth to mcp connect

---
 python/packages/core/agent_framework/_agents.py | 14 +++++++++-----
 python/packages/core/agent_framework/_mcp.py    | 12 +++++++++++-
 2 files changed, 20 insertions(+), 6 deletions(-)

diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py
index a115f600f3d..0853d73e166 100644
--- a/python/packages/core/agent_framework/_agents.py
+++ b/python/packages/core/agent_framework/_agents.py
@@ -1455,11 +1455,20 @@ async def _prepare_run_context(
         # Normalize tools
         normalized_tools = _normalize_tools(tools_)
 
+        # Extract additional function arguments
+        effective_function_invocation_kwargs = (
+            dict(function_invocation_kwargs) if function_invocation_kwargs is not None else {}
+        )
+        additional_function_arguments = {**effective_function_invocation_kwargs, **existing_additional_args}
+
         # Resolve final tool list (configured tools + runtime provided tools + local MCP server tools)
         final_tools = list(base_tools)
         for tool in normalized_tools:
             if isinstance(tool, MCPTool):
                 if not tool.is_connected:
+                    # The handshake and discovery requests are issued before any tool call, so the run's
+                    # kwargs must reach header_provider here or those requests go out unauthenticated.
+                    tool._seed_connection_kwargs(additional_function_arguments)  # pyright: ignore[reportPrivateUsage]
                     await self._async_exit_stack.enter_async_context(tool)
                 _append_unique_tools(
                     final_tools,
@@ -1478,11 +1487,6 @@ async def _prepare_run_context(
                 duplicate_error_message=mcp_duplicate_message,
             )
 
-        effective_function_invocation_kwargs = (
-            dict(function_invocation_kwargs) if function_invocation_kwargs is not None else {}
-        )
-        additional_function_arguments = {**effective_function_invocation_kwargs, **existing_additional_args}
-
         model = opts.pop("model", None)
 
         # Build options dict from run() options merged with provided options
diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py
index f1bf4ea4bc0..91ea3c13c3d 100644
--- a/python/packages/core/agent_framework/_mcp.py
+++ b/python/packages/core/agent_framework/_mcp.py
@@ -1584,6 +1584,10 @@ async def _connect_on_owner(self, *, reset: bool = False, load_configured: bool
                 self._tool_param_names_by_name = param_names_before_discovery
             raise
 
+    def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
+        """Offer run-scoped kwargs to connection-lifetime header resolution."""
+        return
+
     async def _sampling_request_approved(self, params: types.CreateMessageRequestParams) -> bool:
         """Run the configured sampling approval gate.
 
@@ -3286,6 +3290,7 @@ def __init__(
         # when a header_provider is set: parallel invocations on the same instance would
         # otherwise overwrite each other's snapshot and attach the wrong per-call headers.
         self._active_call_headers: dict[str, str] | None = None
+        self._connection_kwargs: dict[str, Any] = {}
         self._call_headers_lock = asyncio.Lock()
         self._header_request_owner = object()
         self._header_hook_client: AsyncClient | None = None
@@ -3358,7 +3363,7 @@ async def _inject_headers(request: Request) -> None:  # ruff:ignore[unused-async
                         if self._header_provider is None:
                             raise RuntimeError("Header injection hook invoked without a header_provider.")
                         try:
-                            headers = self._header_provider({})
+                            headers = self._header_provider(self._connection_kwargs)
                         except KeyError:
                             # A kwargs-dependent provider raises on every ambient request
                             # (initialize, discovery, and recurring pings).
@@ -3422,6 +3427,11 @@ async def _close_on_owner(self) -> None:
         finally:
             self._remove_header_hook()
 
+    def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
+        if self._header_provider is None or self.is_connected:
+            return
+        self._connection_kwargs = dict(kwargs)
+
     async def call_tool(self, tool_name: str, **kwargs: Any) -> str | list[Content]:
         """Call a tool, injecting headers from the header_provider if configured.
 

From 2280b1aabe71b7d4059cd143f61b3c7506bab832 Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Wed, 9 Sep 2026 12:56:33 +0200
Subject: [PATCH 3/8] WIP: added connection_kwargs and tests

---
 python/packages/core/agent_framework/_mcp.py  |  1 +
 python/packages/core/tests/core/test_mcp.py   | 68 ++++++++++++++
 .../core/tests/core/test_mcp_http_auth.py     | 93 +++++++++++++++++++
 3 files changed, 162 insertions(+)

diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py
index 91ea3c13c3d..58eb848affa 100644
--- a/python/packages/core/agent_framework/_mcp.py
+++ b/python/packages/core/agent_framework/_mcp.py
@@ -3425,6 +3425,7 @@ async def _close_on_owner(self) -> None:
         try:
             await super()._close_on_owner()
         finally:
+            self._connection_kwargs = {}
             self._remove_header_hook()
 
     def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py
index b1f79ce4a19..d3e671ad2ae 100644
--- a/python/packages/core/tests/core/test_mcp.py
+++ b/python/packages/core/tests/core/test_mcp.py
@@ -6870,6 +6870,74 @@ async def handler(request: httpx.Request) -> httpx.Response:
     assert initialize_headers[0].get("x-api-key") == "connect-token"
 
 
+async def test_agent_context_manager_authenticates_connect_with_closure_provider(
+    client: SupportsChatGetResponse,
+) -> None:
+    """A constructor-supplied MCP tool authenticates its eager handshake with no run involved.
+
+    Pins that ``header_provider`` already covers construction-time credentials: entering the
+    agent context connects before any run exists, and the server rejects unauthenticated calls.
+    """
+    import httpx
+
+    captured_requests: list[tuple[str, dict[str, str]]] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "DELETE":
+            return httpx.Response(200)
+        if request.method == "GET":
+            return httpx.Response(405)
+        if request.headers.get("x-api-key") != "constructor-token":
+            return httpx.Response(401)
+        body = json.loads(request.content.decode())
+        method = body.get("method", "")
+        captured_requests.append((method, {k.lower(): v for k, v in request.headers.items()}))
+        if method == "initialize":
+            return httpx.Response(
+                200,
+                headers={"mcp-session-id": "test-session"},
+                json={
+                    "jsonrpc": "2.0",
+                    "id": body["id"],
+                    "result": {
+                        "protocolVersion": body["params"]["protocolVersion"],
+                        "capabilities": {"tools": {}},
+                        "serverInfo": {"name": "mock-server", "version": "1.0.0"},
+                    },
+                },
+            )
+        if method == "tools/list":
+            return httpx.Response(
+                200,
+                json={
+                    "jsonrpc": "2.0",
+                    "id": body["id"],
+                    "result": {"tools": [{"name": "greet", "inputSchema": {"type": "object", "properties": {}}}]},
+                },
+            )
+        if "id" in body:
+            return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}})
+        return httpx.Response(202)
+
+    http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    # The credential is known at construction, so the provider closes over it and ignores kwargs.
+    tool = MCPStreamableHTTPTool(
+        name="test",
+        url="http://127.0.0.1:8000/mcp",
+        load_prompts=False,
+        http_client=http_client,
+        header_provider=lambda _kwargs: {"x-api-key": "constructor-token"},
+    )
+    try:
+        async with Agent(client=client, tools=[tool]):
+            assert tool.is_connected
+    finally:
+        await http_client.aclose()
+
+    assert [method for method, _ in captured_requests].count("initialize") == 1
+    assert all(headers.get("x-api-key") == "constructor-token" for _, headers in captured_requests)
+
+
 async def test_mcp_streamable_http_tool_header_provider_applies_across_transport_tasks():
     """Regression test for #7161: header_provider headers must reach tools/call requests.
 
diff --git a/python/packages/core/tests/core/test_mcp_http_auth.py b/python/packages/core/tests/core/test_mcp_http_auth.py
index 04c27676060..6a6af97cce7 100644
--- a/python/packages/core/tests/core/test_mcp_http_auth.py
+++ b/python/packages/core/tests/core/test_mcp_http_auth.py
@@ -732,3 +732,96 @@ async def block_discovery() -> None:
             finally:
                 release_setup.set()
                 await borrowed.close()
+
+
+def _kwargs_dependent_tool(client: httpx.AsyncClient) -> MCPStreamableHTTPTool:
+    """A tool whose header_provider can only authenticate when connection kwargs are seeded."""
+    return MCPStreamableHTTPTool(
+        name="seeded",
+        url="https://mcp.example/mcp",
+        http_client=client,
+        load_prompts=False,
+        header_provider=lambda kwargs: {"Authorization": kwargs["credential"]},
+    )
+
+
+@pytest.mark.parametrize("seeded", [True, False])
+async def test_seeded_connection_kwargs_authenticate_the_handshake(
+    mcp_http_server: MCPHTTPServer, seeded: bool
+) -> None:
+    client, requests, _ = mcp_http_server
+    tool = _kwargs_dependent_tool(client)
+    if seeded:
+        tool._seed_connection_kwargs({"credential": "token-a"})
+    try:
+        if not seeded:
+            with pytest.raises(ToolException):
+                await tool.connect()
+            return
+        async with tool:
+            await tool.call_tool("record", credential="token-a")
+        initializes = [
+            request
+            for request in requests
+            if request.method == "POST" and json.loads(request.content).get("method") == "initialize"
+        ]
+        assert [request.headers.get("Authorization") for request in initializes] == ["token-a"]
+    finally:
+        await tool.close()
+
+
+async def test_connection_kwargs_are_fixed_for_the_connection_and_cleared_on_close(
+    mcp_http_server: MCPHTTPServer,
+) -> None:
+    client, requests, _ = mcp_http_server
+    tool = _kwargs_dependent_tool(client)
+    tool._seed_connection_kwargs({"credential": "token-a"})
+    try:
+        async with tool:
+            # A later run must not re-authenticate an already-established connection.
+            tool._seed_connection_kwargs({"credential": "token-b"})
+            assert tool._connection_kwargs == {"credential": "token-a"}
+            # Call scope resolves independently of the connection scope.
+            await tool.call_tool("record", credential="token-c")
+        initializes = [
+            request
+            for request in requests
+            if request.method == "POST" and json.loads(request.content).get("method") == "initialize"
+        ]
+        assert [request.headers.get("Authorization") for request in initializes] == ["token-a"]
+        assert [request.headers.get("Authorization") for request in _calls(requests)] == ["token-c"]
+        assert tool._connection_kwargs == {}
+        # The credential was released on close, so an unseeded reconnect is rejected.
+        with pytest.raises(ToolException):
+            await tool.connect()
+    finally:
+        await tool.close()
+
+
+async def test_standalone_static_header_provider_authenticates_without_a_run(mcp_http_server: MCPHTTPServer) -> None:
+    """A provider that ignores kwargs authenticates the handshake outside any agent run.
+
+    Pins the boundary of the connection-kwargs seeding: standalone use has no per-call kwargs
+    to seed from, so a closure/token-provider style provider is the supported shape there.
+    """
+    client, requests, _ = mcp_http_server
+    tool = MCPStreamableHTTPTool(
+        name="standalone",
+        url="https://mcp.example/mcp",
+        http_client=client,
+        load_prompts=False,
+        header_provider=lambda _kwargs: {"Authorization": "token-a"},
+    )
+    try:
+        async with tool:
+            await tool.call_tool("record")
+        authenticated = [
+            request
+            for request in requests
+            if request.method == "POST"
+            and json.loads(request.content).get("method") in {"initialize", "tools/list", "tools/call"}
+        ]
+        assert authenticated
+        assert all(request.headers.get("Authorization") == "token-a" for request in authenticated)
+    finally:
+        await tool.close()

From b41e4ac626b9007123315e9406b42a7630697d60 Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Wed, 9 Sep 2026 17:17:55 +0200
Subject: [PATCH 4/8] Python: fix(core): seed run kwargs into MCP
 connection-header resolution

Constructor-supplied MCP tools now seed the run's function_invocation_kwargs before connecting, matching the run-supplied path. Adds coverage for the connect-time credential matrix and fixes the API-key sample to close over a construction-time credential.
---
 .../packages/core/agent_framework/_agents.py  |  1 +
 python/packages/core/tests/core/test_mcp.py   | 70 +++++++++++++++++++
 .../samples/02-agents/mcp/mcp_api_key_auth.py | 33 ++++-----
 3 files changed, 85 insertions(+), 19 deletions(-)

diff --git a/python/packages/core/agent_framework/_agents.py b/python/packages/core/agent_framework/_agents.py
index 0853d73e166..3fa71ed9689 100644
--- a/python/packages/core/agent_framework/_agents.py
+++ b/python/packages/core/agent_framework/_agents.py
@@ -1480,6 +1480,7 @@ async def _prepare_run_context(
 
         for mcp_server in self.mcp_tools:
             if not mcp_server.is_connected:
+                mcp_server._seed_connection_kwargs(additional_function_arguments)  # pyright: ignore[reportPrivateUsage]
                 await self._async_exit_stack.enter_async_context(mcp_server)
             _append_unique_tools(
                 final_tools,
diff --git a/python/packages/core/tests/core/test_mcp.py b/python/packages/core/tests/core/test_mcp.py
index d3e671ad2ae..de4f86372cd 100644
--- a/python/packages/core/tests/core/test_mcp.py
+++ b/python/packages/core/tests/core/test_mcp.py
@@ -6938,6 +6938,76 @@ async def handler(request: httpx.Request) -> httpx.Response:
     assert all(headers.get("x-api-key") == "constructor-token" for _, headers in captured_requests)
 
 
+async def test_constructor_supplied_mcp_tool_uses_run_credentials_on_lazy_connect(
+    client: SupportsChatGetResponse,
+) -> None:
+    """A constructor-supplied MCP tool connected at first run authenticates with that run's kwargs.
+
+    Without the agent context manager the handshake is deferred to ``run()``, so the run's
+    credentials are available and must reach ``header_provider``.
+    """
+    import httpx
+
+    captured_requests: list[tuple[str, dict[str, str]]] = []
+
+    async def handler(request: httpx.Request) -> httpx.Response:
+        if request.method == "DELETE":
+            return httpx.Response(200)
+        if request.method == "GET":
+            return httpx.Response(405)
+        if request.headers.get("x-api-key") != "run-token":
+            return httpx.Response(401)
+        body = json.loads(request.content.decode())
+        method = body.get("method", "")
+        captured_requests.append((method, {k.lower(): v for k, v in request.headers.items()}))
+        if method == "initialize":
+            return httpx.Response(
+                200,
+                headers={"mcp-session-id": "test-session"},
+                json={
+                    "jsonrpc": "2.0",
+                    "id": body["id"],
+                    "result": {
+                        "protocolVersion": body["params"]["protocolVersion"],
+                        "capabilities": {"tools": {}},
+                        "serverInfo": {"name": "mock-server", "version": "1.0.0"},
+                    },
+                },
+            )
+        if method == "tools/list":
+            return httpx.Response(
+                200,
+                json={
+                    "jsonrpc": "2.0",
+                    "id": body["id"],
+                    "result": {"tools": [{"name": "greet", "inputSchema": {"type": "object", "properties": {}}}]},
+                },
+            )
+        if "id" in body:
+            return httpx.Response(200, json={"jsonrpc": "2.0", "id": body["id"], "result": {}})
+        return httpx.Response(202)
+
+    http_client = httpx.AsyncClient(transport=httpx.MockTransport(handler))
+    tool = MCPStreamableHTTPTool(
+        name="test",
+        url="http://127.0.0.1:8000/mcp",
+        load_prompts=False,
+        http_client=http_client,
+        header_provider=lambda kw: {"x-api-key": kw["api_key"]},
+    )
+    agent = Agent(client=client, tools=[tool])
+    try:
+        # No agent context manager, so the tool is still unconnected when the run starts.
+        assert not tool.is_connected
+        await agent.run("Hello", function_invocation_kwargs={"api_key": "run-token"})
+    finally:
+        await tool.close()
+        await http_client.aclose()
+
+    assert [method for method, _ in captured_requests].count("initialize") == 1
+    assert all(headers.get("x-api-key") == "run-token" for _, headers in captured_requests)
+
+
 async def test_mcp_streamable_http_tool_header_provider_applies_across_transport_tasks():
     """Regression test for #7161: header_provider headers must reach tools/call requests.
 
diff --git a/python/samples/02-agents/mcp/mcp_api_key_auth.py b/python/samples/02-agents/mcp/mcp_api_key_auth.py
index 5af7f29009d..b38247f48ea 100644
--- a/python/samples/02-agents/mcp/mcp_api_key_auth.py
+++ b/python/samples/02-agents/mcp/mcp_api_key_auth.py
@@ -13,10 +13,9 @@
 """
 MCP API Key Authentication Example
 
-This sample demonstrates the runtime ``header_provider`` pattern for
-``MCPStreamableHTTPTool``. The MCP tool derives authentication headers from
-``function_invocation_kwargs`` passed to ``Agent.run(...)`` so the API key stays
-in runtime context instead of being baked into a shared ``httpx.AsyncClient``.
+This sample demonstrates the ``header_provider`` pattern for ``MCPStreamableHTTPTool``.
+The MCP tool derives its authentication headers from a provider callable, so the API key
+stays in application code instead of being baked into a shared ``httpx.AsyncClient``.
 
 Replace the ``url`` parameter in the ``MCPStreamableHTTPTool`` with your authenticated server URL and
 run the sample with your API key as a command-line argument:
@@ -27,17 +26,16 @@
 For more complex scenarios, you could implement token refresh logic or support multiple authentication methods
 within the header provider function.
 
-Note on ``function_invocation_kwargs``:
-Values passed this way are shared with every tool in the run, including every attached MCP server, and are
-filtered against each tool's server-declared ``inputSchema.properties``. A server that declares ``mcp_api_key``
-therefore receives it as an ordinary tool argument, and ``header_provider`` does not prevent that - it reads the
-kwargs without consuming them. This sample targets a server you control, so that is fine here. For a server you
-do not control, source the credential outside ``function_invocation_kwargs`` instead, for example by reading a
-``ContextVar`` inside the provider::
+Note on where the credential comes from:
+The provider below closes over ``api_key`` because the value is known before the agent is built. That is also
+what authenticates the MCP handshake: entering the agent context connects the tool before any run, so a provider
+that depended on per-run ``function_invocation_kwargs`` would have nothing to read at that point.
 
-    header_provider=lambda _kwargs: {"Authorization": f"Bearer {token_var.get()}"}
-
-which keeps it out of tool arguments while still allowing a different value per request.
+When the credential only arrives with the run, pass it via ``function_invocation_kwargs`` and supply the tool
+through ``run(tools=[...])`` so the connection happens inside that run. Be aware those values are shared with
+every tool in the run and are filtered against each tool's server-declared ``inputSchema.properties``, so a
+server that declares ``mcp_api_key`` receives it as an ordinary tool argument - ``header_provider`` reads the
+kwargs without consuming them.
 
 For more authentication examples including OAuth 2.0 flows, see:
 - https://github.com/modelcontextprotocol/python-sdk/tree/main/examples/clients/simple-auth-client
@@ -56,15 +54,12 @@ async def api_key_auth_example(api_key: str) -> None:
             name="MCP tool",
             description="MCP tool description.",
             url="<your authenticated server url>",
-            header_provider=lambda kwargs: {"Authorization": f"Bearer {kwargs['mcp_api_key']}"},
+            header_provider=lambda _kwargs: {"Authorization": f"Bearer {api_key}"},
         ),
     ) as agent:
         query = "Use your MCP tool to tell me what tools are available to you."
         print(f"User: {query}")
-        result = await agent.run(
-            query,
-            function_invocation_kwargs={"mcp_api_key": api_key},
-        )
+        result = await agent.run(query)
         print(f"Agent: {result.text}")
 
 

From e14c4b253a5ae88b8b932f836656c50a45962050 Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Thu, 10 Sep 2026 11:29:41 +0200
Subject: [PATCH 5/8] Python: fix(core): fail the MCP handshake when seeded
 kwargs miss the provider's key

Ambient header resolution previously swallowed every KeyError from header_provider, so a run whose kwargs did not carry the key the provider reads would silently send the initialize handshake unauthenticated and surface as a 401. Only tolerate the KeyError when no connection kwargs were seeded at all, which is the case a caller cannot avoid; a key missing from seeded kwargs is a misconfiguration and now propagates.
---
 python/packages/core/agent_framework/_mcp.py  | 20 +++++++-------
 .../core/tests/core/test_mcp_http_auth.py     | 26 +++++++++++++++++++
 2 files changed, 37 insertions(+), 9 deletions(-)

diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py
index 58eb848affa..91b9ed94c7e 100644
--- a/python/packages/core/agent_framework/_mcp.py
+++ b/python/packages/core/agent_framework/_mcp.py
@@ -3354,22 +3354,24 @@ async def _inject_headers(request: Request) -> None:  # ruff:ignore[unused-async
                     if headers is None:
                         # Ambient request made outside call_tool (the initialize handshake,
                         # load_tools/load_prompts discovery, or background pings). Invoke the
-                        # provider with empty kwargs so static providers can authenticate these
-                        # requests too. A provider that indexes a required per-call kwarg (e.g.
-                        # kwargs["api_key"]) raises KeyError on the empty dict; that specific
-                        # case is tolerated so connect still succeeds. Any other error is a
-                        # genuine provider failure and is left to propagate, matching the
-                        # call_tool path which does not catch header_provider exceptions.
+                        # 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)
                         except KeyError:
-                            # A kwargs-dependent provider raises on every ambient request
-                            # (initialize, discovery, and recurring pings).
+                            # Unavoidable only when nothing was seeded: the provider wants per-call
+                            # values a connection-lifetime request cannot have. A key missing from
+                            # seeded kwargs is a misconfiguration, and silently dropping it would
+                            # send the handshake unauthenticated.
+                            if self._connection_kwargs:
+                                raise
                             logger.debug(
                                 "header_provider raised KeyError for MCP server %r on an ambient "
-                                "request (missing per-call kwargs); proceeding without headers.",
+                                "request (no connection kwargs available); proceeding without headers.",
                                 self.name,
                                 exc_info=True,
                             )
diff --git a/python/packages/core/tests/core/test_mcp_http_auth.py b/python/packages/core/tests/core/test_mcp_http_auth.py
index 6a6af97cce7..1c6cbe1acab 100644
--- a/python/packages/core/tests/core/test_mcp_http_auth.py
+++ b/python/packages/core/tests/core/test_mcp_http_auth.py
@@ -825,3 +825,29 @@ async def test_standalone_static_header_provider_authenticates_without_a_run(mcp
         assert all(request.headers.get("Authorization") == "token-a" for request in authenticated)
     finally:
         await tool.close()
+
+
+async def test_seeded_kwargs_missing_the_providers_key_fails_the_handshake(
+    mcp_http_server: MCPHTTPServer,
+) -> None:
+    """A key absent from seeded connection kwargs is a misconfiguration, not a tolerated ambient miss.
+
+    An unseeded connection legitimately has no per-call values, so a KeyError there is tolerated.
+    Once a run supplies kwargs, a provider asking for an absent key must fail loudly instead of
+    letting the handshake go out unauthenticated.
+    """
+    client, _, _ = mcp_http_server
+    tool = MCPStreamableHTTPTool(
+        name="mismatch",
+        url="https://mcp.example/mcp",
+        http_client=client,
+        load_prompts=False,
+        header_provider=lambda kwargs: {"Authorization": kwargs["credential"]},
+    )
+    tool._seed_connection_kwargs({"typo_credential": "token-a"})
+    try:
+        with pytest.raises(ToolException) as error:
+            await tool.connect()
+        assert "'credential'" in str(error.value)
+    finally:
+        await tool.close()

From 0226cf2203712e9ef98f87ce8227a19749de51f4 Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Thu, 10 Sep 2026 13:59:49 +0200
Subject: [PATCH 6/8] Python: fix(core): distinguish unseeded MCP connections
 and release kwargs on failed connect

Addresses review on #8225. An empty mapping could not express the difference between a run that seeded no kwargs and a connection no run ever seeded, so a lazy run with empty kwargs still swallowed the provider's KeyError and sent an unauthenticated handshake. _connection_kwargs is now None until a run seeds it. The seeded kwargs were also only released through close(); a rejected handshake unwinds via _close_and_check_cancelled instead, leaving the failed run's credential for a later unseeded reconnect, so the release now happens there. Documents the connection-lifetime semantics on the public header_provider parameter.
---
 python/packages/core/agent_framework/_mcp.py  | 42 +++++++++++++----
 .../core/tests/core/test_mcp_http_auth.py     | 46 ++++++++++++++++++-
 2 files changed, 79 insertions(+), 9 deletions(-)

diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py
index c5b27f5ddb8..3b275eff5c8 100644
--- a/python/packages/core/agent_framework/_mcp.py
+++ b/python/packages/core/agent_framework/_mcp.py
@@ -1699,6 +1699,11 @@ async def _close_and_check_cancelled(self, ex: BaseException) -> tuple[bool, Bas
         error path holding only a bare cancellation can still describe it.
         """
         cleanup_error = await self._safe_close_exit_stack()
+        # Every abandoned connection attempt lands here, so a rejected handshake cannot
+        # leave one run's credentials visible to a later unseeded reconnect. Deliberately
+        # not in _safe_close_exit_stack: connect(reset=True) closes through that path and
+        # must keep its kwargs to re-authenticate the new connection.
+        self._release_connection_kwargs()
         return _should_propagate_cancelled_error(ex), cleanup_error
 
     def _reset_session_state(self) -> None:
@@ -1894,6 +1899,10 @@ def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
         """Offer run-scoped kwargs to connection-lifetime header resolution."""
         return
 
+    def _release_connection_kwargs(self) -> None:
+        """Drop any run-scoped kwargs held for connection-lifetime header resolution."""
+        return
+
     async def _sampling_request_approved(self, params: types.CreateMessageRequestParams) -> bool:
         """Run the configured sampling approval gate.
 
@@ -3553,6 +3562,18 @@ def __init__(
                 of HTTP headers to inject into every outbound request to the MCP server.
                 Use this to forward per-request context (e.g. authentication tokens set in
                 agent middleware) without creating a separate ``httpx.AsyncClient``.
+                Only tool calls carry a run's kwargs. Connection-lifetime requests - the
+                ``initialize`` handshake, tool and prompt discovery, and background pings -
+                belong to no call, so they reuse the kwargs of the run that established the
+                connection until the tool is closed; a later run's kwargs do not reach them.
+                A tool connected outside any run (eagerly via ``async with``, or standalone)
+                has no kwargs to reuse and the provider is called with an empty mapping, in
+                which case a ``KeyError`` from the provider is tolerated and the request is
+                sent without headers. Once a run has supplied kwargs, a ``KeyError`` is
+                raised instead, since a key missing there is a misconfiguration rather than
+                an unavoidable gap. A credential that must authenticate the handshake should
+                therefore come from somewhere the provider can read without a run - a closure
+                or a ``ContextVar`` - rather than from run kwargs alone.
                 The framework attaches these headers only to requests whose origin (scheme,
                 host, port) matches the configured ``url``, so they are not leaked to other
                 origins on cross-origin redirects; headers injected this way are also removed
@@ -3630,7 +3651,9 @@ def __init__(
         # when a header_provider is set: parallel invocations on the same instance would
         # otherwise overwrite each other's snapshot and attach the wrong per-call headers.
         self._active_call_headers: dict[str, str] | None = None
-        self._connection_kwargs: dict[str, Any] = {}
+        # None means no run seeded this connection, which an empty mapping cannot express:
+        # a run that supplies no kwargs still expects a missing provider key to be an error.
+        self._connection_kwargs: dict[str, Any] | None = None
         self._call_headers_lock = asyncio.Lock()
         self._header_request_owner = object()
         self._header_hook_client: AsyncClient | None = None
@@ -3701,13 +3724,13 @@ async def _inject_headers(request: Request) -> None:  # ruff:ignore[unused-async
                         if self._header_provider is None:
                             raise RuntimeError("Header injection hook invoked without a header_provider.")
                         try:
-                            headers = self._header_provider(self._connection_kwargs)
+                            headers = self._header_provider(self._connection_kwargs or {})
                         except KeyError:
-                            # Unavoidable only when nothing was seeded: the provider wants per-call
-                            # values a connection-lifetime request cannot have. A key missing from
-                            # seeded kwargs is a misconfiguration, and silently dropping it would
-                            # send the handshake unauthenticated.
-                            if self._connection_kwargs:
+                            # Unavoidable only when no run seeded this connection: the provider
+                            # wants per-call values a connection-lifetime request cannot have. Once
+                            # a run has seeded kwargs a missing key is a misconfiguration, and
+                            # silently dropping it would send the handshake unauthenticated.
+                            if self._connection_kwargs is not None:
                                 raise
                             logger.debug(
                                 "header_provider raised KeyError for MCP server %r on an ambient "
@@ -3767,7 +3790,7 @@ async def _close_on_owner(self) -> None:
         try:
             await super()._close_on_owner()
         finally:
-            self._connection_kwargs = {}
+            self._release_connection_kwargs()
             self._remove_header_hook()
 
     def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
@@ -3775,6 +3798,9 @@ def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
             return
         self._connection_kwargs = dict(kwargs)
 
+    def _release_connection_kwargs(self) -> None:
+        self._connection_kwargs = None
+
     async def call_tool(self, tool_name: str, **kwargs: Any) -> str | list[Content]:
         """Call a tool, injecting headers from the header_provider if configured.
 
diff --git a/python/packages/core/tests/core/test_mcp_http_auth.py b/python/packages/core/tests/core/test_mcp_http_auth.py
index 1c6cbe1acab..ca4389f8321 100644
--- a/python/packages/core/tests/core/test_mcp_http_auth.py
+++ b/python/packages/core/tests/core/test_mcp_http_auth.py
@@ -790,7 +790,7 @@ async def test_connection_kwargs_are_fixed_for_the_connection_and_cleared_on_clo
         ]
         assert [request.headers.get("Authorization") for request in initializes] == ["token-a"]
         assert [request.headers.get("Authorization") for request in _calls(requests)] == ["token-c"]
-        assert tool._connection_kwargs == {}
+        assert tool._connection_kwargs is None
         # The credential was released on close, so an unseeded reconnect is rejected.
         with pytest.raises(ToolException):
             await tool.connect()
@@ -851,3 +851,47 @@ async def test_seeded_kwargs_missing_the_providers_key_fails_the_handshake(
         assert "'credential'" in str(error.value)
     finally:
         await tool.close()
+
+
+async def test_run_supplying_no_kwargs_still_fails_a_kwargs_dependent_provider(
+    mcp_http_server: MCPHTTPServer,
+) -> None:
+    """Seeding an empty mapping is still seeding, so the provider's missing key must not be tolerated.
+
+    An empty mapping cannot distinguish a run that supplied no kwargs from a connection no run
+    ever seeded; only the latter has no way to carry the key and may proceed unauthenticated.
+    """
+    client, _, _ = mcp_http_server
+    tool = _kwargs_dependent_tool(client)
+    tool._seed_connection_kwargs({})
+    try:
+        with pytest.raises(ToolException) as error:
+            await tool.connect()
+        assert "'credential'" in str(error.value)
+    finally:
+        await tool.close()
+
+
+async def test_failed_connect_releases_the_seeded_credential(mcp_http_server: MCPHTTPServer) -> None:
+    """An abandoned connection attempt must not leave its credential for a later unseeded connect.
+
+    A rejected handshake unwinds without going through close(), so the release has to happen on
+    the failure path too; otherwise a standalone reconnect re-sends the failed run's credential.
+    """
+    client, requests, _ = mcp_http_server
+    tool = _kwargs_dependent_tool(client)
+    tool._seed_connection_kwargs({"credential": "token-rejected"})
+    try:
+        with pytest.raises(ToolException):
+            await tool.connect()
+        assert tool._connection_kwargs is None
+        with pytest.raises(ToolException):
+            await tool.connect()
+        initializes = [
+            request
+            for request in requests
+            if request.method == "POST" and json.loads(request.content).get("method") == "initialize"
+        ]
+        assert [request.headers.get("Authorization") for request in initializes] == ["token-rejected", None]
+    finally:
+        await tool.close()

From d81daf582ff15a8757868bcf361eedca31bda0f0 Mon Sep 17 00:00:00 2001
From: Jose Alvarez <josealvar@microsoft.com>
Date: Thu, 10 Sep 2026 14:01:05 +0200
Subject: [PATCH 7/8] Python: fix(core): make the MCP connection kwargs seed a
 first-writer claim

Addresses review on #8225. is_connected only turns true after initialize returns, so two concurrent runs could both pass the guard and the second would swap the credential out from under the first run's in-flight handshake, authenticating a shared tool as the wrong caller. The seed is now a first-writer claim, released when the connection closes or its setup fails.
---
 python/packages/core/agent_framework/_mcp.py  |  5 ++++
 .../core/tests/core/test_mcp_http_auth.py     | 24 +++++++++++++++++++
 2 files changed, 29 insertions(+)

diff --git a/python/packages/core/agent_framework/_mcp.py b/python/packages/core/agent_framework/_mcp.py
index 3b275eff5c8..0197d9a3535 100644
--- a/python/packages/core/agent_framework/_mcp.py
+++ b/python/packages/core/agent_framework/_mcp.py
@@ -3796,6 +3796,11 @@ async def _close_on_owner(self) -> None:
     def _seed_connection_kwargs(self, kwargs: Mapping[str, Any]) -> None:
         if self._header_provider is None or self.is_connected:
             return
+        # is_connected stays false until initialize returns, so it alone would let a second
+        # concurrent run swap the credential out from under the first run's in-flight
+        # handshake. The claim is released when the connection closes or its setup fails.
+        if self._connection_kwargs is not None:
+            return
         self._connection_kwargs = dict(kwargs)
 
     def _release_connection_kwargs(self) -> None:
diff --git a/python/packages/core/tests/core/test_mcp_http_auth.py b/python/packages/core/tests/core/test_mcp_http_auth.py
index ca4389f8321..31bb5afae5c 100644
--- a/python/packages/core/tests/core/test_mcp_http_auth.py
+++ b/python/packages/core/tests/core/test_mcp_http_auth.py
@@ -895,3 +895,27 @@ async def test_failed_connect_releases_the_seeded_credential(mcp_http_server: MC
         assert [request.headers.get("Authorization") for request in initializes] == ["token-rejected", None]
     finally:
         await tool.close()
+
+
+async def test_a_second_run_cannot_replace_an_unconnected_claim(mcp_http_server: MCPHTTPServer) -> None:
+    """The first run to seed owns the connection, even before its handshake completes.
+
+    is_connected only turns true after initialize returns, so it cannot by itself stop a
+    concurrent run from swapping the credential mid-handshake and authenticating the shared
+    connection as the wrong caller.
+    """
+    client, requests, _ = mcp_http_server
+    tool = _kwargs_dependent_tool(client)
+    tool._seed_connection_kwargs({"credential": "token-a"})
+    tool._seed_connection_kwargs({"credential": "token-b"})
+    try:
+        async with tool:
+            pass
+        initializes = [
+            request
+            for request in requests
+            if request.method == "POST" and json.loads(request.content).get("method") == "initialize"
+        ]
+        assert [request.headers.get("Authorization") for request in initializes] == ["token-a"]
+    finally:
+        await tool.close()

From 39cb9b1d2a13fe1284148e164edc1760b2bb137f Mon Sep 17 00:00:00 2001
From: Jose Alvarez <jpalvarezl@users.noreply.github.com>
Date: Thu, 10 Sep 2026 14:16:01 +0200
Subject: [PATCH 8/8] Update python/samples/02-agents/mcp/mcp_api_key_auth.py

Co-authored-by: Eduard van Valkenburg <eavanvalkenburg@users.noreply.github.com>
---
 python/samples/02-agents/mcp/mcp_api_key_auth.py | 2 +-
 1 file changed, 1 insertion(+), 1 deletion(-)

diff --git a/python/samples/02-agents/mcp/mcp_api_key_auth.py b/python/samples/02-agents/mcp/mcp_api_key_auth.py
index b38247f48ea..4109a0b435d 100644
--- a/python/samples/02-agents/mcp/mcp_api_key_auth.py
+++ b/python/samples/02-agents/mcp/mcp_api_key_auth.py
@@ -54,7 +54,7 @@ async def api_key_auth_example(api_key: str) -> None:
             name="MCP tool",
             description="MCP tool description.",
             url="<your authenticated server url>",
-            header_provider=lambda _kwargs: {"Authorization": f"Bearer {api_key}"},
+            header_provider=lambda _: {"Authorization": f"Bearer {api_key}"},
         ),
     ) as agent:
         query = "Use your MCP tool to tell me what tools are available to you."