Refactor optional deps imports in apps/rest/*

This ensures the importing of the upper level packages a2a.server.apps
succeed when optional http-server dependencies are not installed.

Author: darkhaniop

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

@@ -1,8 +1,25 @@
 import logging
 
-from typing import Any
+from typing import TYPE_CHECKING, Any
+
+
+if TYPE_CHECKING:
+    from fastapi import APIRouter, FastAPI, Request, Response
+
+    _package_fastapi_installed = True
+else:
+    try:
+        from fastapi import APIRouter, FastAPI, Request, Response
+
+        _package_fastapi_installed = True
+    except ImportError:
+        APIRouter = Any
+        FastAPI = Any
+        Request = Any
+        Response = Any
+
+        _package_fastapi_installed = False
 
-from fastapi import APIRouter, FastAPI, Request, Response
 
 from a2a.server.apps.jsonrpc.jsonrpc_app import CallContextBuilder
 from a2a.server.apps.rest.rest_adapter import RESTAdapter
@@ -40,6 +57,12 @@ def __init__(
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
         """
+        if not _package_fastapi_installed:
+            raise ImportError(
+                'The `fastapi` package is required to use the'
+                ' `A2ARESTFastAPIApplication`. It can be added as a part of'
+                ' `a2a-sdk` optional dependencies, `a2a-sdk[http-server]`.'
+            )
         self._adapter = RESTAdapter(
             agent_card=agent_card,
             http_handler=http_handler,

src/a2a/server/apps/rest/rest_adapter.py

@@ -2,12 +2,35 @@
 import logging
 
 from collections.abc import AsyncIterable, AsyncIterator, Awaitable, Callable
-from typing import Any
+from typing import TYPE_CHECKING, Any
 
 from sse_starlette.sse import EventSourceResponse
 from starlette.requests import Request
 from starlette.responses import JSONResponse, Response
 
+
+if TYPE_CHECKING:
+    from sse_starlette.sse import EventSourceResponse
+    from starlette.requests import Request
+    from starlette.responses import JSONResponse, Response
+
+    _package_starlette_installed = True
+
+else:
+    try:
+        from sse_starlette.sse import EventSourceResponse
+        from starlette.requests import Request
+        from starlette.responses import JSONResponse, Response
+
+        _package_starlette_installed = True
+    except ImportError:
+        EventSourceResponse = Any
+        Request = Any
+        JSONResponse = Any
+        Response = Any
+
+        _package_starlette_installed = False
+
 from a2a.server.apps.jsonrpc import (
     CallContextBuilder,
     DefaultCallContextBuilder,
@@ -49,6 +72,12 @@ def __init__(
               ServerCallContext passed to the http_handler. If None, no
               ServerCallContext is passed.
         """
+        if not _package_starlette_installed:
+            raise ImportError(
+                'Packages `starlette` and `sse-starlette` are required to use'
+                ' the `RESTAdapter`. They can be added as a part of `a2a-sdk`'
+                ' optional dependencies, `a2a-sdk[http-server]`.'
+            )
         self.agent_card = agent_card
         self.handler = RESTHandler(
             agent_card=agent_card, request_handler=http_handler

tests/server/apps/rest/test_rest_fastapi_app.py

@@ -1,5 +1,7 @@
 import logging
 
+from typing import Any
+
 from unittest.mock import MagicMock
 
 import pytest
@@ -9,6 +11,9 @@
 from httpx import ASGITransport, AsyncClient
 
 from a2a.grpc import a2a_pb2
+from a2a.server.apps.rest import fastapi_app
+from a2a.server.apps.rest import rest_adapter
+from a2a.server.apps.rest.rest_adapter import RESTAdapter
 from a2a.server.apps.rest.fastapi_app import A2ARESTFastAPIApplication
 from a2a.server.request_handlers.request_handler import RequestHandler
 from a2a.types import (
@@ -57,6 +62,84 @@ async def client(app: FastAPI) -> AsyncClient:
     )
 
 
+@pytest.fixture
+def mark_pkg_starlette_not_installed():
+    pkg_starlette_installed_flag = rest_adapter._package_starlette_installed
+    rest_adapter._package_starlette_installed = False
+    yield
+    rest_adapter._package_starlette_installed = pkg_starlette_installed_flag
+
+
+@pytest.fixture
+def mark_pkg_fastapi_not_installed():
+    pkg_fastapi_installed_flag = fastapi_app._package_fastapi_installed
+    fastapi_app._package_fastapi_installed = False
+    yield
+    fastapi_app._package_fastapi_installed = pkg_fastapi_installed_flag
+
+
+@pytest.mark.anyio
+async def test_create_rest_adapter_with_present_deps_succeeds(
+    agent_card: AgentCard, request_handler: RequestHandler
+):
+    try:
+        _app = RESTAdapter(agent_card, request_handler)
+    except ImportError:
+        pytest.fail(
+            'With packages starlette and see-starlette present, creating an'
+            ' RESTAdapter instance should not raise ImportError'
+        )
+
+
+@pytest.mark.anyio
+async def test_create_rest_adapter_with_missing_deps_raises_importerror(
+    agent_card: AgentCard,
+    request_handler: RequestHandler,
+    mark_pkg_starlette_not_installed: Any,
+):
+    with pytest.raises(
+        ImportError,
+        match=(
+            'Packages `starlette` and `sse-starlette` are required to use'
+            ' the `RESTAdapter`.'
+        ),
+    ):
+        _app = RESTAdapter(agent_card, request_handler)
+
+
+@pytest.mark.anyio
+async def test_create_a2a_rest_fastapi_app_with_present_deps_succeeds(
+    agent_card: AgentCard, request_handler: RequestHandler
+):
+    try:
+        _app = A2ARESTFastAPIApplication(agent_card, request_handler).build(
+            agent_card_url='/well-known/agent.json', rpc_url=''
+        )
+    except ImportError:
+        pytest.fail(
+            'With the fastapi package present, creating a'
+            ' A2ARESTFastAPIApplication instance should not raise ImportError'
+        )
+
+
+@pytest.mark.anyio
+async def test_create_a2a_rest_fastapi_app_with_missing_deps_raises_importerror(
+    agent_card: AgentCard,
+    request_handler: RequestHandler,
+    mark_pkg_fastapi_not_installed: Any,
+):
+    with pytest.raises(
+        ImportError,
+        match=(
+            'The `fastapi` package is required to use the'
+            ' `A2ARESTFastAPIApplication`'
+        ),
+    ):
+        _app = A2ARESTFastAPIApplication(agent_card, request_handler).build(
+            agent_card_url='/well-known/agent.json', rpc_url=''
+        )
+
+
 @pytest.mark.anyio
 async def test_send_message_success_message(
     client: AsyncClient, request_handler: MagicMock