Python: [BREAKING] Guard Copilot runtime connections

docs/decisions/0046-python-copilot-runtime-connections.md

@@ -0,0 +1,75 @@
+---
+status: proposed
+contact: rogerbarreto
+date: 2026-10-22
+deciders: rogerbarreto, eavanvalkenburg, westey-m
+---
+
+# Let the Copilot agent own runtime connection configuration
+
+## Context and Problem Statement
+
+The Invocations host passes application-parsed options to agents. For the Copilot
+agent, provider settings, MCP settings, custom-agent MCP settings, session
+credentials, authentication controls, and discovery roots are SDK session
+configuration. Forwarding unfiltered request options can therefore change the
+connections selected for a session.
+
+The same option names can have different meanings for other agent implementations.
+An Invocations-wide denylist also rejects connection settings produced by trusted
+application parsers and option hooks. The compatibility concern is tracked in
+[PR #9611](https://github.com/microsoft/agent-framework/pull/9611).
+
+## Decision Drivers
+
+- Keep application-owned connections protected by default.
+- Preserve deliberate runtime connection configuration through explicit opt-in.
+- Keep ordinary per-run generation options and agent defaults available.
+- Apply the policy to direct calls and hosting without provider-specific host coupling.
+
+## Considered Options
+
+- Reserve connection names in Invocations. This rejects input early, but changes
+  every hosted agent's contract and also blocks trusted parser and hook output.
+- Disable all runtime options. This is simple, but unnecessarily removes model
+  and timeout overrides.
+- Keep runtime connections enabled by default. This preserves compatibility, but
+  leaves applications forwarding unfiltered options without default protection.
+- Make the Copilot agent reject runtime connections by default with constructor
+  opt-in. This scopes the policy to the component interpreting the configuration,
+  but still requires an explicit migration for existing Copilot runtime callers.
+
+## Decision Outcome
+
+Choose the agent-owned policy. Both `GitHubCopilotAgent` and
+`RawGitHubCopilotAgent` expose `allow_runtime_connection_overrides=False`.
+They reject per-run provider configuration, MCP configuration, session credentials,
+authentication controls, and discovery roots, including empty or null values, before
+SDK startup and again before assembling SDK session configuration. Application
+`default_options` remain allowed, runtime custom agents without MCP configuration
+remain supported, and ordinary runtime options retain their existing behavior. The
+constructor flag is never read from run options or forwarded to the SDK.
+
+The agents expose a side-effect-free `validate_run_options` preflight. Invocations
+uses it after resolving the request-scoped agent and before entering its async
+context or accessing session storage. Direct runs still validate at entry, and
+session assembly validates again.
+
+Trusted applications may opt in to preserve runtime connection precedence over
+defaults. This is an intentional behavioral compatibility change: existing callers
+must move connections into defaults or enable the constructor flag.
+
+The package bounds `github-copilot-sdk` at 1.0.25 because it is the latest
+reviewed session configuration surface. Widening the bound requires classifying new
+`create_session` and `resume_session` parameters and adding regression coverage before
+publication.
+
+Invocations no longer reserves these connection names for every agent. It reports
+the existing `AgentInvalidRequestException` as a sanitized HTTP 400 or typed SSE
+client error, while retaining generic provider-failure sanitization and legacy wire
+behavior. Other agents retain responsibility for their own option semantics.
+An updated host alone does not protect older Copilot package versions; release and
+deployment must include the updated agent policy.
+
+The proposed deciders include Python code owners; maintainer and engineering-management
+approval is still required.

python/packages/foundry_hosting/README.md

@@ -397,15 +397,25 @@ runtime options. It also rejects agent execution controls: `additional_function_
 arguments belong in developer-configured agent defaults or middleware/factories, not in request options or hook output.
 A hook may strip denied caller fields before validation; allowed generation and provider-specific options remain
 available. These restrictions apply to both wire formats and every `unsupported_options` policy.
+Connection option names such as `provider` and `mcp_servers` are not reserved host-wide.
+The hosted agent owns their policy. Agents may expose `validate_run_options(options)` for a side-effect-free
+preflight before the host enters a request-scoped agent context or reads or writes session storage. The updated
+`GitHubCopilotAgent` and `RawGitHubCopilotAgent` use this preflight to reject runtime connections by default while
+preserving connections in `default_options`. Trusted applications can enable them with the constructor-only
+`allow_runtime_connection_overrides=True` flag and filter request input through `prepare_options`.
+Update the Copilot package as well as the host to obtain that default-deny policy; older Copilot versions retain their
+existing runtime connection behavior.
 When an agent cannot accept runtime options, `unsupported_options="warn"` (default) logs and ignores
 them; `"ignore"` silently drops them and `"error"` rejects them. For request-scoped factories, unsupported options
 discovered after streaming starts are reported as an SSE `error` event with `status: 400`.
 
 **New default:** non-streaming success is JSON `{"response": "..."}`. Streaming success is real SSE
 `event: delta` with `{"text": "..."}`, followed by `event: done` with the platform sandbox `session_id`.
 The `done` event is sent only after the final MAF `ResponseStream` is finalized and the session is persisted.
-Client validation errors return JSON HTTP 400 before streaming where possible. Provider errors are logged and
-sanitized as JSON HTTP 500 or an SSE `error`; a cross-process ETag conflict is JSON HTTP 409 or an SSE `error`
+Client validation errors return JSON HTTP 400 before streaming where possible. An agent's `AgentInvalidRequestException`
+is reported as a sanitized JSON HTTP 400 or SSE `error` with `code: "invalid_request"` and `status: 400`.
+Provider errors are logged and sanitized as JSON HTTP 500 or an SSE `error`; a cross-process ETag conflict is
+JSON HTTP 409 or an SSE `error`
 with `code: "session_conflict"` and `status: 409`. A stream may emit deltas before an error. The host serializes
 same-session requests in one process, but a CAS conflict can still follow external tool effects in separate
 processes; it does not guarantee exactly-once execution. See the

python/packages/foundry_hosting/agent_framework_foundry_hosting/_invocations.py

@@ -35,6 +35,7 @@
     WorkflowExecutor,
 )
 from agent_framework._telemetry import mark_feature_used
+from agent_framework.exceptions import AgentInvalidRequestException
 from anyio import CancelScope
 from azure.ai.agentserver.core import (
     FoundryAgentRequestContext,
@@ -251,12 +252,22 @@ def _sse(event: str, data: Mapping[str, Any]) -> str:
 def _invocation_failure(exc: Exception) -> tuple[str, int, str]:
     cause: BaseException | None = exc
     while cause is not None:
+        if isinstance(cause, AgentInvalidRequestException):
+            return "The agent rejected the invocation request.", 400, "invalid_request"
         if isinstance(cause, (FoundryStorageConflictError, FoundryStoragePreconditionError)):
             return "Another request advanced this agent session; reload before retrying.", 409, "session_conflict"
         cause = cause.__cause__
     return "Agent invocation failed.", 500, "invocation_failed"
 
 
+def _contains_exception(cause: BaseException | None, exception_type: type[BaseException]) -> bool:
+    while cause is not None:
+        if isinstance(cause, exception_type):
+            return True
+        cause = cause.__cause__
+    return False
+
+
 def _workflow_failure(exc: Exception) -> tuple[str, int, str]:
     if isinstance(exc, _WorkflowOutputError):
         return str(exc), 500, "invalid_workflow_output"
@@ -517,12 +528,13 @@ def _warn_legacy_workflow(self) -> None:
             logger.warning("DEPRECATION: %s", message)
 
     @asynccontextmanager
-    async def _request_agent(self) -> AsyncGenerator[SupportsAgentRun]:
+    async def _request_agent(self, options: Mapping[str, Any]) -> AsyncGenerator[SupportsAgentRun]:
         if self._agent is None:
             raise RuntimeError("No agent is configured for Invocations.")
         agent = await resolve_agent(self._agent)
         if isinstance(agent, WorkflowAgent):
             self._warn_legacy_workflow()
+        await self._validate_agent_options(agent, options)
         async with AsyncExitStack() as resources:
             if self._owns_request_agent and isinstance(agent, AbstractAsyncContextManager):
                 await resources.enter_async_context(agent)
@@ -555,17 +567,18 @@ async def _request_session(
             yield session
         finally:
             failure = sys.exc_info()[1]
-            try:
-                await store.set(session_id, session)
-            except Exception as exc:
-                logger.exception("Failed to persist invocation session")
-                if failure is None:
-                    raise
-                if isinstance(failure, Exception):
-                    raise RuntimeError(
-                        f"Invocation failed: {str(failure) or type(failure).__name__}; "
-                        f"session persistence also failed: {str(exc) or type(exc).__name__}"
-                    ) from exc
+            if not _contains_exception(failure, AgentInvalidRequestException):
+                try:
+                    await store.set(session_id, session)
+                except Exception as exc:
+                    logger.exception("Failed to persist invocation session")
+                    if failure is None:
+                        raise
+                    if isinstance(failure, Exception):
+                        raise RuntimeError(
+                            f"Invocation failed: {str(failure) or type(failure).__name__}; "
+                            f"session persistence also failed: {str(exc) or type(exc).__name__}"
+                        ) from exc
 
     async def _parse(self, request: Request) -> InvocationRun | WorkflowTurn[Any]:
         if self._parse_request is not None:
@@ -623,6 +636,15 @@ def _agent_kwargs(self, agent: SupportsAgentRun, options: dict[str, Any]) -> dic
             return {}
         return {"options": options}
 
+    @staticmethod
+    async def _validate_agent_options(agent: SupportsAgentRun, options: Mapping[str, Any]) -> None:
+        validator = getattr(agent, "validate_run_options", None)
+        if not callable(validator):
+            return
+        result = validator(options)
+        if inspect.isawaitable(result):
+            await result
+
     @staticmethod
     async def _close_interrupted_stream(stream: object) -> None:
         if isinstance(stream, ResponseStream):
@@ -669,7 +691,7 @@ async def stream_response() -> AsyncGenerator[str]:
                 token: Token[FoundryAgentRequestContext] | None = set_request_context(context)
                 try:
                     lock = self._session_locks.setdefault(partition_key, asyncio.Lock())
-                    async with lock, self._request_agent() as agent:
+                    async with lock, self._request_agent(options) as agent:
                         run_kwargs = self._agent_kwargs(agent, options)
                         async with self._request_session(partition_key, context, hosted_scope=hosted_scope) as session:
                             stream = agent.run(parsed.messages, session=session, stream=True, **run_kwargs)
@@ -728,7 +750,7 @@ async def stream_response() -> AsyncGenerator[str]:
 
         try:
             lock = self._session_locks.setdefault(partition_key, asyncio.Lock())
-            async with lock, self._request_agent() as agent:
+            async with lock, self._request_agent(options) as agent:
                 run_kwargs = self._agent_kwargs(agent, options)
                 async with self._request_session(partition_key, context, hosted_scope=hosted_scope) as session:
                     response = await agent.run(parsed.messages, session=session, **run_kwargs)

python/packages/foundry_hosting/tests/test_invocations.py

@@ -67,6 +67,36 @@
 
 pytestmark = pytest.mark.filterwarnings("ignore:.*SessionStore is experimental.*")
 
+_COPILOT_RUNTIME_CONNECTION_OPTIONS: list[dict[str, Any]] = [
+    {"provider": {"type": "openai", "base_url": "https://example.com/api"}},
+    {
+        "providers": [{"name": "runtime", "type": "openai", "base_url": "https://example.com/api"}],
+        "models": [{"id": "runtime-model", "provider": "runtime"}],
+        "model": "runtime/runtime-model",
+    },
+    {"plugin_directories": ["C:\\caller-plugin"]},
+    {"agent_directories": ["C:\\caller-agents"]},
+    {"config_directory": "C:\\caller-config"},
+    {"enable_config_discovery": True},
+    {"working_directory": "C:\\caller-workspace"},
+    {"github_token": "caller-token"},
+    {"auth_client_id_metadata_url": "https://caller.example/oauth-client.json"},
+    {"mcp_oauth_token_storage": "persistent"},
+    {"github_mcp_tool_config": {"enable_all_tools": True}},
+    {"disabled_mcp_servers": ["application"]},
+    {"capi": {"enable_web_socket_responses": False}},
+    {"mcp_servers": {"application": {"type": "http", "url": "https://example.com/mcp", "tools": ["*"]}}},
+    {
+        "custom_agents": [
+            {
+                "name": "runtime",
+                "prompt": "Use the runtime MCP server.",
+                "mcp_servers": {"runtime": {"type": "http", "url": "https://example.com/mcp", "tools": ["*"]}},
+            }
+        ]
+    },
+]
+
 # region Helpers
 
 
@@ -944,6 +974,106 @@ def create_host() -> InvocationsHostServer:
         assert agent.calls == []
         assert provider.contexts == []
 
+    @pytest.mark.parametrize("stream", [False, True])
+    @pytest.mark.parametrize("source", ["request", "parser", "hook"])
+    @pytest.mark.parametrize("options", _COPILOT_RUNTIME_CONNECTION_OPTIONS)
+    async def test_connection_options_are_governed_by_the_agent(
+        self, stream: bool, source: str, options: dict[str, Any]
+    ) -> None:
+        agent = _make_agent(response_text="ok", stream_texts=["ok"])
+        server = InvocationsHostServer(
+            agent,
+            agent_session_store_provider=_SessionStoreProvider(_mock_session_store()),
+            parse_request=(lambda _request: InvocationRun(messages="hi", options=options, stream=stream))
+            if source == "parser"
+            else None,
+            prepare_options=(lambda _request, _options: options) if source == "hook" else None,
+        )
+        with _request_context(session_id="connection-options"):
+            response = await server._handle_invoke(  # pyright: ignore[reportPrivateUsage]
+                _make_request({"message": "hi", "stream": stream, "options": options if source == "request" else {}})
+            )
+            assert await _success_text(response) == "ok"
+
+        assert agent.calls[0]["options"] == options
+
+    @pytest.mark.parametrize("stream", [False, True])
+    @pytest.mark.parametrize("resume", [False, True])
+    @pytest.mark.parametrize("source", ["request", "parser", "hook"])
+    @pytest.mark.parametrize("agent_source", ["instance", "factory"])
+    @pytest.mark.parametrize("agent_type", ["RawGitHubCopilotAgent", "GitHubCopilotAgent"])
+    @pytest.mark.parametrize("options", _COPILOT_RUNTIME_CONNECTION_OPTIONS)
+    async def test_http_route_reports_copilot_connection_policy_rejections_as_client_errors(
+        self, stream: bool, resume: bool, source: str, agent_source: str, agent_type: str, options: dict[str, Any]
+    ) -> None:
+        copilot_agents = pytest.importorskip("agent_framework_github_copilot")
+        sdk_client = MagicMock()
+        sdk_client.start = AsyncMock()
+        sdk_client.create_session = AsyncMock(side_effect=AssertionError("create_session must not be called"))
+        sdk_client.resume_session = AsyncMock(side_effect=AssertionError("resume_session must not be called"))
+        agent = getattr(copilot_agents, agent_type)(client=sdk_client)
+        store = _mock_session_store()
+        if resume:
+            store.get.return_value = AgentSession(
+                session_id="connection-policy",
+                service_session_id="existing-session",
+            )
+        server = InvocationsHostServer(
+            agent if agent_source == "instance" else lambda: agent,
+            agent_session_store_provider=_SessionStoreProvider(store),
+            parse_request=(lambda _request: InvocationRun(messages="hi", options=options, stream=stream))
+            if source == "parser"
+            else None,
+            prepare_options=(lambda _request, _options: options) if source == "hook" else None,
+        )
+        async with httpx.AsyncClient(transport=httpx.ASGITransport(app=server), base_url="http://test") as client:
+            response = await client.post(
+                "/invocations",
+                params={"agent_session_id": f"connection-policy-{uuid.uuid4().hex}"},
+                json={"message": "hi", "stream": stream, "options": options if source == "request" else {}},
+            )
+
+        if stream:
+            assert response.status_code == 200
+            assert _parse_sse_events(response.text) == [
+                (
+                    "error",
+                    {
+                        "message": "The agent rejected the invocation request.",
+                        "code": "invalid_request",
+                        "status": 400,
+                    },
+                )
+            ]
+        else:
+            assert response.status_code == 400
+            assert response.json() == {"error": "The agent rejected the invocation request."}
+        sdk_client.start.assert_not_awaited()
+        sdk_client.create_session.assert_not_awaited()
+        sdk_client.resume_session.assert_not_awaited()
+        store.get.assert_not_awaited()
+        store.set.assert_not_awaited()
+
+    @pytest.mark.parametrize("stream", [False, True])
+    async def test_generation_options_preserve_application_connection_defaults(self, stream: bool) -> None:
+        agent = _make_agent(response_text="ok", stream_texts=["ok"])
+        agent.default_options = {
+            "mcp_servers": {"application": {"type": "http", "url": "https://example.com/mcp", "tools": ["*"]}},
+            "provider": {"type": "openai", "base_url": "https://example.com/api"},
+        }
+        server = InvocationsHostServer(agent)
+        with _request_context(session_id="generation-options"):
+            response = await server._handle_invoke(  # pyright: ignore[reportPrivateUsage]
+                _make_request({"message": "hi", "stream": stream, "options": {"model": "request-model"}})
+            )
+            assert await _success_text(response) == "ok"
+
+        assert agent.calls[0]["options"] == {"model": "request-model"}
+        assert agent.default_options == {
+            "mcp_servers": {"application": {"type": "http", "url": "https://example.com/mcp", "tools": ["*"]}},
+            "provider": {"type": "openai", "base_url": "https://example.com/api"},
+        }
+
     @pytest.mark.parametrize("result", [None, [], {1: "invalid"}, {"session_id": "forged"}])
     async def test_options_hook_must_return_safe_string_keyed_mapping(self, result: Any) -> None:
         agent = _make_agent(response_text="ok")