add-fastapi-application

martimfasantos · martimfasantos · commit 1da4b3bd35c5 · 2025-05-15T16:31:44.000+01:00
diff --git a/pyproject.toml b/pyproject.toml
@@ -5,6 +5,7 @@ description = "A2A Python SDK"
 readme = "README.md"
 requires-python = ">=3.13"
 dependencies = [
+    "fastapi>=0.115.12",
     "httpx>=0.28.1",
     "httpx-sse>=0.4.0",
     "opentelemetry-api>=1.33.0",
@@ -13,6 +14,7 @@ dependencies = [
     "sse-starlette>=2.3.3",
     "starlette>=0.46.2",
     "typing-extensions>=4.13.2",
+    "uvicorn>=0.34.2",
 ]
 
 [tool.hatch.build.targets.wheel]
diff --git a/src/a2a/server/apps/__init__.py b/src/a2a/server/apps/__init__.py
@@ -1,5 +1,12 @@
 from a2a.server.apps.http_app import HttpApp
+from a2a.server.apps.default_app import DefaultA2AApplication
 from a2a.server.apps.starlette_app import A2AStarletteApplication
+from a2a.server.apps.fastapi_app import A2AFastAPIApplication
 
 
-__all__ = ['A2AStarletteApplication', 'HttpApp']
+__all__ = [
+    'DefaultA2AApplication',
+    'A2AStarletteApplication', 
+    'A2AFastAPIApplication', 
+    'HttpApp'
+]
diff --git a/src/a2a/server/apps/default_app.py b/src/a2a/server/apps/default_app.py
@@ -0,0 +1,272 @@
+import json
+import logging
+import traceback
+from abc import ABC, abstractmethod
+
+from collections.abc import AsyncGenerator
+from typing import Any, Optional, Union
+
+from pydantic import ValidationError
+from sse_starlette.sse import EventSourceResponse
+from starlette.applications import Starlette
+from fastapi import FastAPI
+from starlette.requests import Request
+from starlette.responses import JSONResponse, Response
+
+from a2a.server.request_handlers.jsonrpc_handler import (
+    JSONRPCHandler,
+    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.utils.errors import MethodNotImplementedError
+
+logger = logging.getLogger(__name__)
+
+
+class DefaultA2AApplication(ABC):
+    """Base class for A2A applications.
+
+    Args:
+        agent_card: The AgentCard describing the agent's capabilities.
+        http_handler: The handler instance responsible for processing A2A
+            requests via http.
+    """
+
+    def __init__(self, agent_card: AgentCard, http_handler: RequestHandler):
+        """Initializes the A2AApplication.
+
+        Args:
+            agent_card: The AgentCard describing the agent's capabilities.
+            http_handler: The handler instance responsible for processing A2A
+              requests via http.
+        """
+        self.agent_card = agent_card
+        self.handler = JSONRPCHandler(
+            agent_card=agent_card, request_handler=http_handler
+        )
+
+    def _generate_error_response(
+        self, request_id: str | int | None, error: JSONRPCError | A2AError
+    ) -> JSONResponse:
+        """Creates a JSONResponse for a JSON-RPC error."""
+        error_resp = JSONRPCErrorResponse(
+            id=request_id,
+            error=error if isinstance(error, JSONRPCError) else error.root,
+        )
+
+        log_level = (
+            logging.ERROR
+            if not isinstance(error, A2AError)
+            or isinstance(error.root, InternalError)
+            else logging.WARNING
+        )
+        logger.log(
+            log_level,
+            f'Request Error (ID: {request_id}: '
+            f"Code={error_resp.error.code}, Message='{error_resp.error.message}'"
+            f'{", Data=" + str(error_resp.error.data) if hasattr(error, "data") and error_resp.error.data else ""}',
+        )
+        return JSONResponse(
+            error_resp.model_dump(mode='json', exclude_none=True),
+            status_code=200,
+        )
+
+    async def _handle_requests(self, request: Request) -> Response:
+        """Handles incoming POST requests to the main A2A endpoint.
+
+        Parses the request body as JSON, validates it against A2A request types,
+        dispatches it to the appropriate handler method, and returns the response.
+        Handles JSON parsing errors, validation errors, and other exceptions,
+        returning appropriate JSON-RPC error responses.
+        """
+        request_id = None
+        body = None
+
+        try:
+            body = await request.json()
+            a2a_request = A2ARequest.model_validate(body)
+
+            request_id = a2a_request.root.id
+            request_obj = a2a_request.root
+
+            if isinstance(
+                request_obj,
+                TaskResubscriptionRequest | SendStreamingMessageRequest,
+            ):
+                return await self._process_streaming_request(
+                    request_id, a2a_request
+                )
+
+            return await self._process_non_streaming_request(
+                request_id, a2a_request
+            )
+        except MethodNotImplementedError:
+            traceback.print_exc()
+            return self._generate_error_response(
+                request_id, A2AError(root=UnsupportedOperationError())
+            )
+        except json.decoder.JSONDecodeError as e:
+            traceback.print_exc()
+            return self._generate_error_response(
+                None, A2AError(root=JSONParseError(message=str(e)))
+            )
+        except ValidationError as e:
+            traceback.print_exc()
+            return self._generate_error_response(
+                request_id,
+                A2AError(root=InvalidRequestError(data=json.loads(e.json()))),
+            )
+        except Exception as e:
+            logger.error(f'Unhandled exception: {e}')
+            traceback.print_exc()
+            return self._generate_error_response(
+                request_id, A2AError(root=InternalError(message=str(e)))
+            )
+
+    async def _process_streaming_request(
+        self, request_id: str | int | None, a2a_request: A2ARequest
+    ) -> Response:
+        """Processes streaming requests.
+
+        Args:
+            request_id: The ID of the request.
+            a2a_request: The validated A2ARequest object.
+        """
+        request_obj = a2a_request.root
+        handler_result: Any = None
+        if isinstance(
+            request_obj,
+            SendStreamingMessageRequest,
+        ):
+            handler_result = self.handler.on_message_send_stream(request_obj)
+        elif isinstance(request_obj, TaskResubscriptionRequest):
+            handler_result = self.handler.on_resubscribe_to_task(request_obj)
+
+        return self._create_response(handler_result)
+
+    async def _process_non_streaming_request(
+        self, request_id: str | int | None, a2a_request: A2ARequest
+    ) -> Response:
+        """Processes non-streaming requests.
+
+        Args:
+            request_id: The ID of the request.
+            a2a_request: The validated A2ARequest object.
+        """
+        request_obj = a2a_request.root
+        handler_result: Any = None
+        match request_obj:
+            case SendMessageRequest():
+                handler_result = await self.handler.on_message_send(request_obj)
+            case CancelTaskRequest():
+                handler_result = await self.handler.on_cancel_task(request_obj)
+            case GetTaskRequest():
+                handler_result = await self.handler.on_get_task(request_obj)
+            case SetTaskPushNotificationConfigRequest():
+                handler_result = await self.handler.set_push_notification(
+                    request_obj
+                )
+            case GetTaskPushNotificationConfigRequest():
+                handler_result = await self.handler.get_push_notification(
+                    request_obj
+                )
+            case _:
+                logger.error(
+                    f'Unhandled validated request type: {type(request_obj)}'
+                )
+                error = UnsupportedOperationError(
+                    message=f'Request type {type(request_obj).__name__} is unknown.'
+                )
+                handler_result = JSONRPCErrorResponse(
+                    id=request_id, error=error
+                )
+
+        return self._create_response(handler_result)
+
+    def _create_response(
+        self,
+        handler_result: (
+            AsyncGenerator[SendStreamingMessageResponse, None]
+            | JSONRPCErrorResponse
+            | JSONRPCResponse
+        ),
+    ) -> Response:
+        """Creates a Starlette Response based on the result from the request handler.
+
+        Handles:
+        - AsyncGenerator for Server-Sent Events (SSE).
+        - JSONRPCErrorResponse for explicit errors returned by handlers.
+        - Pydantic RootModels (like GetTaskResponse) containing success or error
+        payloads.
+        - Unexpected types by returning an InternalError.
+
+        Args:
+            handler_result: AsyncGenerator of SendStreamingMessageResponse
+
+        Returns:
+            A Starlette JSONResponse or EventSourceResponse.
+        """
+        if isinstance(handler_result, AsyncGenerator):
+            # Result is a stream of SendStreamingMessageResponse objects
+            async def event_generator(
+                stream: AsyncGenerator[SendStreamingMessageResponse, None],
+            ) -> AsyncGenerator[dict[str, str], None]:
+                async for item in stream:
+                    yield {'data': item.root.model_dump_json(exclude_none=True)}
+
+            return EventSourceResponse(event_generator(handler_result))
+        if isinstance(handler_result, JSONRPCErrorResponse):
+            return JSONResponse(
+                handler_result.model_dump(
+                    mode='json',
+                    exclude_none=True,
+                )
+            )
+
+        return JSONResponse(
+            handler_result.root.model_dump(mode='json', exclude_none=True)
+        )
+
+    async def _handle_get_agent_card(self, request: Request) -> JSONResponse:
+        """Handles GET requests for the agent card."""
+        return JSONResponse(
+            self.agent_card.model_dump(mode='json', exclude_none=True)
+        )
+
+    @abstractmethod
+    def build(
+        self,
+        agent_card_url: str = '/.well-known/agent.json',
+        rpc_url: str = '/',
+        **kwargs: Any,
+    ) -> Union[Starlette, 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
+            **kwargs: Additional keyword arguments to pass to the FastAPI constructor.
+
+        Returns:
+            A configured FastAPI application instance.
+        """
+        pass
diff --git a/src/a2a/server/apps/fastapi_app.py b/src/a2a/server/apps/fastapi_app.py
@@ -0,0 +1,55 @@
+import logging
+from typing import Any
+
+from fastapi import FastAPI, Request
+
+from a2a.server.apps import DefaultA2AApplication
+from a2a.server.request_handlers.jsonrpc_handler import RequestHandler
+from a2a.types import AgentCard
+
+logger = logging.getLogger(__name__)
+
+
+class A2AFastAPIApplication(DefaultA2AApplication):
+    """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):
+        """Initializes the A2A FastAPI application.
+
+        Args:
+            agent_card: The AgentCard describing the agent's capabilities.
+            http_handler: The handler instance responsible for processing A2A requests via http.
+        """
+        super().__init__(agent_card, http_handler)
+
+    def build(
+        self,
+        agent_card_url: str = '/.well-known/agent.json',
+        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
+            **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)
+
+        return app
diff --git a/src/a2a/server/apps/starlette_app.py b/src/a2a/server/apps/starlette_app.py
diff --git a/uv.lock b/uv.lock
