Python: reject unsupported Copilot provider function middleware

python/packages/github_copilot/README.md

@@ -10,6 +10,17 @@ pip install agent-framework-github-copilot
 
 The GitHub Copilot agent enables integration with GitHub Copilot, allowing you to interact with Copilot's agentic capabilities through the Agent Framework.
 
+## Context providers and function middleware
+
+Context providers can contribute messages, instructions, and tools to a run.
+`GitHubCopilotAgent` supports agent middleware, but the SDK owns its tool-calling loop
+and does not run Agent Framework function middleware. If a context provider contributes
+function middleware through `SessionContext.extend_middleware()`, the run raises
+`MiddlewareException` before creating or resuming an SDK session or registering its tools.
+This applies to both response modes and to `RawGitHubCopilotAgent`.
+
+Use `Agent` with a chat client when provider-contributed function middleware is required.
+
 ## Tool approval (`approval_mode="always_require"`)
 
 The GitHub Copilot SDK owns the tool-calling loop for this provider, so approval for
@@ -117,3 +128,19 @@ directory that defines hooks it is not loading. See
 [`github_copilot_with_file_hooks.py`](../../samples/02-agents/providers/github_copilot/github_copilot_with_file_hooks.py)
 for a runnable example.
 
+## MCP configuration and session lifetime
+
+Constructor `default_options` provide SDK session defaults. Per-run `options` override
+the corresponding defaults for that request, including `mcp_servers`. Reusing an
+`AgentSession` with a `service_session_id` resumes SDK/CLI-owned service-session state.
+Forwarding a narrower MCP configuration, such as changing a server's `tools` from
+`["*"]` to `[]`, is not an authorization or revocation guarantee for capabilities
+already resident in that session. Enforce authorization at the MCP service/tool boundary.
+
+Use a fresh service session when narrowing configured capabilities: create
+`fresh_session = agent.create_session()`, then pass it to
+`await agent.run("Start a new conversation", session=fresh_session, options=restricted_options)`,
+where `restricted_options` contains the narrowed `mcp_servers` configuration.
+The new session has no `service_session_id`; do not copy the previous service-session
+state or reuse its SDK `session_id` in constructor or per-run options. A fresh session
+starts a new conversation rather than continuing the previous one.

python/packages/github_copilot/agent_framework_github_copilot/_agent.py

@@ -43,6 +43,7 @@
     normalize_messages,
 )
 from agent_framework._mcp import MCPTool
+from agent_framework._middleware import categorize_middleware
 from agent_framework._settings import load_settings
 from agent_framework._telemetry import mark_feature_used
 from agent_framework._tools import FunctionTool, ToolTypes
@@ -51,7 +52,7 @@
     _get_data_bytes_as_str,  # pyright: ignore[reportPrivateUsage]
     normalize_tools,
 )
-from agent_framework.exceptions import AgentException, ContentError
+from agent_framework.exceptions import AgentException, ContentError, MiddlewareException
 from agent_framework.observability import AgentTelemetryLayer
 
 from ._feature_usage import FeatureIndex
@@ -891,6 +892,7 @@ def run(
 
         Raises:
             AgentException: If the request fails.
+            MiddlewareException: If a context provider contributes unsupported function middleware.
         """
         if middleware:
             logger.warning(
@@ -1098,6 +1100,7 @@ async def _stream_updates(
 
         Raises:
             AgentException: If the request fails.
+            MiddlewareException: If a context provider contributes unsupported function middleware.
         """
         if not self._started:
             await self.start()
@@ -1264,6 +1267,10 @@ async def _run_before_providers(
 
         Returns:
             The SessionContext with provider context populated.
+
+        Raises:
+            MiddlewareException: If a context provider contributes function middleware,
+                which cannot be applied to the SDK-owned tool loop.
         """
         session_context = SessionContext(
             session_id=session.session_id,
@@ -1282,6 +1289,12 @@ async def _run_before_providers(
                 state=session.state.setdefault(provider.source_id, {}),
             )
 
+        if categorize_middleware(session_context.get_middleware())["function"]:
+            raise MiddlewareException(
+                "GitHubCopilotAgent does not support function middleware contributed by context providers. "
+                "Use Agent with a chat client when function middleware is required."
+            )
+
         return session_context
 
     @staticmethod
@@ -1731,6 +1744,11 @@ def _build_session_kwargs(
         merged with the agent's tools and converted to SDK tools, and approval callbacks
         are turned into ``hooks``.
 
+        Forwarding options alone is not an authorization or revocation guarantee for
+        capabilities already resident in a resumed SDK/CLI service session. Enforce
+        authorization at the MCP service/tool boundary and use a fresh service session
+        when narrowing configured capabilities.
+
         Args:
             streaming: Whether to enable streaming for the session.
             runtime_options: Runtime options that take precedence over default_options.
@@ -1869,10 +1887,10 @@ class GitHubCopilotAgent(  # type: ignore[misc]
     RawGitHubCopilotAgent[OptionsT],
     Generic[OptionsT],
 ):
-    """A GitHub Copilot Agent with full middleware and telemetry support.
+    """A GitHub Copilot Agent with agent middleware and telemetry support.
 
     This is the recommended agent class for most use cases. It includes
-    middleware support and OpenTelemetry-based telemetry for observability,
+    agent middleware and OpenTelemetry-based telemetry for observability,
     with middleware running outside the telemetry span so middleware execution
     time is not captured in traces. For a minimal implementation without these
     layers, use :class:`RawGitHubCopilotAgent`.
@@ -1922,7 +1940,7 @@ def __init__(
         env_file_path: str | None = None,
         env_file_encoding: str | None = None,
     ) -> None:
-        """Initialize a GitHub Copilot Agent with full middleware and telemetry.
+        """Initialize a GitHub Copilot Agent with agent middleware and telemetry.
 
         Args:
             instructions: System message for the agent.

python/packages/github_copilot/tests/conftest.py

@@ -0,0 +1,145 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+from collections.abc import AsyncIterable, Awaitable, Callable, Mapping, Sequence
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+
+import pytest
+from agent_framework import (
+    BaseChatClient,
+    ChatOptions,
+    ChatResponse,
+    ChatResponseUpdate,
+    Content,
+    ContextProvider,
+    FunctionInvocationContext,
+    FunctionInvocationLayer,
+    FunctionMiddleware,
+    FunctionTool,
+    Message,
+    MiddlewareBundle,
+    MiddlewareTypes,
+    ResponseStream,
+    SessionContext,
+    function_middleware,
+    tool,
+)
+from copilot.session_events import SessionEvent
+
+
+@pytest.fixture
+def completed_copilot_session(
+    mock_session: MagicMock,
+    assistant_message_event: SessionEvent,
+    session_idle_event: SessionEvent,
+) -> MagicMock:
+    mock_session.send_and_wait.return_value = assistant_message_event
+
+    def subscribe(handler: Callable[[SessionEvent], None]) -> Callable[[], None]:
+        handler(session_idle_event)
+        return lambda: None
+
+    mock_session.on.side_effect = subscribe
+    return mock_session
+
+
+@pytest.fixture(params=["class", "decorated", "annotated", "bundle"])
+def provider_function_middleware(
+    request: pytest.FixtureRequest,
+) -> tuple[ContextProvider, FunctionTool, AsyncMock, AsyncMock]:
+    tool_body = AsyncMock()
+    policy = AsyncMock(side_effect=RuntimeError("Function policy denied the call"))
+
+    @tool(approval_mode="never_require")
+    async def provider_tool(value: str) -> str:
+        """Return the supplied value."""
+        await tool_body(value)
+        return value
+
+    async def deny_call(
+        context: FunctionInvocationContext,
+        call_next: Callable[[], Awaitable[None]],
+    ) -> None:
+        await policy(context)
+
+    class DenyMiddleware(FunctionMiddleware):
+        async def process(
+            self,
+            context: FunctionInvocationContext,
+            call_next: Callable[[], Awaitable[None]],
+        ) -> None:
+            await policy(context)
+
+    middleware: MiddlewareTypes
+    if request.param == "class":
+        middleware = DenyMiddleware()
+    elif request.param == "decorated":
+        middleware = function_middleware(deny_call)
+    elif request.param == "annotated":
+        middleware = deny_call
+    else:
+        middleware = MiddlewareBundle([deny_call])
+
+    class ToolProvider(ContextProvider):
+        async def before_run(
+            self,
+            *,
+            agent: Any,
+            session: Any,
+            context: SessionContext,
+            state: dict[str, Any],
+        ) -> None:
+            context.extend_tools(self.source_id, [provider_tool])
+            if state.get("include_middleware", True):
+                context.extend_middleware(self.source_id, middleware)
+
+    return ToolProvider(source_id="tool-policy"), provider_tool, tool_body, policy
+
+
+@pytest.fixture
+def tool_calling_chat_client(
+    provider_function_middleware: tuple[ContextProvider, FunctionTool, AsyncMock, AsyncMock],
+) -> BaseChatClient[ChatOptions[Any]]:
+    _, provider_tool, _, _ = provider_function_middleware
+
+    class ToolCallingClient(FunctionInvocationLayer[ChatOptions[Any]], BaseChatClient[ChatOptions[Any]]):
+        def _inner_get_response(
+            self,
+            *,
+            messages: Sequence[Message],
+            stream: bool,
+            options: Mapping[str, Any],
+            **kwargs: Any,
+        ) -> Awaitable[ChatResponse] | ResponseStream[ChatResponseUpdate, ChatResponse]:
+            has_result = any(content.type == "function_result" for message in messages for content in message.contents)
+            contents = (
+                [Content.from_text("Done")]
+                if has_result
+                else [
+                    Content.from_function_call(
+                        call_id="provider-call",
+                        name=provider_tool.name,
+                        arguments={"value": "test"},
+                    )
+                ]
+            )
+
+            if stream:
+
+                async def updates() -> AsyncIterable[ChatResponseUpdate]:
+                    yield ChatResponseUpdate(
+                        role="assistant",
+                        contents=contents,
+                        finish_reason="stop" if has_result else "tool_calls",
+                    )
+
+                return ResponseStream(updates(), finalizer=ChatResponse.from_updates)
+
+            async def response() -> ChatResponse:
+                return ChatResponse(messages=[Message(role="assistant", contents=contents)])
+
+            return response()
+
+    return ToolCallingClient()