feat(a2a): support mounts for containerized deployments (#524)

* feat(a2a): support mounts for containerized deployments

* feat(a2a): escape hatch for load balancers which strip paths

* feat(a2a): formatting

---------

Co-authored-by: jer <jerebill@amazon.com>

Author: 2 people

src/strands/multiagent/a2a/server.py

@@ -6,6 +6,7 @@
 
 import logging
 from typing import Any, Literal
+from urllib.parse import urlparse
 
 import uvicorn
 from a2a.server.apps import A2AFastAPIApplication, A2AStarletteApplication
@@ -31,6 +32,8 @@ def __init__(
         # AgentCard
         host: str = "0.0.0.0",
         port: int = 9000,
+        http_url: str | None = None,
+        serve_at_root: bool = False,
         version: str = "0.0.1",
         skills: list[AgentSkill] | None = None,
     ):
@@ -40,13 +43,34 @@ def __init__(
             agent: The Strands Agent to wrap with A2A compatibility.
             host: The hostname or IP address to bind the A2A server to. Defaults to "0.0.0.0".
             port: The port to bind the A2A server to. Defaults to 9000.
+            http_url: The public HTTP URL where this agent will be accessible. If provided,
+                this overrides the generated URL from host/port and enables automatic
+                path-based mounting for load balancer scenarios.
+                Example: "http://my-alb.amazonaws.com/agent1"
+            serve_at_root: If True, forces the server to serve at root path regardless of
+                http_url path component. Use this when your load balancer strips path prefixes.
+                Defaults to False.
             version: The version of the agent. Defaults to "0.0.1".
             skills: The list of capabilities or functions the agent can perform.
         """
         self.host = host
         self.port = port
-        self.http_url = f"http://{self.host}:{self.port}/"
         self.version = version
+
+        if http_url:
+            # Parse the provided URL to extract components for mounting
+            self.public_base_url, self.mount_path = self._parse_public_url(http_url)
+            self.http_url = http_url.rstrip("/") + "/"
+
+            # Override mount path if serve_at_root is requested
+            if serve_at_root:
+                self.mount_path = ""
+        else:
+            # Fall back to constructing the URL from host and port
+            self.public_base_url = f"http://{host}:{port}"
+            self.http_url = f"{self.public_base_url}/"
+            self.mount_path = ""
+
         self.strands_agent = agent
         self.name = self.strands_agent.name
         self.description = self.strands_agent.description
@@ -58,6 +82,25 @@ def __init__(
         self._agent_skills = skills
         logger.info("Strands' integration with A2A is experimental. Be aware of frequent breaking changes.")
 
+    def _parse_public_url(self, url: str) -> tuple[str, str]:
+        """Parse the public URL into base URL and mount path components.
+
+        Args:
+            url: The full public URL (e.g., "http://my-alb.amazonaws.com/agent1")
+
+        Returns:
+            tuple: (base_url, mount_path) where base_url is the scheme+netloc
+                  and mount_path is the path component
+
+        Example:
+            _parse_public_url("http://my-alb.amazonaws.com/agent1")
+            Returns: ("http://my-alb.amazonaws.com", "/agent1")
+        """
+        parsed = urlparse(url.rstrip("/"))
+        base_url = f"{parsed.scheme}://{parsed.netloc}"
+        mount_path = parsed.path if parsed.path != "/" else ""
+        return base_url, mount_path
+
     @property
     def public_agent_card(self) -> AgentCard:
         """Get the public AgentCard for this agent.
@@ -119,24 +162,42 @@ def agent_skills(self, skills: list[AgentSkill]) -> None:
     def to_starlette_app(self) -> Starlette:
         """Create a Starlette application for serving this agent via HTTP.
 
-        This method creates a Starlette application that can be used to serve
-        the agent via HTTP using the A2A protocol.
+        Automatically handles path-based mounting if a mount path was derived
+        from the http_url parameter.
 
         Returns:
             Starlette: A Starlette application configured to serve this agent.
         """
-        return A2AStarletteApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build()
+        a2a_app = A2AStarletteApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build()
+
+        if self.mount_path:
+            # Create parent app and mount the A2A app at the specified path
+            parent_app = Starlette()
+            parent_app.mount(self.mount_path, a2a_app)
+            logger.info("Mounting A2A server at path: %s", self.mount_path)
+            return parent_app
+
+        return a2a_app
 
     def to_fastapi_app(self) -> FastAPI:
         """Create a FastAPI application for serving this agent via HTTP.
 
-        This method creates a FastAPI application that can be used to serve
-        the agent via HTTP using the A2A protocol.
+        Automatically handles path-based mounting if a mount path was derived
+        from the http_url parameter.
 
         Returns:
             FastAPI: A FastAPI application configured to serve this agent.
         """
-        return A2AFastAPIApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build()
+        a2a_app = A2AFastAPIApplication(agent_card=self.public_agent_card, http_handler=self.request_handler).build()
+
+        if self.mount_path:
+            # Create parent app and mount the A2A app at the specified path
+            parent_app = FastAPI()
+            parent_app.mount(self.mount_path, a2a_app)
+            logger.info("Mounting A2A server at path: %s", self.mount_path)
+            return parent_app
+
+        return a2a_app
 
     def serve(
         self,

src/strands/session/repository_session_manager.py

@@ -133,9 +133,7 @@ def initialize(self, agent: "Agent", **kwargs: Any) -> None:
             agent.state = AgentState(session_agent.state)
 
             # Restore the conversation manager to its previous state, and get the optional prepend messages
-            prepend_messages = agent.conversation_manager.restore_from_session(
-                session_agent.conversation_manager_state
-            )
+            prepend_messages = agent.conversation_manager.restore_from_session(session_agent.conversation_manager_state)
 
             if prepend_messages is None:
                 prepend_messages = []