tests working

Author: dmandar

pyproject.toml

@@ -70,6 +70,7 @@ dev = [
     "pytest-asyncio>=0.26.0",
     "pytest-cov>=6.1.1",
     "pytest-mock>=3.14.0",
+    "respx>=0.20.2",
     "ruff>=0.11.6",
     "uv-dynamic-versioning>=0.8.2",
 ]

src/a2a/client/__init__.py

@@ -1,19 +1,25 @@
+# a2a/client/__init__.py
+
 """Client-side components for interacting with an A2A agent."""
 
+from a2a.client.auth import (AuthInterceptor, CredentialService,
+                             InMemoryContextCredentialStore)
 from a2a.client.client import A2ACardResolver, A2AClient
-from a2a.client.errors import (
-    A2AClientError,
-    A2AClientHTTPError,
-    A2AClientJSONError,
-)
+from a2a.client.errors import (A2AClientError, A2AClientHTTPError,
+                               A2AClientJSONError)
 from a2a.client.helpers import create_text_message_object
-
+from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
 
 __all__ = [
     'A2ACardResolver',
     'A2AClient',
     'A2AClientError',
     'A2AClientHTTPError',
     'A2AClientJSONError',
+    'AuthInterceptor',
+    'ClientCallContext',
+    'ClientCallInterceptor',
+    'CredentialService',
+    'InMemoryContextCredentialStore',
     'create_text_message_object',
-]
+]

src/a2a/client/auth/__init__.py

@@ -0,0 +1,10 @@
+"""Client-side authentication components for the A2A Python SDK."""
+
+from .credentials import CredentialService, InMemoryContextCredentialStore
+from .interceptor import AuthInterceptor
+
+__all__ = [
+    'CredentialService',
+    'InMemoryContextCredentialStore',
+    'AuthInterceptor',
+]

src/a2a/client/auth/credentials.py

@@ -0,0 +1,52 @@
+# a2a/client/auth/credentials.py
+
+from abc import ABC, abstractmethod
+
+from a2a.client.middleware import ClientCallContext
+
+
+class CredentialService(ABC):
+    """An abstract service for retrieving credentials."""
+
+    @abstractmethod
+    async def get_credentials(
+        self,
+        security_scheme_name: str,
+        context: ClientCallContext | None,
+    ) -> str | None:
+        """
+        Retrieves a credential (e.g., token) for a security scheme.
+        """
+        pass
+
+
+class InMemoryContextCredentialStore(CredentialService):
+    """
+    A simple in-memory store for context-keyed credentials.
+
+    This class uses the 'contextId' from the ClientCallContext state to
+    store and retrieve credentials, providing a simple, user-specific
+    credential mechanism without requiring a full user authentication system.
+    """
+
+    def __init__(self):
+        # {context_id: {scheme_name: credential}}
+        self._store: dict[str, dict[str, str]] = {}
+
+    async def get_credentials(
+        self,
+        security_scheme_name: str,
+        context: ClientCallContext | None,
+    ) -> str | None:
+        if not context or 'contextId' not in context.state:
+            return None
+        context_id = context.state['contextId']
+        return self._store.get(context_id, {}).get(security_scheme_name)
+
+    async def set_credentials(
+        self, context_id: str, security_scheme_name: str, credential: str
+    ):
+        """Method to populate the store."""
+        if context_id not in self._store:
+            self._store[context_id] = {}
+        self._store[context_id][security_scheme_name] = credential

src/a2a/client/auth/interceptor.py

@@ -0,0 +1,61 @@
+# a2a/client/auth/interceptor.py
+
+import logging
+from typing import Any
+
+from a2a.client.auth.credentials import CredentialService
+from a2a.client.middleware import ClientCallContext, ClientCallInterceptor
+from a2a.types import AgentCard, APIKeySecurityScheme, HTTPAuthSecurityScheme
+
+logger = logging.getLogger(__name__)
+
+
+class AuthInterceptor(ClientCallInterceptor):
+    """
+    An interceptor that automatically adds authentication details to requests
+    based on the agent's security schemes.
+    """
+
+    def __init__(self, credential_service: CredentialService):
+        self._credential_service = credential_service
+
+    async def intercept(
+        self,
+        method_name: str,
+        request_payload: dict[str, Any],
+        http_kwargs: dict[str, Any],
+        agent_card: AgentCard | None,
+        context: ClientCallContext | None,
+    ) -> tuple[dict[str, Any], dict[str, Any]]:
+        """
+        Adds authentication headers to the request if credentials can be found.
+        """
+        if not agent_card or not agent_card.security or not agent_card.securitySchemes:
+            return request_payload, http_kwargs
+
+        for requirement in agent_card.security:
+            for scheme_name in requirement:
+                credential = await self._credential_service.get_credentials(
+                    scheme_name, context
+                )
+                if credential and scheme_name in agent_card.securitySchemes:
+                    scheme_def = agent_card.securitySchemes[scheme_name].root
+                    headers = http_kwargs.get('headers', {})
+
+                    if isinstance(scheme_def, HTTPAuthSecurityScheme):
+                        headers['Authorization'] = f"{scheme_def.scheme} {credential}"
+                        http_kwargs['headers'] = headers
+                        logger.debug(f"Added HTTP Auth for scheme '{scheme_name}'.")
+                        return request_payload, http_kwargs
+                    elif isinstance(scheme_def, APIKeySecurityScheme):
+                        if scheme_def.in_ == 'header':
+                            headers[scheme_def.name] = credential
+                            http_kwargs['headers'] = headers
+                            logger.debug(f"Added API Key Header for scheme '{scheme_name}'.")
+                            return request_payload, http_kwargs
+                        else:
+                            logger.warning(
+                                f"API Key in '{scheme_def.in_}' not supported by this interceptor."
+                            )
+
+        return request_payload, http_kwargs

src/a2a/client/auth/user.py

@@ -0,0 +1,29 @@
+"""Authenticated user information."""
+
+from abc import ABC, abstractmethod
+
+
+class User(ABC):
+    """A representation of an authenticated user."""
+
+    @property
+    @abstractmethod
+    def is_authenticated(self) -> bool:
+        """Returns whether the current user is authenticated."""
+
+    @property
+    @abstractmethod
+    def user_name(self) -> str:
+        """Returns the user name of the current user."""
+
+
+class UnauthenticatedUser(User):
+    """A representation that no user has been authenticated in the request."""
+
+    @property
+    def is_authenticated(self):
+        return False
+
+    @property
+    def user_name(self) -> str:
+        return ''