refactor(server)!: add build_user function to DefaultContextBuilder to allow A2A user creation customization (#925)

# Description

- Add build_user function to the DefaultContextBuilder to allow user
customization
- Renamed CallContextBuilder / DefaultCallContextBuilder ->
ServerCallContextBuilder / DefaultServerCallContextBuilder
- Centralizes HTTP context-building logic into a new
a2a.server.routes.common module, eliminating duplication between the
JSON-RPC and REST dispatchers.

Fixes #924  🦕

Author: guglielmo-san

src/a2a/compat/v0_3/grpc_handler.py

@@ -23,8 +23,8 @@
 from a2a.server.context import ServerCallContext
 from a2a.server.request_handlers.grpc_handler import (
     _ERROR_CODE_MAP,
-    CallContextBuilder,
-    DefaultCallContextBuilder,
+    DefaultGrpcServerCallContextBuilder,
+    GrpcServerCallContextBuilder,
 )
 from a2a.server.request_handlers.request_handler import RequestHandler
 from a2a.types.a2a_pb2 import AgentCard
@@ -44,7 +44,7 @@ def __init__(
         self,
         agent_card: AgentCard,
         request_handler: RequestHandler,
-        context_builder: CallContextBuilder | None = None,
+        context_builder: GrpcServerCallContextBuilder | None = None,
         card_modifier: Callable[[AgentCard], Awaitable[AgentCard] | AgentCard]
         | None = None,
     ):
@@ -61,7 +61,9 @@ def __init__(
         """
         self.agent_card = agent_card
         self.handler03 = RequestHandler03(request_handler=request_handler)
-        self.context_builder = context_builder or DefaultCallContextBuilder()
+        self._context_builder = (
+            context_builder or DefaultGrpcServerCallContextBuilder()
+        )
         self.card_modifier = card_modifier
 
     async def _handle_unary(
@@ -72,7 +74,7 @@ async def _handle_unary(
     ) -> TResponse:
         """Centralized error handling and context management for unary calls."""
         try:
-            server_context = self.context_builder.build(context)
+            server_context = self._context_builder.build(context)
             result = await handler_func(server_context)
             self._set_extension_metadata(context, server_context)
         except A2AError as e:
@@ -88,7 +90,7 @@ async def _handle_stream(
     ) -> AsyncIterable[TResponse]:
         """Centralized error handling and context management for streaming calls."""
         try:
-            server_context = self.context_builder.build(context)
+            server_context = self._context_builder.build(context)
             async for item in handler_func(server_context):
                 yield item
             self._set_extension_metadata(context, server_context)

src/a2a/compat/v0_3/jsonrpc_adapter.py

@@ -11,7 +11,6 @@
     from starlette.requests import Request
 
     from a2a.server.request_handlers.request_handler import RequestHandler
-    from a2a.server.routes import CallContextBuilder
     from a2a.types.a2a_pb2 import AgentCard
 
     _package_starlette_installed = True
@@ -38,6 +37,10 @@
 from a2a.server.jsonrpc_models import (
     JSONRPCError as CoreJSONRPCError,
 )
+from a2a.server.routes.common import (
+    DefaultServerCallContextBuilder,
+    ServerCallContextBuilder,
+)
 from a2a.utils import constants
 from a2a.utils.errors import ExtendedAgentCardNotConfiguredError
 from a2a.utils.helpers import maybe_await, validate_version
@@ -67,7 +70,7 @@ def __init__(  # noqa: PLR0913
         agent_card: 'AgentCard',
         http_handler: 'RequestHandler',
         extended_agent_card: 'AgentCard | None' = None,
-        context_builder: 'CallContextBuilder | None' = None,
+        context_builder: 'ServerCallContextBuilder | None' = None,
         card_modifier: 'Callable[[AgentCard], Awaitable[AgentCard] | AgentCard] | None' = None,
         extended_card_modifier: 'Callable[[AgentCard, ServerCallContext], Awaitable[AgentCard] | AgentCard] | None' = None,
     ):
@@ -78,7 +81,9 @@ def __init__(  # noqa: PLR0913
         self.handler = RequestHandler03(
             request_handler=http_handler,
         )
-        self._context_builder = context_builder
+        self._context_builder = (
+            context_builder or DefaultServerCallContextBuilder()
+        )
 
     def supports_method(self, method: str) -> bool:
         """Returns True if the v0.3 adapter supports the given method name."""
@@ -126,11 +131,7 @@ async def handle_request(
                     CoreInvalidRequestError(data=str(e)),
                 )
 
-            call_context = (
-                self._context_builder.build(request)
-                if self._context_builder
-                else ServerCallContext()
-            )
+            call_context = self._context_builder.build(request)
             call_context.tenant = (
                 getattr(specific_request.params, 'tenant', '')
                 if hasattr(specific_request, 'params')

src/a2a/compat/v0_3/rest_adapter.py

@@ -34,7 +34,10 @@
 from a2a.compat.v0_3 import conversions
 from a2a.compat.v0_3.rest_handler import REST03Handler
 from a2a.server.context import ServerCallContext
-from a2a.server.routes import CallContextBuilder, DefaultCallContextBuilder
+from a2a.server.routes.common import (
+    DefaultServerCallContextBuilder,
+    ServerCallContextBuilder,
+)
 from a2a.utils.error_handlers import (
     rest_error_handler,
     rest_stream_error_handler,
@@ -60,7 +63,7 @@ def __init__(  # noqa: PLR0913
         agent_card: 'AgentCard',
         http_handler: 'RequestHandler',
         extended_agent_card: 'AgentCard | None' = None,
-        context_builder: 'CallContextBuilder | None' = None,
+        context_builder: 'ServerCallContextBuilder | None' = None,
         card_modifier: 'Callable[[AgentCard], Awaitable[AgentCard] | AgentCard] | None' = None,
         extended_card_modifier: 'Callable[[AgentCard, ServerCallContext], Awaitable[AgentCard] | AgentCard] | None' = None,
     ):
@@ -71,7 +74,9 @@ def __init__(  # noqa: PLR0913
         self.handler = REST03Handler(
             agent_card=agent_card, request_handler=http_handler
         )
-        self._context_builder = context_builder or DefaultCallContextBuilder()
+        self._context_builder = (
+            context_builder or DefaultServerCallContextBuilder()
+        )
 
     @rest_error_handler
     async def _handle_request(

src/a2a/server/request_handlers/__init__.py

@@ -19,7 +19,9 @@
 
 try:
     from a2a.server.request_handlers.grpc_handler import (
+        DefaultGrpcServerCallContextBuilder,
         GrpcHandler,  # type: ignore
+        GrpcServerCallContextBuilder,
     )
 except ImportError as e:
     _original_error = e
@@ -39,8 +41,10 @@ def __init__(self, *args, **kwargs):
 
 
 __all__ = [
+    'DefaultGrpcServerCallContextBuilder',
     'DefaultRequestHandler',
     'GrpcHandler',
+    'GrpcServerCallContextBuilder',
     'RequestHandler',
     'build_error_response',
     'prepare_response_object',

src/a2a/server/request_handlers/grpc_handler.py

@@ -24,7 +24,7 @@
 import a2a.types.a2a_pb2_grpc as a2a_grpc
 
 from a2a import types
-from a2a.auth.user import UnauthenticatedUser
+from a2a.auth.user import UnauthenticatedUser, User
 from a2a.extensions.common import (
     HTTP_EXTENSION_HEADER,
     get_requested_extensions,
@@ -41,15 +41,32 @@
 
 logger = logging.getLogger(__name__)
 
-# For now we use a trivial wrapper on the grpc context object
 
-
-class CallContextBuilder(ABC):
-    """A class for building ServerCallContexts using the Starlette Request."""
+class GrpcServerCallContextBuilder(ABC):
+    """Interface for building ServerCallContext from gRPC context."""
 
     @abstractmethod
     def build(self, context: grpc.aio.ServicerContext) -> ServerCallContext:
-        """Builds a ServerCallContext from a gRPC Request."""
+        """Builds a ServerCallContext from a gRPC ServicerContext."""
+
+
+class DefaultGrpcServerCallContextBuilder(GrpcServerCallContextBuilder):
+    """Default implementation of GrpcServerCallContextBuilder."""
+
+    def build(self, context: grpc.aio.ServicerContext) -> ServerCallContext:
+        """Builds a ServerCallContext from a gRPC ServicerContext."""
+        state = {'grpc_context': context}
+        return ServerCallContext(
+            user=self.build_user(context),
+            state=state,
+            requested_extensions=get_requested_extensions(
+                _get_metadata_value(context, HTTP_EXTENSION_HEADER)
+            ),
+        )
+
+    def build_user(self, context: grpc.aio.ServicerContext) -> User:
+        """Builds a User from a gRPC ServicerContext."""
+        return UnauthenticatedUser()
 
 
 def _get_metadata_value(
@@ -67,22 +84,6 @@ def _get_metadata_value(
     ]
 
 
-class DefaultCallContextBuilder(CallContextBuilder):
-    """A default implementation of CallContextBuilder."""
-
-    def build(self, context: grpc.aio.ServicerContext) -> ServerCallContext:
-        """Builds the ServerCallContext."""
-        user = UnauthenticatedUser()
-        state = {'grpc_context': context}
-        return ServerCallContext(
-            user=user,
-            state=state,
-            requested_extensions=get_requested_extensions(
-                _get_metadata_value(context, HTTP_EXTENSION_HEADER)
-            ),
-        )
-
-
 _ERROR_CODE_MAP = {
     types.InvalidRequestError: grpc.StatusCode.INVALID_ARGUMENT,
     types.MethodNotFoundError: grpc.StatusCode.NOT_FOUND,
@@ -110,7 +111,7 @@ def __init__(
         self,
         agent_card: AgentCard,
         request_handler: RequestHandler,
-        context_builder: CallContextBuilder | None = None,
+        context_builder: GrpcServerCallContextBuilder | None = None,
         card_modifier: Callable[[AgentCard], Awaitable[AgentCard] | AgentCard]
         | None = None,
     ):
@@ -120,14 +121,17 @@ def __init__(
             agent_card: The AgentCard describing the agent's capabilities.
             request_handler: The underlying `RequestHandler` instance to
                              delegate requests to.
-            context_builder: The CallContextBuilder object. If none the
-                             DefaultCallContextBuilder is used.
+            context_builder: The GrpcContextBuilder used to construct the
+              ServerCallContext passed to the request_handler. If None the
+              DefaultGrpcContextBuilder is used.
             card_modifier: An optional callback to dynamically modify the public
               agent card before it is served.
         """
         self.agent_card = agent_card
         self.request_handler = request_handler
-        self.context_builder = context_builder or DefaultCallContextBuilder()
+        self._context_builder = (
+            context_builder or DefaultGrpcServerCallContextBuilder()
+        )
         self.card_modifier = card_modifier
 
     async def _handle_unary(
@@ -451,6 +455,6 @@ def _build_call_context(
         context: grpc.aio.ServicerContext,
         request: message.Message,
     ) -> ServerCallContext:
-        server_context = self.context_builder.build(context)
+        server_context = self._context_builder.build(context)
         server_context.tenant = getattr(request, 'tenant', '')
         return server_context

src/a2a/server/routes/__init__.py

@@ -1,17 +1,17 @@
 """A2A Routes."""
 
 from a2a.server.routes.agent_card_routes import create_agent_card_routes
-from a2a.server.routes.jsonrpc_dispatcher import (
-    CallContextBuilder,
-    DefaultCallContextBuilder,
+from a2a.server.routes.common import (
+    DefaultServerCallContextBuilder,
+    ServerCallContextBuilder,
 )
 from a2a.server.routes.jsonrpc_routes import create_jsonrpc_routes
 from a2a.server.routes.rest_routes import create_rest_routes
 
 
 __all__ = [
-    'CallContextBuilder',
-    'DefaultCallContextBuilder',
+    'DefaultServerCallContextBuilder',
+    'ServerCallContextBuilder',
     'create_agent_card_routes',
     'create_jsonrpc_routes',
     'create_rest_routes',

src/a2a/server/routes/common.py

@@ -0,0 +1,85 @@
+from abc import ABC, abstractmethod
+from typing import TYPE_CHECKING, Any
+
+
+if TYPE_CHECKING:
+    from starlette.authentication import BaseUser
+    from starlette.requests import Request
+else:
+    try:
+        from starlette.authentication import BaseUser
+        from starlette.requests import Request
+    except ImportError:
+        Request = Any
+        BaseUser = Any
+
+from a2a.auth.user import UnauthenticatedUser, User
+from a2a.extensions.common import (
+    HTTP_EXTENSION_HEADER,
+    get_requested_extensions,
+)
+from a2a.server.context import ServerCallContext
+
+
+class StarletteUser(User):
+    """Adapts a Starlette BaseUser to the A2A User interface."""
+
+    def __init__(self, user: BaseUser):
+        self._user = user
+
+    @property
+    def is_authenticated(self) -> bool:
+        """Returns whether the current user is authenticated."""
+        return self._user.is_authenticated
+
+    @property
+    def user_name(self) -> str:
+        """Returns the user name of the current user."""
+        return self._user.display_name
+
+
+class ServerCallContextBuilder(ABC):
+    """A class for building ServerCallContexts using the Starlette Request."""
+
+    @abstractmethod
+    def build(self, request: Request) -> ServerCallContext:
+        """Builds a ServerCallContext from a Starlette Request."""
+
+
+class DefaultServerCallContextBuilder(ServerCallContextBuilder):
+    """A default implementation of ServerCallContextBuilder."""
+
+    def build(self, request: Request) -> ServerCallContext:
+        """Builds a ServerCallContext from a Starlette Request.
+
+        Args:
+            request: The incoming Starlette Request object.
+
+        Returns:
+            A ServerCallContext instance populated with user and state
+            information from the request.
+        """
+        state = {}
+        if 'auth' in request.scope:
+            state['auth'] = request.auth
+        state['headers'] = dict(request.headers)
+        return ServerCallContext(
+            user=self.build_user(request),
+            state=state,
+            requested_extensions=get_requested_extensions(
+                request.headers.getlist(HTTP_EXTENSION_HEADER)
+            ),
+        )
+
+    def build_user(self, request: Request) -> User:
+        """Builds a User from a Starlette Request.
+
+        Args:
+            request: The incoming Starlette Request object.
+
+        Returns:
+            A User instance populated with user information from the request.
+        """
+        if 'user' in request.scope:
+            return StarletteUser(request.user)
+        return UnauthenticatedUser()