diff --git a/src/a2a/server/apps/jsonrpc/fastapi_app.py b/src/a2a/server/apps/jsonrpc/fastapi_app.py index 9e1e396d..ac68c394 100644 --- a/src/a2a/server/apps/jsonrpc/fastapi_app.py +++ b/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 from a2a.utils.constants import ( AGENT_CARD_WELL_KNOWN_PATH, DEFAULT_RPC_URL, @@ -28,32 +25,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, - ) -> 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, @@ -69,22 +40,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, diff --git a/src/a2a/server/apps/jsonrpc/jsonrpc_app.py b/src/a2a/server/apps/jsonrpc/jsonrpc_app.py index 5fa1812e..5617249d 100644 --- a/src/a2a/server/apps/jsonrpc/jsonrpc_app.py +++ b/src/a2a/server/apps/jsonrpc/jsonrpc_app.py @@ -45,6 +45,7 @@ from a2a.utils.constants import ( AGENT_CARD_WELL_KNOWN_PATH, DEFAULT_RPC_URL, + EXTENDED_AGENT_CARD_PATH, ) from a2a.utils.errors import MethodNotImplementedError @@ -433,18 +434,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 = AGENT_CARD_WELL_KNOWN_PATH, rpc_url: str = DEFAULT_RPC_URL, + extended_agent_card_url: str = EXTENDED_AGENT_CARD_PATH, **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: diff --git a/src/a2a/server/apps/jsonrpc/starlette_app.py b/src/a2a/server/apps/jsonrpc/starlette_app.py index 70758739..0b0c5586 100644 --- a/src/a2a/server/apps/jsonrpc/starlette_app.py +++ b/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 from a2a.utils.constants import ( AGENT_CARD_WELL_KNOWN_PATH, DEFAULT_RPC_URL, @@ -29,32 +26,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, - ) -> 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 = AGENT_CARD_WELL_KNOWN_PATH, @@ -71,31 +42,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,