feat(a2a): expose a2a_task_store and a2a_push_config_store in get_fast_api_app

## Problem

`get_fast_api_app()` unconditionally creates an `InMemoryTaskStore` and
`InMemoryPushNotificationConfigStore`, making it impossible for callers to
inject persistent or shared stores without patching ADK internals.

This is especially painful in production deployments where:
- Multiple replicas need a shared task store to route A2A callbacks correctly
- Operators want task state to survive server restarts (e.g. SQLite/Postgres)

## Solution

Adds two new optional keyword arguments to `get_fast_api_app()`:

- `a2a_task_store: Optional[Any] = None`
- `a2a_push_config_store: Optional[Any] = None`

When `None` (the default), the existing `InMemory*` defaults are used —
fully backward-compatible. When provided, the caller-supplied instances are
forwarded directly to `DefaultRequestHandler`.

This mirrors the pattern introduced for the lower-level `to_a2a()` helper
in PR #3839.

## Usage

    from a2a.server.tasks import DatabaseTaskStore
    from sqlalchemy.ext.asyncio import create_async_engine

    engine = create_async_engine("postgresql+asyncpg://user:pw@host/db")
    app = get_fast_api_app(
        agents_dir="agents/",
        web=True,
        a2a=True,
        a2a_task_store=DatabaseTaskStore(engine),
    )

## Tests

Two new test cases added to tests/unittests/cli/test_fast_api.py:
- test_a2a_uses_in_memory_task_store_by_default
- test_a2a_custom_task_store_bypasses_in_memory_default

Author: STHITAPRAJNAS

src/google/adk/cli/fast_api.py

