Introduce simple server request auth structures

Author: mikeas1

src/a2a/auth/user.py

@@ -1,4 +1,29 @@
-class User:
+"""Authenticated user information."""
+
+from abc import ABC, abstractmethod
+
+
+class User(ABC):
     """A representation of an authenticated user."""
-    def __init__(self):
-        pass
+
+    @abstractmethod
+    @property
+    def is_authenticated(self) -> bool:
+        """Returns whether the current user is authenticated."""
+
+    @abstractmethod
+    @property
+    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 ''

src/a2a/server/apps/starlette_app.py

@@ -1,32 +1,52 @@
+import contextlib
 import json
 import logging
 import traceback
+
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator
 from typing import Any
 
 from pydantic import ValidationError
 from sse_starlette.sse import EventSourceResponse
 from starlette.applications import Starlette
+from starlette.authentication import BaseUser
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 from starlette.routing import Route
 
+from a2a.auth.user import User as A2AUser
 from a2a.server.context import ServerCallContext
 from a2a.server.request_handlers.jsonrpc_handler import JSONRPCHandler
 from a2a.server.request_handlers.request_handler import RequestHandler
-from a2a.types import (A2AError, A2ARequest, AgentCard, CancelTaskRequest,
-                       GetTaskPushNotificationConfigRequest, GetTaskRequest,
-                       InternalError, InvalidRequestError, JSONParseError,
-                       JSONRPCError, JSONRPCErrorResponse, JSONRPCResponse,
-                       SendMessageRequest, SendStreamingMessageRequest,
-                       SendStreamingMessageResponse,
-                       SetTaskPushNotificationConfigRequest,
-                       TaskResubscriptionRequest, UnsupportedOperationError)
+from a2a.types import (
+    A2AError,
+    A2ARequest,
+    AgentCard,
+    CancelTaskRequest,
+    GetTaskPushNotificationConfigRequest,
+    GetTaskRequest,
+    InternalError,
+    InvalidRequestError,
+    JSONParseError,
+    JSONRPCError,
+    JSONRPCErrorResponse,
+    JSONRPCResponse,
+    SendMessageRequest,
+    SendStreamingMessageRequest,
+    SendStreamingMessageResponse,
+    SetTaskPushNotificationConfigRequest,
+    TaskResubscriptionRequest,
+    UnsupportedOperationError,
+)
 from a2a.utils.errors import MethodNotImplementedError
 
+
 logger = logging.getLogger(__name__)
 
+# Register Starlette User as an implementation of a2a.auth.user.User
+A2AUser.register(BaseUser)
+
 
 class CallContextBuilder(ABC):
     """A class for building ServerCallContexts using the Starlette Request."""
@@ -36,6 +56,18 @@ def build(self, request: Request) -> ServerCallContext:
         """Builds a ServerCallContext from a Starlette Request."""
 
 
+class DefaultCallContextBuilder(CallContextBuilder):
+    """A default implementation of CallContextBuilder."""
+
+    def build(self, request: Request) -> ServerCallContext:
+        user = None
+        state = {}
+        with contextlib.suppress(Exception):
+            user = request.user
+            state['auth'] = request.auth
+        return ServerCallContext(user=user, state=state)
+
+
 class A2AStarletteApplication:
     """A Starlette application implementing the A2A protocol server endpoints.
 
@@ -75,7 +107,7 @@ def __init__(
             logger.error(
                 'AgentCard.supportsAuthenticatedExtendedCard is True, but no extended_agent_card was provided. The /agent/authenticatedExtendedCard endpoint will return 404.'
             )
-        self._context_builder = context_builder
+        self._context_builder = context_builder or DefaultCallContextBuilder()
 
     def _generate_error_response(
         self, request_id: str | int | None, error: JSONRPCError | A2AError
@@ -137,11 +169,7 @@ async def _handle_requests(self, request: Request) -> Response:
         try:
             body = await request.json()
             a2a_request = A2ARequest.model_validate(body)
-            call_context = (
-                self._context_builder.build(request)
-                if self._context_builder
-                else None
-            )
+            call_context = self._context_builder.build(request)
 
             request_id = a2a_request.root.id
             request_obj = a2a_request.root
@@ -344,7 +372,9 @@ async def _handle_get_authenticated_extended_agent_card(
         # extended_agent_card was provided during server initialization,
         # return a 404
         return JSONResponse(
-            {'error': 'Authenticated extended agent card is supported but not configured on the server.'},
+            {
+                'error': 'Authenticated extended agent card is supported but not configured on the server.'
+            },
             status_code=404,
         )
 

src/a2a/server/context.py

@@ -3,6 +3,8 @@
 import collections.abc
 import typing
 
+from a2a.auth.user import UnauthenticatedUser, User
+
 
 State = collections.abc.MutableMapping[str, typing.Any]
 
@@ -13,10 +15,14 @@ class ServerCallContext:
     This class allows storing arbitrary user data in the state attribute.
     """
 
-    def __init__(self, state: State | None = None):
-        if state is None:
-            state = {}
-        self._state = state
+    def __init__(self, state: State | None = None, user: User | None = None):
+        self._state = state or {}
+        self._user = user or UnauthenticatedUser()
+
+    @property
+    def user(self) -> User:
+        """Get the user associated with this context, or UnauthenticatedUser."""
+        return self._user
 
     @property
     def state(self) -> State: