feat: support serving a to_a2a agent under a path prefix

to_a2a always mounted the JSON-RPC and agent-card routes at the root and
advertised an RPC URL ending in "/", so several agents could not be
hosted behind one gateway without colliding. Add an optional keyword-only
rpc_path param that normalizes slashes, threads a prefix into the
advertised RPC URL, and mounts both routes under that prefix. It defaults
to "" so existing behavior (root mount) is unchanged.

Close #4448

Co-authored-by: George Weale <gweale@google.com>
PiperOrigin-RevId: 953444584

Author: GWeale

src/google/adk/a2a/utils/agent_to_a2a.py

@@ -81,6 +81,7 @@ def to_a2a(
     host: str = "localhost",
     port: int = 8000,
     protocol: str = "http",
+    rpc_path: str = "",
     agent_card: AgentCard | str | None = None,
     push_config_store: PushNotificationConfigStore | None = None,
     task_store: TaskStore | None = None,
@@ -95,6 +96,12 @@ def to_a2a(
       host: The host for the A2A RPC URL (default: "localhost")
       port: The port for the A2A RPC URL (default: 8000)
       protocol: The protocol for the A2A RPC URL (default: "http")
+      rpc_path: Optional path prefix to serve the agent under, e.g.
+        "analysis-agent". Leading/trailing slashes are ignored. When set, both
+        the JSON-RPC route and the well-known agent-card route are mounted under
+        this prefix instead of at the root. Defaults to "" (mount at root). A
+        caller-provided agent_card's advertised url is not rewritten; a warning
+        is logged if both agent_card and a non-empty rpc_path are supplied.
       agent_card: Optional pre-built AgentCard object or path to agent card
         JSON. If not provided, will be built automatically from the agent.
       push_config_store: Optional A2A push notification config store. If not
@@ -180,9 +187,19 @@ def create_runner() -> Runner:
     push_config_store = InMemoryPushNotificationConfigStore()
 
   # Use provided agent card or build one from the agent
-  rpc_url = f"{protocol}://{host}:{port}/"
+  normalized_path = rpc_path.strip("/")
+  prefix = f"/{normalized_path}" if normalized_path else ""
+  rpc_url = f"{protocol}://{host}:{port}{prefix}/"
   provided_agent_card = _load_agent_card(agent_card)
 
+  if provided_agent_card is not None and normalized_path:
+    adk_logger.warning(
+        "Both agent_card and rpc_path were provided; routes are mounted under"
+        " %r but the provided agent_card's advertised url is left unchanged, so"
+        " clients reading the card may not reach the served location.",
+        prefix,
+    )
+
   card_builder = AgentCardBuilder(
       agent=agent,
       rpc_url=rpc_url,
@@ -202,6 +219,7 @@ async def setup_a2a(app: Starlette):
         agent_executor=agent_executor,
         task_store=task_store,
         push_config_store=push_config_store,
+        prefix=prefix,
     )
 
   # Compose a lifespan that runs A2A setup and the user's lifespan

tests/unittests/a2a/utils/test_agent_to_a2a.py

@@ -26,6 +26,7 @@
 (driving the app lifespan and asserting routes are attached).
 """
 
+import logging
 from unittest.mock import ANY
 from unittest.mock import AsyncMock
 from unittest.mock import Mock
@@ -225,6 +226,100 @@ def test_to_a2a_custom_host_port(
         agent=self.mock_agent, rpc_url="http://example.com:9000/"
     )
 
+  @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
+  @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
+  @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
+  @patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
+  def test_to_a2a_with_rpc_path(
+      self,
+      mock_starlette_class,
+      mock_card_builder_class,
+      mock_task_store_class,
+      mock_agent_executor_class,
+  ):
+    """rpc_path is threaded into the advertised RPC URL."""
+    mock_starlette_class.return_value = Mock(spec=Starlette)
+    mock_agent_executor_class.return_value = Mock(spec=A2aAgentExecutor)
+    mock_card_builder_class.return_value = Mock(spec=AgentCardBuilder)
+
+    to_a2a(self.mock_agent, rpc_path="/analysis-agent")
+
+    mock_card_builder_class.assert_called_once_with(
+        agent=self.mock_agent, rpc_url="http://localhost:8000/analysis-agent/"
+    )
+
+  @pytest.mark.parametrize(
+      "rpc_path", ["analysis-agent", "/analysis-agent", "/analysis-agent/"]
+  )
+  @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
+  @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
+  @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
+  @patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
+  def test_to_a2a_rpc_path_normalizes_slashes(
+      self,
+      mock_starlette_class,
+      mock_card_builder_class,
+      mock_task_store_class,
+      mock_agent_executor_class,
+      rpc_path,
+  ):
+    """Leading/trailing slashes on rpc_path are normalized identically."""
+    mock_starlette_class.return_value = Mock(spec=Starlette)
+    mock_agent_executor_class.return_value = Mock(spec=A2aAgentExecutor)
+    mock_card_builder_class.return_value = Mock(spec=AgentCardBuilder)
+
+    to_a2a(self.mock_agent, rpc_path=rpc_path)
+
+    mock_card_builder_class.assert_called_once_with(
+        agent=self.mock_agent, rpc_url="http://localhost:8000/analysis-agent/"
+    )
+
+  @pytest.mark.parametrize("rpc_path", ["/", "//", "///"])
+  @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
+  @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
+  @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
+  @patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
+  def test_to_a2a_all_slash_rpc_path_collapses_to_root(
+      self,
+      mock_starlette_class,
+      mock_card_builder_class,
+      mock_task_store_class,
+      mock_agent_executor_class,
+      rpc_path,
+  ):
+    """An all-slash rpc_path collapses to a root mount."""
+    mock_starlette_class.return_value = Mock(spec=Starlette)
+    mock_agent_executor_class.return_value = Mock(spec=A2aAgentExecutor)
+    mock_card_builder_class.return_value = Mock(spec=AgentCardBuilder)
+
+    to_a2a(self.mock_agent, rpc_path=rpc_path)
+
+    mock_card_builder_class.assert_called_once_with(
+        agent=self.mock_agent, rpc_url="http://localhost:8000/"
+    )
+
+  @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
+  @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
+  @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
+  @patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
+  def test_to_a2a_multi_segment_rpc_path(
+      self,
+      mock_starlette_class,
+      mock_card_builder_class,
+      mock_task_store_class,
+      mock_agent_executor_class,
+  ):
+    """A multi-segment rpc_path is preserved in the RPC URL."""
+    mock_starlette_class.return_value = Mock(spec=Starlette)
+    mock_agent_executor_class.return_value = Mock(spec=A2aAgentExecutor)
+    mock_card_builder_class.return_value = Mock(spec=AgentCardBuilder)
+
+    to_a2a(self.mock_agent, rpc_path="/team/agent")
+
+    mock_card_builder_class.assert_called_once_with(
+        agent=self.mock_agent, rpc_url="http://localhost:8000/team/agent/"
+    )
+
   @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
   @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
   @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
@@ -518,6 +613,26 @@ async def test_setup_a2a_builds_card_and_attaches_routes(
     mock_card_builder.build.assert_called_once()
     _assert_a2a_routes_attached(app)
 
+  async def test_to_a2a_rpc_path_mounts_prefixed_routes(self):
+    """rpc_path mounts the JSON-RPC and agent-card routes under the prefix."""
+    agent = LlmAgent(
+        name="prefixed_agent", description="d", model="gemini-2.0-flash"
+    )
+    app = to_a2a(agent, port=8001, rpc_path="/analysis-agent")
+
+    async with app.router.lifespan_context(app):
+      paths = _route_paths(app)
+
+    assert (
+        "/analysis-agent" in paths
+    ), f"missing prefixed RPC route; got {paths}"
+    assert any(
+        p and p.startswith("/analysis-agent/.well-known") for p in paths
+    ), f"missing prefixed agent-card route; got {paths}"
+    assert (
+        "/" not in paths
+    ), f"root RPC route should not be mounted; got {paths}"
+
   @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
   @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
   @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
@@ -560,6 +675,34 @@ async def test_to_a2a_with_custom_agent_card_object(
     mock_card_builder.build.assert_not_called()
     _assert_a2a_routes_attached(app)
 
+  @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
+  @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
+  @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")
+  @patch("google.adk.a2a.utils.agent_to_a2a.Starlette")
+  def test_to_a2a_warns_when_agent_card_and_rpc_path_both_set(
+      self,
+      mock_starlette_class,
+      mock_card_builder_class,
+      mock_task_store_class,
+      mock_agent_executor_class,
+      caplog,
+  ):
+    """A provided agent_card plus a non-empty rpc_path logs a mismatch warning."""
+    mock_starlette_class.return_value = Mock(spec=Starlette)
+    mock_agent_executor_class.return_value = Mock(spec=A2aAgentExecutor)
+    mock_card_builder_class.return_value = Mock(spec=AgentCardBuilder)
+
+    with caplog.at_level(logging.WARNING, logger="google_adk"):
+      to_a2a(
+          self.mock_agent,
+          rpc_path="/analysis-agent",
+          agent_card=_make_minimal_agent_card(),
+      )
+
+    assert any(
+        "agent_card and rpc_path" in r.message for r in caplog.records
+    ), f"expected mismatch warning; got {[r.message for r in caplog.records]}"
+
   @patch("google.adk.a2a.utils.agent_to_a2a.A2aAgentExecutor")
   @patch("google.adk.a2a.utils.agent_to_a2a.InMemoryTaskStore")
   @patch("google.adk.a2a.utils.agent_to_a2a.AgentCardBuilder")