@@ -85,6 +85,8 @@ def get_fast_api_app(
     allow_origins: Optional[list[str]] = None,
     web: bool,
     a2a: bool = False,
+    a2a_task_store: Optional[Any] = None,
+    a2a_push_config_store: Optional[Any] = None,
     host: str = "127.0.0.1",
     port: int = 8000,
     url_prefix: Optional[str] = None,
@@ -128,6 +130,23 @@ def get_fast_api_app(
     allow_origins: List of allowed origins for CORS.
     web: Whether to enable the web UI and serve its assets.
     a2a: Whether to enable Agent-to-Agent (A2A) protocol support.
+    a2a_task_store: Optional A2A TaskStore instance. Defaults to
+      InMemoryTaskStore when a2a=True. Pass a DatabaseTaskStore (from the
+      a2a-sdk) for persistence across server restarts and horizontal replicas.
+      Example::
+
+          from a2a.server.tasks import DatabaseTaskStore
+          from sqlalchemy.ext.asyncio import create_async_engine
+
+          engine = create_async_engine("postgresql+asyncpg://user:pw@host/db")
+          app = get_fast_api_app(
+              agents_dir="agents/", web=True, a2a=True,
+              a2a_task_store=DatabaseTaskStore(engine),
+          )
+
+    a2a_push_config_store: Optional A2A PushNotificationConfigStore instance.
+      Defaults to InMemoryPushNotificationConfigStore when a2a=True. Pass a
+      DatabasePushNotificationConfigStore for persistence across restarts.
     host: Host address for the server (defaults to 127.0.0.1).
     port: Port number for the server (defaults to 8000).
     url_prefix: Optional prefix for all URL routes.
@@ -598,7 +617,8 @@ async def get_agent_builder(
     base_path = Path.cwd() / agents_dir
     # the root agents directory should be an existing folder
     if base_path.exists() and base_path.is_dir():
-      a2a_task_store = InMemoryTaskStore()
+      if a2a_task_store is None:
+        a2a_task_store = InMemoryTaskStore()
 
       def create_a2a_runner_loader(captured_app_name: str):
         """Factory function to create A2A runner with proper closure."""
@@ -626,7 +646,11 @@ async def _get_a2a_runner_async() -> Runner:
               runner=create_a2a_runner_loader(app_name),
           )
 
-          push_config_store = InMemoryPushNotificationConfigStore()
+          push_config_store = (
+              a2a_push_config_store
+              if a2a_push_config_store is not None
+              else InMemoryPushNotificationConfigStore()
+          )
 
           request_handler = DefaultRequestHandler(
               agent_executor=agent_executor,

tests/unittests/cli/test_fast_api.py

@@ -1866,6 +1866,138 @@ def test_a2a_disabled_by_default(test_app):
   logger.info("A2A disabled by default test passed")
 
 
+def test_a2a_uses_in_memory_task_store_by_default(
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+    temp_agents_dir_with_a2a,
+    monkeypatch,
+):
+  """Test that InMemoryTaskStore is created when no task_store is provided."""
+  with (
+      patch("signal.signal", return_value=None),
+      patch(
+          "google.adk.cli.fast_api.create_session_service_from_options",
+          return_value=mock_session_service,
+      ),
+      patch(
+          "google.adk.cli.fast_api.create_artifact_service_from_options",
+          return_value=mock_artifact_service,
+      ),
+      patch(
+          "google.adk.cli.fast_api.create_memory_service_from_options",
+          return_value=mock_memory_service,
+      ),
+      patch(
+          "google.adk.cli.fast_api.AgentLoader",
+          return_value=mock_agent_loader,
+      ),
+      patch(
+          "google.adk.cli.fast_api.LocalEvalSetsManager",
+          return_value=mock_eval_sets_manager,
+      ),
+      patch(
+          "google.adk.cli.fast_api.LocalEvalSetResultsManager",
+          return_value=mock_eval_set_results_manager,
+      ),
+      patch("a2a.server.tasks.InMemoryTaskStore") as mock_task_store_class,
+      patch("a2a.server.tasks.InMemoryPushNotificationConfigStore"),
+      patch("google.adk.a2a.executor.a2a_agent_executor.A2aAgentExecutor"),
+      patch("a2a.server.request_handlers.DefaultRequestHandler"),
+      patch("a2a.server.apps.A2AStarletteApplication") as mock_a2a_app,
+  ):
+    mock_a2a_app.return_value.routes.return_value = []
+    monkeypatch.chdir(temp_agents_dir_with_a2a)
+
+    _ = get_fast_api_app(
+        agents_dir=".",
+        web=True,
+        session_service_uri="",
+        artifact_service_uri="",
+        memory_service_uri="",
+        allow_origins=["*"],
+        a2a=True,
+        host="127.0.0.1",
+        port=8000,
+    )
+
+    mock_task_store_class.assert_called_once()
+
+
+def test_a2a_custom_task_store_bypasses_in_memory_default(
+    mock_session_service,
+    mock_artifact_service,
+    mock_memory_service,
+    mock_agent_loader,
+    mock_eval_sets_manager,
+    mock_eval_set_results_manager,
+    temp_agents_dir_with_a2a,
+    monkeypatch,
+):
+  """Test that a custom task_store is forwarded and InMemoryTaskStore is not created."""
+  custom_task_store = MagicMock()
+
+  with (
+      patch("signal.signal", return_value=None),
+      patch(
+          "google.adk.cli.fast_api.create_session_service_from_options",
+          return_value=mock_session_service,
+      ),
+      patch(
+          "google.adk.cli.fast_api.create_artifact_service_from_options",
+          return_value=mock_artifact_service,
+      ),
+      patch(
+          "google.adk.cli.fast_api.create_memory_service_from_options",
+          return_value=mock_memory_service,
+      ),
+      patch(
+          "google.adk.cli.fast_api.AgentLoader",
+          return_value=mock_agent_loader,
+      ),
+      patch(
+          "google.adk.cli.fast_api.LocalEvalSetsManager",
+          return_value=mock_eval_sets_manager,
+      ),
+      patch(
+          "google.adk.cli.fast_api.LocalEvalSetResultsManager",
+          return_value=mock_eval_set_results_manager,
+      ),
+      patch("a2a.server.tasks.InMemoryTaskStore") as mock_task_store_class,
+      patch("a2a.server.tasks.InMemoryPushNotificationConfigStore"),
+      patch("google.adk.a2a.executor.a2a_agent_executor.A2aAgentExecutor"),
+      patch(
+          "a2a.server.request_handlers.DefaultRequestHandler"
+      ) as mock_handler,
+      patch("a2a.server.apps.A2AStarletteApplication") as mock_a2a_app,
+  ):
+    mock_a2a_app.return_value.routes.return_value = []
+    monkeypatch.chdir(temp_agents_dir_with_a2a)
+
+    _ = get_fast_api_app(
+        agents_dir=".",
+        web=True,
+        session_service_uri="",
+        artifact_service_uri="",
+        memory_service_uri="",
+        allow_origins=["*"],
+        a2a=True,
+        a2a_task_store=custom_task_store,
+        host="127.0.0.1",
+        port=8000,
+    )
+
+    # InMemoryTaskStore must NOT be instantiated when a custom store is supplied
+    mock_task_store_class.assert_not_called()
+
+    # The custom store must be passed through to DefaultRequestHandler
+    call_kwargs = mock_handler.call_args.kwargs
+    assert call_kwargs["task_store"] is custom_task_store
+
+
 def test_patch_memory(test_app, create_test_session, mock_memory_service):
   """Test adding a session to memory."""
   info = create_test_session