feat: add update_stream_delivery_config to MemoryClient

Add a focused method to update stream delivery configuration on a memory
without requiring callers to drop down to MemoryControlPlaneClient.

- Calls gmcp_client.update_memory with only memoryId, clientToken, and
  streamDeliveryResources — no other memory properties are modified
- Normalizes response via _normalize_memory_response
- Does not shadow the existing update_memory passthrough in _ALLOWED_GMCP_METHODS
- Unit tests: config passthrough, response normalization, ClientError propagation
- Integration test: builds on PR aws#317 TestStreamDeliveryResources with
  test_stream_delivery_update (FULL_CONTENT -> METADATA_ONLY with wait/poll)

Author: Hweinstock

src/bedrock_agentcore/memory/client.py

@@ -1259,6 +1259,33 @@ def delete_memory(self, memory_id: str) -> Dict[str, Any]:
             logger.error("Failed to delete memory: %s", e)
             raise
 
+    def update_stream_delivery_config(
+        self,
+        memory_id: str,
+        stream_delivery_resources: Dict[str, Any],
+    ) -> Dict[str, Any]:
+        """Update the stream delivery configuration for a memory.
+
+        Args:
+            memory_id: Memory resource ID
+            stream_delivery_resources: Delivery configuration for streaming memory records
+
+        Returns:
+            Updated memory object
+        """
+        try:
+            response = self.gmcp_client.update_memory(
+                memoryId=memory_id,
+                clientToken=str(uuid.uuid4()),
+                streamDeliveryResources=stream_delivery_resources,
+            )
+            memory = self._normalize_memory_response(response["memory"])
+            logger.info("Updated stream delivery config for memory: %s", memory_id)
+            return memory
+        except ClientError as e:
+            logger.error("Failed to update stream delivery config: %s", e)
+            raise
+
     def delete_memory_and_wait(self, memory_id: str, max_wait: int = 300, poll_interval: int = 10) -> Dict[str, Any]:
         """Delete a memory and wait for deletion to complete.
 

tests/bedrock_agentcore/memory/test_client.py

@@ -6,6 +6,7 @@
 from datetime import datetime
 from unittest.mock import MagicMock, patch
 
+import pytest
 from botocore.exceptions import ClientError
 
 from bedrock_agentcore.memory import MemoryClient
@@ -765,6 +766,66 @@ def test_delete_memory():
             assert kwargs["clientToken"] == "12345678-1234-5678-1234-567812345678"
 
 
+def test_update_stream_delivery_config():
+    """Test update_stream_delivery_config passes config and normalizes response."""
+    with patch("boto3.client"):
+        client = MemoryClient()
+
+        mock_gmcp = MagicMock()
+        client.gmcp_client = mock_gmcp
+
+        delivery_config = {
+            "resources": [
+                {
+                    "kinesis": {
+                        "dataStreamArn": "arn:aws:kinesis:us-west-2:123456789012:stream/test",
+                        "contentConfigurations": [{"type": "MEMORY_RECORDS", "level": "FULL_CONTENT"}],
+                    }
+                }
+            ]
+        }
+
+        mock_gmcp.update_memory.return_value = {
+            "memory": {
+                "memoryId": "mem-123",
+                "status": "ACTIVE",
+                "streamDeliveryResources": delivery_config,
+            }
+        }
+
+        with patch("uuid.uuid4", return_value=uuid.UUID("12345678-1234-5678-1234-567812345678")):
+            result = client.update_stream_delivery_config(
+                memory_id="mem-123", stream_delivery_resources=delivery_config
+            )
+
+            assert result["memoryId"] == "mem-123"
+            assert result["id"] == "mem-123"
+            assert result["streamDeliveryResources"] == delivery_config
+
+            args, kwargs = mock_gmcp.update_memory.call_args
+            assert kwargs["memoryId"] == "mem-123"
+            assert kwargs["clientToken"] == "12345678-1234-5678-1234-567812345678"
+            assert kwargs["streamDeliveryResources"] == delivery_config
+            assert set(kwargs.keys()) == {"memoryId", "clientToken", "streamDeliveryResources"}
+
+
+def test_update_stream_delivery_config_client_error():
+    """Test update_stream_delivery_config raises ClientError."""
+    with patch("boto3.client"):
+        client = MemoryClient()
+
+        mock_gmcp = MagicMock()
+        client.gmcp_client = mock_gmcp
+
+        mock_gmcp.update_memory.side_effect = ClientError(
+            {"Error": {"Code": "ResourceNotFoundException", "Message": "Not found"}},
+            "UpdateMemory",
+        )
+
+        with pytest.raises(ClientError):
+            client.update_stream_delivery_config(memory_id="mem-999", stream_delivery_resources={})
+
+
 def test_get_memory_status():
     """Test get_memory_status functionality."""
     with patch("boto3.client"):

tests_integ/memory/test_stream_delivery.py

@@ -53,6 +53,28 @@ def _make_delivery_config(self, level="FULL_CONTENT"):
             ]
         }
 
+    def _get_delivery_level(self, memory_id):
+        """Get the current stream delivery level for a memory."""
+        detail = self.client.gmcp_client.get_memory(memoryId=memory_id)["memory"]
+        for resource in detail.get("streamDeliveryResources", {}).get("resources", []):
+            configs = resource.get("kinesis", {}).get("contentConfigurations", [])
+            if configs:
+                return configs[0].get("level")
+        return None
+
+    def _wait_for_active(self, memory_id, max_wait=180, poll_interval=10):
+        """Poll until memory returns to ACTIVE status."""
+        deadline = time.time() + max_wait
+        while time.time() < deadline:
+            status = self.client.get_memory_status(memory_id)
+            if status == "ACTIVE":
+                return
+            if status == "FAILED":
+                pytest.fail(f"Memory {memory_id} entered FAILED status")
+            time.sleep(poll_interval)
+        pytest.fail(f"Memory {memory_id} did not reach ACTIVE within {max_wait}s")
+
+    @pytest.mark.order(1)
     def test_stream_delivery_create(self):
         """Create memory with stream delivery config and verify via get_memory."""
         delivery_config = self._make_delivery_config("FULL_CONTENT")
@@ -68,3 +90,26 @@ def test_stream_delivery_create(self):
         self.memory_ids.append(memory_id)
 
         assert memory["streamDeliveryResources"] == delivery_config
+
+    @pytest.mark.order(2)
+    def test_stream_delivery_update(self):
+        """Create with FULL_CONTENT, update to METADATA_ONLY, verify change."""
+        memory = self.client.create_memory_and_wait(
+            name=f"{self.test_prefix}_update",
+            strategies=[],
+            memory_execution_role_arn=self.execution_role_arn,
+            stream_delivery_resources=self._make_delivery_config("FULL_CONTENT"),
+        )
+
+        memory_id = memory.get("memoryId", memory.get("id"))
+        self.memory_ids.append(memory_id)
+
+        assert self._get_delivery_level(memory_id) == "FULL_CONTENT"
+
+        self.client.update_stream_delivery_config(
+            memory_id=memory_id,
+            stream_delivery_resources=self._make_delivery_config("METADATA_ONLY"),
+        )
+        self._wait_for_active(memory_id)
+
+        assert self._get_delivery_level(memory_id) == "METADATA_ONLY"