feat: add endpoint_url parameter to S3SessionManager

src/strands/session/s3_session_manager.py

@@ -51,6 +51,7 @@ def __init__(
         boto_session: boto3.Session | None = None,
         boto_client_config: BotocoreConfig | None = None,
         region_name: str | None = None,
+        endpoint_url: str | None = None,
         **kwargs: Any,
     ):
         """Initialize S3SessionManager with S3 storage.
@@ -63,6 +64,8 @@ def __init__(
             boto_session: Optional boto3 session
             boto_client_config: Optional boto3 client configuration
             region_name: AWS region for S3 storage
+            endpoint_url: Custom endpoint URL for S3-compatible storage backends (e.g., MinIO, LocalStack)
+                or VPC endpoints (PrivateLink)
             **kwargs: Additional keyword arguments for future extensibility.
         """
         self.bucket = bucket
@@ -82,7 +85,7 @@ def __init__(
         else:
             client_config = BotocoreConfig(user_agent_extra="strands-agents")
 
-        self.client = session.client(service_name="s3", config=client_config)
+        self.client = session.client(service_name="s3", config=client_config, endpoint_url=endpoint_url)
         super().__init__(session_id=session_id, session_repository=self)
 
     def _get_session_path(self, session_id: str) -> str:

tests/strands/session/test_s3_session_manager.py

@@ -89,6 +89,13 @@ def test_init_s3_session_manager_with_existing_user_agent(mocked_aws, s3_bucket)
     assert "strands-agents" in session_manager.client.meta.config.user_agent_extra
 
 
+def test_init_s3_session_manager_with_endpoint_url(mocked_aws, s3_bucket, monkeypatch):
+    custom_endpoint = "http://localhost:9000"
+    monkeypatch.setenv("MOTO_S3_CUSTOM_ENDPOINTS", custom_endpoint)
+    session_manager = S3SessionManager(session_id="test", bucket=s3_bucket, endpoint_url=custom_endpoint)
+    assert session_manager.client.meta.endpoint_url == custom_endpoint
+
+
 def test_empty_prefix_session_roundtrip(mocked_aws, s3_bucket, sample_session, sample_agent):
     """Test that session data can be written and read back with default empty prefix."""
     manager = S3SessionManager(session_id="test", bucket=s3_bucket, prefix="", region_name="us-west-2")