feat!: Add FastAPI JSONRPC Application

.github/actions/spelling/allow.txt

@@ -1,6 +1,7 @@
 ACard
 AClient
 AError
+AFast
 ARequest
 ARun
 AServer

pyproject.toml

@@ -8,6 +8,7 @@ authors = [{ name = "Google LLC", email = "[email protected]" }]
 requires-python = ">=3.10"
 keywords = ["A2A", "A2A SDK", "A2A Protocol", "Agent2Agent"]
 dependencies = [
+    "fastapi>=0.115.12",
     "httpx>=0.28.1",
     "httpx-sse>=0.4.0",
     "opentelemetry-api>=1.33.0",

src/a2a/server/apps/__init__.py

@@ -1,6 +1 @@
 """HTTP application components for the A2A server."""
-
-from a2a.server.apps.starlette_app import A2AStarletteApplication
-
-
-__all__ = ['A2AStarletteApplication']

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

@@ -0,0 +1,13 @@
+"""A2A JSON-RPC Applications."""
+
+from .jsonrpc_app import CallContextBuilder, JSONRPCApplication
+from .fastapi_app import A2AFastAPIApplication
+from .starlette_app import A2AStarletteApplication
+
+
+__all__ = [
+    'A2AFastAPIApplication',
+    'A2AStarletteApplication',
+    'CallContextBuilder',
+    'JSONRPCApplication',
+]

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

@@ -0,0 +1,82 @@
+import logging
+
+from typing import Any
+
+from fastapi import FastAPI, Request
+
+from .jsonrpc_app import CallContextBuilder, JSONRPCApplication
+from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
+from a2a.types import AgentCard
+
+
+logger = logging.getLogger(__name__)
+
+
+class A2AFastAPIApplication(JSONRPCApplication):
+    """A FastAPI application implementing the A2A protocol server endpoints.
+
+    Handles incoming JSON-RPC requests, routes them to the appropriate
+    handler methods, and manages response generation including Server-Sent Events
+    (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 build(
+        self,
+        agent_card_url: str = '/.well-known/agent.json',
+        extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
+        rpc_url: str = '/',
+        **kwargs: Any,
+    ) -> FastAPI:
+        """Builds and returns the FastAPI application instance.
+
+        Args:
+            agent_card_url: The URL for the agent card 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:
+            A configured FastAPI application instance.
+        """
+        app = FastAPI(**kwargs)
+
+        @app.post(rpc_url)
+        async def handle_a2a_request(request: Request):
+            return await self._handle_requests(request)
+
+        @app.get(agent_card_url)
+        async def get_agent_card(request: Request):
+            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):
+                return await self._handle_get_authenticated_extended_agent_card(request)
+
+        return app

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

@@ -5,15 +5,15 @@
 
 from abc import ABC, abstractmethod
 from collections.abc import AsyncGenerator
-from typing import Any
+from typing import Any, Union
 
+from fastapi import FastAPI
 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 UnauthenticatedUser
 from a2a.auth.user import User as A2AUser
@@ -81,8 +81,8 @@ def build(self, request: Request) -> ServerCallContext:
         return ServerCallContext(user=user, state=state)
 
 
-class A2AStarletteApplication:
-    """A Starlette application implementing the A2A protocol server endpoints.
+class JSONRPCApplication(ABC):
+    """Base class for A2A JSONRPC applications.
 
     Handles incoming JSON-RPC requests, routes them to the appropriate
     handler methods, and manages response generation including Server-Sent Events
@@ -391,73 +391,23 @@ async def _handle_get_authenticated_extended_agent_card(
             status_code=404,
         )
 
-    def routes(
-        self,
-        agent_card_url: str = '/.well-known/agent.json',
-        extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
-        rpc_url: str = '/',
-    ) -> list[Route]:
-        """Returns the Starlette Routes for handling A2A requests.
-
-        Args:
-            agent_card_url: The URL path for the agent card endpoint.
-            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.
-
-        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
-
+    @abstractmethod
     def build(
         self,
         agent_card_url: str = '/.well-known/agent.json',
-        extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
         rpc_url: str = '/',
         **kwargs: Any,
-    ) -> Starlette:
-        """Builds and returns the Starlette application instance.
+    ) -> Union[FastAPI, Starlette]:
+        """Builds and returns the JSONRPC application instance.
 
         Args:
-            agent_card_url: The URL path for the agent card endpoint.
-            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.
-            **kwargs: Additional keyword arguments to pass to the Starlette
-              constructor.
+            agent_card_url: The URL for the agent card endpoint.
+            rpc_url: The URL for the A2A JSON-RPC endpoint
+            **kwargs: Additional keyword arguments to pass to the FastAPI constructor.
 
         Returns:
-            A configured Starlette application instance.
+            A configured JSONRPC application instance.
         """
-        app_routes = self.routes(
-            agent_card_url, extended_agent_card_url, rpc_url
+        raise NotImplementedError(
+            'Subclasses must implement the build method to create the application instance.'
         )
-        if 'routes' in kwargs:
-            kwargs['routes'].extend(app_routes)
-        else:
-            kwargs['routes'] = app_routes
-
-        return Starlette(**kwargs)

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

@@ -0,0 +1,119 @@
+import logging
+
+from typing import Any
+
+from starlette.applications import Starlette
+from starlette.routing import Route
+
+from .jsonrpc_app import CallContextBuilder, JSONRPCApplication
+from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
+from a2a.types import AgentCard
+
+
+logger = logging.getLogger(__name__)
+
+
+class A2AStarletteApplication(JSONRPCApplication):
+    """A Starlette application implementing the A2A protocol server endpoints.
+
+    Handles incoming JSON-RPC requests, routes them to the appropriate
+    handler methods, and manages response generation including Server-Sent Events
+    (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',
+        extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
+        rpc_url: str = '/',
+    ) -> list[Route]:
+        """Returns the Starlette Routes for handling A2A requests.
+
+        Args:
+            agent_card_url: The URL path for the agent card endpoint.
+            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.
+
+        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
+
+    def build(
+        self,
+        agent_card_url: str = '/.well-known/agent.json',
+        extended_agent_card_url: str = '/agent/authenticatedExtendedCard',
+        rpc_url: str = '/',
+        **kwargs: Any,
+    ) -> Starlette:
+        """Builds and returns the Starlette application instance.
+
+        Args:
+            agent_card_url: The URL path for the agent card endpoint.
+            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.
+            **kwargs: Additional keyword arguments to pass to the Starlette
+              constructor.
+
+        Returns:
+            A configured Starlette application instance.
+        """
+        app_routes = self.routes(
+            agent_card_url, extended_agent_card_url, rpc_url
+        )
+        if 'routes' in kwargs:
+            kwargs['routes'].extend(app_routes)
+        else:
+            kwargs['routes'] = app_routes
+
+        return Starlette(**kwargs)