Python: Added AgentMiddleware to Copilot and Claude agents

[dmytrostruk]

Motivation and Context

Added AgentMiddleware support to Copilot and Claude agents.

Contribution Checklist

• The code builds clean without any errors or warnings
• The PR follows the Contribution Guidelines
• All unit tests pass, and I have added new tests where possible
• Is this a breaking change? If yes, add "[BREAKING]" prefix to the title of the PR.

[markwallace-microsoft]

Python Test Coverage Report •

File	Stmts	Miss	Cover	Missing
TOTAL	22700	2601	88%

report-only-changed-files is enabled. No files were changed during this commit :)

Python Unit Test Overview

Tests	Skipped	Failures	Errors	Time
4964	20 💤	0 ❌	0 🔥	1m 19s ⏱️

[Copilot]

Pull request overview

Adds Agent Framework AgentMiddleware support to the Python GitHub Copilot and Claude agents by layering AgentMiddlewareLayer into their public agent types, with new unit tests validating middleware execution and streaming transform hooks.

Changes:

• Wrap GitHub Copilot agent with AgentMiddlewareLayer via a new GitHubCopilotAgent class and rename the core implementation to RawGitHubCopilotAgent.
• Add AgentMiddlewareLayer to ClaudeAgent (telemetry-enabled) and add middleware-focused tests for Claude and Copilot.
• Refactor streaming/non-streaming run() paths in both agents to centralize stream construction.

Reviewed changes

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

File	Description
python/packages/github_copilot/agent_framework_github_copilot/_agent.py	Introduces middleware-enabled GitHubCopilotAgent wrapper and refactors run()/streaming helpers.
python/packages/github_copilot/tests/test_github_copilot_agent.py	Adds middleware execution + streaming transform hook tests for Copilot agent.
python/packages/claude/agent_framework_claude/_agent.py	Adds AgentMiddlewareLayer to ClaudeAgent and refactors stream construction; updates docs/signatures.
python/packages/claude/tests/test_claude_agent.py	Adds middleware execution + streaming transform hook tests for Claude agent.

Comments suppressed due to low confidence (1)

python/packages/claude/agent_framework_claude/_agent.py:739

• ClaudeAgent inherits AgentTelemetryLayer before AgentMiddlewareLayer, so the effective run() signature comes from AgentTelemetryLayer.run() (which doesn't surface the middleware parameter in its overloads). This makes the new middleware support hard to discover/type-check for users. Consider either (a) swapping the inheritance order so AgentMiddlewareLayer.run() is the visible signature, or (b) adding explicit run() overloads on ClaudeAgent that include middleware and forward to super().run().

class ClaudeAgent(
    AgentTelemetryLayer,
    AgentMiddlewareLayer,
    RawClaudeAgent[OptionsT],
    Generic[OptionsT],
):
    """Claude Agent with OpenTelemetry instrumentation.

[Copilot]

Pull request overview

Copilot reviewed 4 out of 4 changed files in this pull request and generated 2 comments.

Comments suppressed due to low confidence (3)

python/packages/claude/agent_framework_claude/_agent.py:628

• For consistency with the rest of the framework (e.g., RawAgent.run(..., options=...)), consider adding an explicit kw-only options parameter to RawClaudeAgent.run() instead of only pulling it from kwargs. This makes the supported runtime options discoverable via type hints and avoids relying on kwargs.pop('options') for normal use.

    @overload
    def run(
        self,
        messages: AgentRunInputs | None = None,
        *,
        stream: Literal[False] = ...,
        session: AgentSession | None = None,
        middleware: Sequence[AgentMiddlewareTypes] | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse[Any]]: ...

    @overload
    def run(
        self,
        messages: AgentRunInputs | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        middleware: Sequence[AgentMiddlewareTypes] | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse[Any]]: ...

    def run(
        self,
        messages: AgentRunInputs | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        middleware: Sequence[AgentMiddlewareTypes] | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse[Any]] | ResponseStream[AgentResponseUpdate, AgentResponse[Any]]:
        """Run the agent with the given messages.

        Args:
            messages: The messages to process.

        Keyword Args:
            stream: If True, returns an async iterable of updates. If False (default),
                returns an awaitable AgentResponse.
            session: The conversation session. If session has service_session_id set,
                the agent will resume that session.
            middleware: Optional per-run AgentMiddleware applied on top of constructor middleware.
            kwargs: Additional keyword arguments including 'options' for runtime options
                (model, permission_mode can be changed per-request).

        Returns:
            When stream=True: An ResponseStream for streaming updates.
            When stream=False: An Awaitable[AgentResponse] with the complete response.
        """
        options = kwargs.pop("options", None)
        response = self._run_stream_impl(messages=messages, session=session, options=options, **kwargs)

python/packages/github_copilot/agent_framework_github_copilot/_agent.py:132

• RawGitHubCopilotAgent is described like the primary public agent (“A GitHub Copilot Agent…”) and its examples refer to GitHubCopilotAgent, but the class name implies it is the lower-level/no-layers variant. Consider updating this docstring to explicitly state what is “raw” about it and to direct most users to GitHubCopilotAgent for middleware/telemetry support.

This issue also appears on line 300 of the same file.

class RawGitHubCopilotAgent(BaseAgent, Generic[OptionsT]):
    """A GitHub Copilot Agent.

    This agent wraps the GitHub Copilot SDK to provide Copilot agentic capabilities
    within the Agent Framework. It supports both streaming and non-streaming responses,
    custom tools, and session management.

python/packages/github_copilot/agent_framework_github_copilot/_agent.py:357

• run() no longer has an explicit options kw-only parameter (it’s now only extracted from kwargs). This reduces IDE autocomplete and makes this agent inconsistent with the core RawAgent.run(..., options=...) API. Consider reintroducing an explicit options: OptionsT | None = None kw-only parameter (and keep the kwargs.pop('options', ...) fallback if you need backward compatibility).

    @overload
    def run(
        self,
        messages: AgentRunInputs | None = None,
        *,
        stream: Literal[False] = False,
        session: AgentSession | None = None,
        middleware: Sequence[AgentMiddlewareTypes] | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse]: ...

    @overload
    def run(
        self,
        messages: AgentRunInputs | None = None,
        *,
        stream: Literal[True],
        session: AgentSession | None = None,
        middleware: Sequence[AgentMiddlewareTypes] | None = None,
        **kwargs: Any,
    ) -> ResponseStream[AgentResponseUpdate, AgentResponse]: ...

    def run(
        self,
        messages: AgentRunInputs | None = None,
        *,
        stream: bool = False,
        session: AgentSession | None = None,
        middleware: Sequence[AgentMiddlewareTypes] | None = None,
        **kwargs: Any,
    ) -> Awaitable[AgentResponse] | ResponseStream[AgentResponseUpdate, AgentResponse]:
        """Get a response from the agent.

        This method returns the final result of the agent's execution
        as a single AgentResponse object when stream=False. When stream=True,
        it returns a ResponseStream that yields AgentResponseUpdate objects.

        Args:
            messages: The message(s) to send to the agent.

        Keyword Args:
            stream: Whether to stream the response. Defaults to False.
            session: The conversation session associated with the message(s).
            middleware: Runtime middleware parameter accepted for compatibility with middleware layer routing.
            kwargs: Additional keyword arguments, including ``options`` for runtime options
                (model, timeout, etc.).

        Returns:
            When stream=False: An Awaitable[AgentResponse].
            When stream=True: A ResponseStream of AgentResponseUpdate items.

        Raises:
            AgentException: If the request fails.
        """
        options = cast(OptionsT | None, kwargs.pop("options", None))
        if stream:
            return self._run_stream_impl(messages=messages, session=session, options=options, **kwargs)
        return self._run_impl(messages=messages, session=session, options=options, **kwargs)