Python: A2A AuthInterceptor Support (#1317)

* a2a authinterceptor support

* Update python/packages/a2a/agent_framework_a2a/_agent.py

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

* small fix

---------

Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com>

python/packages/a2a/agent_framework_a2a/_agent.py

@@ -9,6 +9,7 @@ from typing import Any, cast
 
 import httpx
 from a2a.client import Client, ClientConfig, ClientFactory, minimal_agent_card
+from a2a.client.auth.interceptor import AuthInterceptor
 from a2a.types import (
     AgentCard,
     Artifact,
@@ -78,6 +79,7 @@ class A2AAgent(BaseAgent):
         url: str | None = None,
         client: Client | None = None,
         http_client: httpx.AsyncClient | None = None,
+        auth_interceptor: AuthInterceptor | None = None,
         **kwargs: Any,
     ) -> None:
         """Initialize the A2AAgent.
@@ -90,6 +92,7 @@ class A2AAgent(BaseAgent):
             url: The URL for the A2A server.
             client: The A2A client for the agent.
             http_client: Optional httpx.AsyncClient to use.
+            auth_interceptor: Optional authentication interceptor for secured endpoints.
             kwargs: any additional properties, passed to BaseAgent.
         """
         super().__init__(id=id, name=name, description=description, **kwargs)
@@ -123,7 +126,8 @@ class A2AAgent(BaseAgent):
             supported_transports=[TransportProtocol.jsonrpc],
         )
         factory = ClientFactory(config)
-        self.client = factory.create(agent_card)
+        interceptors = [auth_interceptor] if auth_interceptor is not None else None
+        self.client = factory.create(agent_card, interceptors=interceptors)  # type: ignore
 
     async def __aenter__(self) -> "A2AAgent":
         """Async context manager entry."""

python/packages/a2a/tests/test_a2a_agent.py

@@ -497,3 +497,21 @@ def test_a2a_parts_to_contents_with_hosted_file_uri() -> None:
     assert isinstance(contents[0], UriContent)
     assert contents[0].uri == "hosted://storage/document.pdf"
     assert contents[0].media_type == ""  # Converted None to empty string
+
+
+def test_auth_interceptor_parameter() -> None:
+    """Test that auth_interceptor parameter is accepted without errors."""
+    # Create a mock auth interceptor
+    mock_auth_interceptor = MagicMock()
+
+    # Test that A2AAgent can be created with auth_interceptor parameter
+    # Using url parameter for simplicity
+    agent = A2AAgent(
+        name="test-agent",
+        url="https://test-agent.example.com",
+        auth_interceptor=mock_auth_interceptor,
+    )
+
+    # Verify the agent was created successfully
+    assert agent.name == "test-agent"
+    assert agent.client is not None