Python: Fixed use_agent_middleware calling private _normalize_messages (#3264)

* Fix use_agent_middleware calling private _normalize_messages

* Fixed A2A and Copilot Studio agent

python/packages/core/tests/core/test_middleware_with_agent.py

@@ -1902,3 +1902,59 @@ class TestChatAgentChatMiddleware:
         assert modified_kwargs["max_tokens"] == 500
         assert modified_kwargs["new_param"] == "added_by_middleware"
         assert modified_kwargs["custom_param"] == "test_value"  # Should still be there
+
+
+class TestMiddlewareWithProtocolOnlyAgent:
+    """Test use_agent_middleware with agents implementing only AgentProtocol."""
+
+    async def test_middleware_with_protocol_only_agent(self) -> None:
+        """Verify middleware works without BaseAgent inheritance for both run and run_stream."""
+        from collections.abc import AsyncIterable
+
+        from agent_framework import AgentProtocol, AgentResponse, AgentResponseUpdate, use_agent_middleware
+
+        execution_order: list[str] = []
+
+        class TrackingMiddleware(AgentMiddleware):
+            async def process(
+                self, context: AgentRunContext, next: Callable[[AgentRunContext], Awaitable[None]]
+            ) -> None:
+                execution_order.append("before")
+                await next(context)
+                execution_order.append("after")
+
+        @use_agent_middleware
+        class ProtocolOnlyAgent:
+            """Minimal agent implementing only AgentProtocol, not inheriting from BaseAgent."""
+
+            def __init__(self):
+                self.id = "protocol-only-agent"
+                self.name = "Protocol Only Agent"
+                self.description = "Test agent"
+                self.middleware = [TrackingMiddleware()]
+
+            async def run(self, messages=None, *, thread=None, **kwargs) -> AgentResponse:
+                return AgentResponse(messages=[ChatMessage(role=Role.ASSISTANT, text="response")])
+
+            def run_stream(self, messages=None, *, thread=None, **kwargs) -> AsyncIterable[AgentResponseUpdate]:
+                async def _stream():
+                    yield AgentResponseUpdate()
+
+                return _stream()
+
+            def get_new_thread(self, **kwargs):
+                return None
+
+        agent = ProtocolOnlyAgent()
+        assert isinstance(agent, AgentProtocol)
+
+        # Test run (non-streaming)
+        response = await agent.run("test message")
+        assert response is not None
+        assert execution_order == ["before", "after"]
+
+        # Test run_stream (streaming)
+        execution_order.clear()
+        async for _ in agent.run_stream("test message"):
+            pass
+        assert execution_order == ["before", "after"]