From 126de1a043eebbe72b047d61b0127c593952253e Mon Sep 17 00:00:00 2001 From: jer Date: Wed, 23 Jul 2025 10:27:42 -0400 Subject: [PATCH 1/3] feat(a2a): support mounts for containerized deployments --- src/strands/multiagent/a2a/server.py | 67 ++++++- tests/strands/multiagent/a2a/test_server.py | 190 ++++++++++++++++++++ 2 files changed, 250 insertions(+), 7 deletions(-) diff --git a/src/strands/multiagent/a2a/server.py b/src/strands/multiagent/a2a/server.py index de891499d..618a01258 100644 --- a/src/strands/multiagent/a2a/server.py +++ b/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,7 @@ def __init__( # AgentCard host: str = "0.0.0.0", port: int = 9000, + http_url: str | None = None, version: str = "0.0.1", skills: list[AgentSkill] | None = None, ): @@ -40,13 +42,27 @@ 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" 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("/") + "/" + 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 +74,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 +154,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, diff --git a/tests/strands/multiagent/a2a/test_server.py b/tests/strands/multiagent/a2a/test_server.py index 74f470741..1cf6e4ef3 100644 --- a/tests/strands/multiagent/a2a/test_server.py +++ b/tests/strands/multiagent/a2a/test_server.py @@ -509,3 +509,193 @@ def test_serve_handles_general_exception(mock_run, mock_strands_agent, caplog): assert "Strands A2A server encountered exception" in caplog.text assert "Strands A2A server has shutdown" in caplog.text + + +# Tests for http_url parameter and path mounting functionality + + +def test_initialization_with_http_url_no_path(mock_strands_agent): + """Test initialization with http_url containing no path.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer( + mock_strands_agent, host="0.0.0.0", port=8080, http_url="http://my-alb.amazonaws.com", skills=[] + ) + + assert a2a_agent.host == "0.0.0.0" + assert a2a_agent.port == 8080 + assert a2a_agent.http_url == "http://my-alb.amazonaws.com/" + assert a2a_agent.public_base_url == "http://my-alb.amazonaws.com" + assert a2a_agent.mount_path == "" + + +def test_initialization_with_http_url_with_path(mock_strands_agent): + """Test initialization with http_url containing a path for mounting.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer( + mock_strands_agent, host="0.0.0.0", port=8080, http_url="http://my-alb.amazonaws.com/agent1", skills=[] + ) + + assert a2a_agent.host == "0.0.0.0" + assert a2a_agent.port == 8080 + assert a2a_agent.http_url == "http://my-alb.amazonaws.com/agent1/" + assert a2a_agent.public_base_url == "http://my-alb.amazonaws.com" + assert a2a_agent.mount_path == "/agent1" + + +def test_initialization_with_https_url(mock_strands_agent): + """Test initialization with HTTPS URL.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="https://my-alb.amazonaws.com/secure-agent", skills=[]) + + assert a2a_agent.http_url == "https://my-alb.amazonaws.com/secure-agent/" + assert a2a_agent.public_base_url == "https://my-alb.amazonaws.com" + assert a2a_agent.mount_path == "/secure-agent" + + +def test_initialization_with_http_url_with_port(mock_strands_agent): + """Test initialization with http_url containing explicit port.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="http://my-server.com:8080/api/agent", skills=[]) + + assert a2a_agent.http_url == "http://my-server.com:8080/api/agent/" + assert a2a_agent.public_base_url == "http://my-server.com:8080" + assert a2a_agent.mount_path == "/api/agent" + + +def test_parse_public_url_method(mock_strands_agent): + """Test the _parse_public_url method directly.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + a2a_agent = A2AServer(mock_strands_agent, skills=[]) + + # Test various URL formats + base_url, mount_path = a2a_agent._parse_public_url("http://example.com/path") + assert base_url == "http://example.com" + assert mount_path == "/path" + + base_url, mount_path = a2a_agent._parse_public_url("https://example.com:443/deep/path") + assert base_url == "https://example.com:443" + assert mount_path == "/deep/path" + + base_url, mount_path = a2a_agent._parse_public_url("http://example.com/") + assert base_url == "http://example.com" + assert mount_path == "" + + base_url, mount_path = a2a_agent._parse_public_url("http://example.com") + assert base_url == "http://example.com" + assert mount_path == "" + + +def test_public_agent_card_with_http_url(mock_strands_agent): + """Test that public_agent_card uses the http_url when provided.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="https://my-alb.amazonaws.com/agent1", skills=[]) + + card = a2a_agent.public_agent_card + + assert isinstance(card, AgentCard) + assert card.url == "https://my-alb.amazonaws.com/agent1/" + assert card.name == "Test Agent" + assert card.description == "A test agent for unit testing" + + +def test_to_starlette_app_with_mounting(mock_strands_agent): + """Test that to_starlette_app creates mounted app when mount_path exists.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="http://example.com/agent1", skills=[]) + + app = a2a_agent.to_starlette_app() + + assert isinstance(app, Starlette) + # The returned app should be the parent app with the A2A app mounted + # We can't easily inspect the mounting, but we can verify it's a Starlette app + + +def test_to_starlette_app_without_mounting(mock_strands_agent): + """Test that to_starlette_app creates regular app when no mount_path.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="http://example.com", skills=[]) + + app = a2a_agent.to_starlette_app() + + assert isinstance(app, Starlette) + + +def test_to_fastapi_app_with_mounting(mock_strands_agent): + """Test that to_fastapi_app creates mounted app when mount_path exists.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="http://example.com/agent1", skills=[]) + + app = a2a_agent.to_fastapi_app() + + assert isinstance(app, FastAPI) + + +def test_to_fastapi_app_without_mounting(mock_strands_agent): + """Test that to_fastapi_app creates regular app when no mount_path.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="http://example.com", skills=[]) + + app = a2a_agent.to_fastapi_app() + + assert isinstance(app, FastAPI) + + +def test_backwards_compatibility_without_http_url(mock_strands_agent): + """Test that the old behavior is preserved when http_url is not provided.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, host="localhost", port=9000, skills=[]) + + # Should behave exactly like before + assert a2a_agent.host == "localhost" + assert a2a_agent.port == 9000 + assert a2a_agent.http_url == "http://localhost:9000/" + assert a2a_agent.public_base_url == "http://localhost:9000" + assert a2a_agent.mount_path == "" + + # Agent card should use the traditional URL + card = a2a_agent.public_agent_card + assert card.url == "http://localhost:9000/" + + +def test_mount_path_logging(mock_strands_agent, caplog): + """Test that mounting logs the correct message.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + a2a_agent = A2AServer(mock_strands_agent, http_url="http://example.com/test-agent", skills=[]) + + # Test Starlette app mounting logs + caplog.clear() + a2a_agent.to_starlette_app() + assert "Mounting A2A server at path: /test-agent" in caplog.text + + # Test FastAPI app mounting logs + caplog.clear() + a2a_agent.to_fastapi_app() + assert "Mounting A2A server at path: /test-agent" in caplog.text + + +def test_http_url_trailing_slash_handling(mock_strands_agent): + """Test that trailing slashes in http_url are handled correctly.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + # Test with trailing slash + a2a_agent1 = A2AServer(mock_strands_agent, http_url="http://example.com/agent1/", skills=[]) + + # Test without trailing slash + a2a_agent2 = A2AServer(mock_strands_agent, http_url="http://example.com/agent1", skills=[]) + + # Both should result in the same normalized URL + assert a2a_agent1.http_url == "http://example.com/agent1/" + assert a2a_agent2.http_url == "http://example.com/agent1/" + assert a2a_agent1.mount_path == "/agent1" + assert a2a_agent2.mount_path == "/agent1" From ff8d828451c7ac6612d15512e2067e29d22524f8 Mon Sep 17 00:00:00 2001 From: jer Date: Wed, 23 Jul 2025 11:29:03 -0400 Subject: [PATCH 2/3] feat(a2a): escape hatch for load balancers which strip paths --- src/strands/multiagent/a2a/server.py | 8 + .../session/repository_session_manager.py | 4 +- tests/strands/multiagent/a2a/test_server.py | 158 ++++++++++++++++++ 3 files changed, 167 insertions(+), 3 deletions(-) diff --git a/src/strands/multiagent/a2a/server.py b/src/strands/multiagent/a2a/server.py index 618a01258..fa7b6b887 100644 --- a/src/strands/multiagent/a2a/server.py +++ b/src/strands/multiagent/a2a/server.py @@ -33,6 +33,7 @@ def __init__( 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, ): @@ -46,6 +47,9 @@ def __init__( 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. """ @@ -57,6 +61,10 @@ def __init__( # 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}" diff --git a/src/strands/session/repository_session_manager.py b/src/strands/session/repository_session_manager.py index 18a6ac474..75058b251 100644 --- a/src/strands/session/repository_session_manager.py +++ b/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 = [] diff --git a/tests/strands/multiagent/a2a/test_server.py b/tests/strands/multiagent/a2a/test_server.py index 1cf6e4ef3..c6f36818e 100644 --- a/tests/strands/multiagent/a2a/test_server.py +++ b/tests/strands/multiagent/a2a/test_server.py @@ -699,3 +699,161 @@ def test_http_url_trailing_slash_handling(mock_strands_agent): assert a2a_agent2.http_url == "http://example.com/agent1/" assert a2a_agent1.mount_path == "/agent1" assert a2a_agent2.mount_path == "/agent1" + + +def test_serve_at_root_default_behavior(mock_strands_agent): + """Test default behavior extracts mount path from http_url.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + server = A2AServer(mock_strands_agent, http_url="http://my-alb.com/agent1", skills=[]) + + assert server.mount_path == "/agent1" + assert server.http_url == "http://my-alb.com/agent1/" + + +def test_serve_at_root_overrides_mounting(mock_strands_agent): + """Test serve_at_root=True overrides automatic path mounting.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + server = A2AServer(mock_strands_agent, http_url="http://my-alb.com/agent1", serve_at_root=True, skills=[]) + + assert server.mount_path == "" # Should be empty despite path in URL + assert server.http_url == "http://my-alb.com/agent1/" # Public URL unchanged + + +def test_serve_at_root_with_no_path(mock_strands_agent): + """Test serve_at_root=True when no path in URL (redundant but valid).""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + server = A2AServer(mock_strands_agent, host="localhost", port=8080, serve_at_root=True, skills=[]) + + assert server.mount_path == "" + assert server.http_url == "http://localhost:8080/" + + +def test_serve_at_root_complex_path(mock_strands_agent): + """Test serve_at_root=True with complex nested paths.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + server = A2AServer( + mock_strands_agent, http_url="http://api.example.com/v1/agents/my-agent", serve_at_root=True, skills=[] + ) + + assert server.mount_path == "" + assert server.http_url == "http://api.example.com/v1/agents/my-agent/" + + +def test_serve_at_root_fastapi_mounting_behavior(mock_strands_agent): + """Test FastAPI mounting behavior with serve_at_root.""" + from fastapi.testclient import TestClient + + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + # Normal mounting + server_mounted = A2AServer(mock_strands_agent, http_url="http://my-alb.com/agent1", skills=[]) + app_mounted = server_mounted.to_fastapi_app() + client_mounted = TestClient(app_mounted) + + # Should work at mounted path + response = client_mounted.get("/agent1/.well-known/agent.json") + assert response.status_code == 200 + + # Should not work at root + response = client_mounted.get("/.well-known/agent.json") + assert response.status_code == 404 + + +def test_serve_at_root_fastapi_root_behavior(mock_strands_agent): + """Test FastAPI serve_at_root behavior.""" + from fastapi.testclient import TestClient + + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + # Serve at root + server_root = A2AServer(mock_strands_agent, http_url="http://my-alb.com/agent1", serve_at_root=True, skills=[]) + app_root = server_root.to_fastapi_app() + client_root = TestClient(app_root) + + # Should work at root + response = client_root.get("/.well-known/agent.json") + assert response.status_code == 200 + + # Should not work at mounted path (since we're serving at root) + response = client_root.get("/agent1/.well-known/agent.json") + assert response.status_code == 404 + + +def test_serve_at_root_starlette_behavior(mock_strands_agent): + """Test Starlette serve_at_root behavior.""" + from starlette.testclient import TestClient + + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + # Normal mounting + server_mounted = A2AServer(mock_strands_agent, http_url="http://my-alb.com/agent1", skills=[]) + app_mounted = server_mounted.to_starlette_app() + client_mounted = TestClient(app_mounted) + + # Should work at mounted path + response = client_mounted.get("/agent1/.well-known/agent.json") + assert response.status_code == 200 + + # Serve at root + server_root = A2AServer(mock_strands_agent, http_url="http://my-alb.com/agent1", serve_at_root=True, skills=[]) + app_root = server_root.to_starlette_app() + client_root = TestClient(app_root) + + # Should work at root + response = client_root.get("/.well-known/agent.json") + assert response.status_code == 200 + + +def test_serve_at_root_alb_scenarios(mock_strands_agent): + """Test common ALB deployment scenarios.""" + from fastapi.testclient import TestClient + + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + # ALB with path preservation + server_preserved = A2AServer(mock_strands_agent, http_url="http://my-alb.amazonaws.com/agent1", skills=[]) + app_preserved = server_preserved.to_fastapi_app() + client_preserved = TestClient(app_preserved) + + # Container receives /agent1/.well-known/agent.json + response = client_preserved.get("/agent1/.well-known/agent.json") + assert response.status_code == 200 + agent_data = response.json() + assert agent_data["url"] == "http://my-alb.amazonaws.com/agent1/" + + # ALB with path stripping + server_stripped = A2AServer( + mock_strands_agent, http_url="http://my-alb.amazonaws.com/agent1", serve_at_root=True, skills=[] + ) + app_stripped = server_stripped.to_fastapi_app() + client_stripped = TestClient(app_stripped) + + # Container receives /.well-known/agent.json (path stripped by ALB) + response = client_stripped.get("/.well-known/agent.json") + assert response.status_code == 200 + agent_data = response.json() + assert agent_data["url"] == "http://my-alb.amazonaws.com/agent1/" + + +def test_serve_at_root_edge_cases(mock_strands_agent): + """Test edge cases for serve_at_root parameter.""" + mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} + + # Root path in URL + server1 = A2AServer(mock_strands_agent, http_url="http://example.com/", skills=[]) + assert server1.mount_path == "" + + # serve_at_root should be redundant but not cause issues + server2 = A2AServer(mock_strands_agent, http_url="http://example.com/", serve_at_root=True, skills=[]) + assert server2.mount_path == "" + + # Multiple nested paths + server3 = A2AServer( + mock_strands_agent, http_url="http://api.example.com/v1/agents/team1/agent1", serve_at_root=True, skills=[] + ) + assert server3.mount_path == "" + assert server3.http_url == "http://api.example.com/v1/agents/team1/agent1/" From 5205873fef776d28c3798f6d8f37dc9dc32ba39d Mon Sep 17 00:00:00 2001 From: jer Date: Wed, 23 Jul 2025 11:51:47 -0400 Subject: [PATCH 3/3] feat(a2a): formatting --- tests/strands/multiagent/a2a/test_server.py | 5 ----- 1 file changed, 5 deletions(-) diff --git a/tests/strands/multiagent/a2a/test_server.py b/tests/strands/multiagent/a2a/test_server.py index c6f36818e..fc76b5f1d 100644 --- a/tests/strands/multiagent/a2a/test_server.py +++ b/tests/strands/multiagent/a2a/test_server.py @@ -511,9 +511,6 @@ def test_serve_handles_general_exception(mock_run, mock_strands_agent, caplog): assert "Strands A2A server has shutdown" in caplog.text -# Tests for http_url parameter and path mounting functionality - - def test_initialization_with_http_url_no_path(mock_strands_agent): """Test initialization with http_url containing no path.""" mock_strands_agent.tool_registry.get_all_tools_config.return_value = {} @@ -612,8 +609,6 @@ def test_to_starlette_app_with_mounting(mock_strands_agent): app = a2a_agent.to_starlette_app() assert isinstance(app, Starlette) - # The returned app should be the parent app with the A2A app mounted - # We can't easily inspect the mounting, but we can verify it's a Starlette app def test_to_starlette_app_without_mounting(mock_strands_agent):