refactor: Reduce Duplication in server/apps/jsonrpc

Author: holtskinner

src/a2a/server/apps/jsonrpc/fastapi_app.py

@@ -2,14 +2,11 @@
 
 from typing import Any
 
-from fastapi import FastAPI, Request, Response
+from fastapi import FastAPI
 
 from a2a.server.apps.jsonrpc.jsonrpc_app import (
-    CallContextBuilder,
     JSONRPCApplication,
 )
-from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
-from a2a.types import AgentCard
 
 
 logger = logging.getLogger(__name__)
@@ -23,32 +20,6 @@ class A2AFastAPIApplication(JSONRPCApplication):
     (SSE).
     """
 
-    def __init__(
-        self,
-        agent_card: AgentCard,
-        http_handler: RequestHandler,
-        extended_agent_card: AgentCard | None = None,
-        context_builder: CallContextBuilder | None = None,
-    ):
-        """Initializes the A2AStarletteApplication.
-
-        Args:
-            agent_card: The AgentCard describing the agent's capabilities.
-            http_handler: The handler instance responsible for processing A2A
-              requests via http.
-            extended_agent_card: An optional, distinct AgentCard to be served
-              at the authenticated extended card endpoint.
-            context_builder: The CallContextBuilder used to construct the
-              ServerCallContext passed to the http_handler. If None, no
-              ServerCallContext is passed.
-        """
-        super().__init__(
-            agent_card=agent_card,
-            http_handler=http_handler,
-            extended_agent_card=extended_agent_card,
-            context_builder=context_builder,
-        )
-
     def add_routes_to_app(
         self,
         app: FastAPI,
@@ -64,22 +35,12 @@ def add_routes_to_app(
             rpc_url: The URL for the A2A JSON-RPC endpoint.
             extended_agent_card_url: The URL for the authenticated extended agent card endpoint.
         """
+        route_definitions = self._get_route_definitions(
+            agent_card_url, rpc_url, extended_agent_card_url
+        )
 
-        @app.post(rpc_url)
-        async def handle_a2a_request(request: Request) -> Response:
-            return await self._handle_requests(request)
-
-        @app.get(agent_card_url)
-        async def get_agent_card(request: Request) -> Response:
-            return await self._handle_get_agent_card(request)
-
-        if self.agent_card.supportsAuthenticatedExtendedCard:
-
-            @app.get(extended_agent_card_url)
-            async def get_extended_agent_card(request: Request) -> Response:
-                return await self._handle_get_authenticated_extended_agent_card(
-                    request
-                )
+        for route_def in route_definitions:
+            app.add_api_route(**route_def)
 
     def build(
         self,

src/a2a/server/apps/jsonrpc/jsonrpc_app.py

@@ -409,18 +409,54 @@ async def _handle_get_authenticated_extended_agent_card(
             status_code=404,
         )
 
+    def _get_route_definitions(
+        self,
+        agent_card_url: str,
+        rpc_url: str,
+        extended_agent_card_url: str,
+    ) -> list[dict[str, Any]]:
+        """Generates a list of route definitions for the application."""
+        routes = [
+            {
+                'path': rpc_url,
+                'endpoint': self._handle_requests,
+                'methods': ['POST'],
+                'name': 'a2a_handler',
+            },
+            {
+                'path': agent_card_url,
+                'endpoint': self._handle_get_agent_card,
+                'methods': ['GET'],
+                'name': 'agent_card',
+            },
+        ]
+
+        if self.agent_card.supportsAuthenticatedExtendedCard:
+            routes.append(
+                {
+                    'path': extended_agent_card_url,
+                    'endpoint': self._handle_get_authenticated_extended_agent_card,
+                    'methods': ['GET'],
+                    'name': 'authenticated_extended_agent_card',
+                }
+            )
+        return routes
+
     @abstractmethod
     def build(
         self,
         agent_card_url: str = '/.well-known/agent.json',
         rpc_url: str = '/',
+        extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
         **kwargs: Any,
     ) -> FastAPI | Starlette:
         """Builds and returns the JSONRPC application instance.
 
         Args:
             agent_card_url: The URL for the agent card endpoint.
-            rpc_url: The URL for the A2A JSON-RPC endpoint
+            rpc_url: The URL for the A2A JSON-RPC endpoint.
+            extended_agent_card_url: The URL for the authenticated extended
+              agent card endpoint.
             **kwargs: Additional keyword arguments to pass to the FastAPI constructor.
 
         Returns:

src/a2a/server/apps/jsonrpc/starlette_app.py

@@ -6,11 +6,8 @@
 from starlette.routing import Route
 
 from a2a.server.apps.jsonrpc.jsonrpc_app import (
-    CallContextBuilder,
     JSONRPCApplication,
 )
-from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
-from a2a.types import AgentCard
 
 
 logger = logging.getLogger(__name__)
@@ -24,32 +21,6 @@ class A2AStarletteApplication(JSONRPCApplication):
     (SSE).
     """
 
-    def __init__(
-        self,
-        agent_card: AgentCard,
-        http_handler: RequestHandler,
-        extended_agent_card: AgentCard | None = None,
-        context_builder: CallContextBuilder | None = None,
-    ):
-        """Initializes the A2AStarletteApplication.
-
-        Args:
-            agent_card: The AgentCard describing the agent's capabilities.
-            http_handler: The handler instance responsible for processing A2A
-              requests via http.
-            extended_agent_card: An optional, distinct AgentCard to be served
-              at the authenticated extended card endpoint.
-            context_builder: The CallContextBuilder used to construct the
-              ServerCallContext passed to the http_handler. If None, no
-              ServerCallContext is passed.
-        """
-        super().__init__(
-            agent_card=agent_card,
-            http_handler=http_handler,
-            extended_agent_card=extended_agent_card,
-            context_builder=context_builder,
-        )
-
     def routes(
         self,
         agent_card_url: str = '/.well-known/agent.json',
@@ -66,31 +37,10 @@ def routes(
         Returns:
             A list of Starlette Route objects.
         """
-        app_routes = [
-            Route(
-                rpc_url,
-                self._handle_requests,
-                methods=['POST'],
-                name='a2a_handler',
-            ),
-            Route(
-                agent_card_url,
-                self._handle_get_agent_card,
-                methods=['GET'],
-                name='agent_card',
-            ),
-        ]
-
-        if self.agent_card.supportsAuthenticatedExtendedCard:
-            app_routes.append(
-                Route(
-                    extended_agent_card_url,
-                    self._handle_get_authenticated_extended_agent_card,
-                    methods=['GET'],
-                    name='authenticated_extended_agent_card',
-                )
-            )
-        return app_routes
+        route_definitions = self._get_route_definitions(
+            agent_card_url, rpc_url, extended_agent_card_url
+        )
+        return [Route(**route_def) for route_def in route_definitions]
 
     def add_routes_to_app(
         self,
@@ -107,12 +57,12 @@ def add_routes_to_app(
             rpc_url: The URL path for the A2A JSON-RPC endpoint (POST requests).
             extended_agent_card_url: The URL for the authenticated extended agent card endpoint.
         """
-        routes = self.routes(
+        app_routes = self.routes(
             agent_card_url=agent_card_url,
             rpc_url=rpc_url,
             extended_agent_card_url=extended_agent_card_url,
         )
-        app.routes.extend(routes)
+        app.routes.extend(app_routes)
 
     def build(
         self,