fix(agent-memory): encode memory and message identifiers in URL paths

src/sap_cloud_sdk/agent_memory/client.py

@@ -88,6 +88,18 @@ def _validate_filter_clauses(
             )
 
 
+def _encode_entity_id(value: object) -> str:
+    """Percent-encode an identifier for use in a URL path.
+
+    Encoding failures raise AgentMemoryHttpError.
+    """
+    segment = f"{value}"
+    try:
+        return quote(segment, safe="")
+    except UnicodeEncodeError as exc:
+        raise AgentMemoryHttpError(str(exc)) from exc
+
+
 class AgentMemoryClient:
     """Client for the SAP Agent Memory service (v1 API).
 
@@ -240,7 +252,9 @@ def get_memory(self, memory_id: str) -> Memory:
         """
         _require_non_empty(memory_id=memory_id)
         data = self._request(
-            HttpMethod.GET, f"{MEMORIES}({memory_id})", tenant_subdomain=self._tenant
+            HttpMethod.GET,
+            f"{MEMORIES}({_encode_entity_id(memory_id)})",
+            tenant_subdomain=self._tenant,
         )
         return Memory.from_dict(data)
 
@@ -277,7 +291,7 @@ def update_memory(
             payload[_METADATA_KEY] = metadata
         self._request(
             HttpMethod.PATCH,
-            f"{MEMORIES}({memory_id})",
+            f"{MEMORIES}({_encode_entity_id(memory_id)})",
             json=payload,
             tenant_subdomain=self._tenant,
         )
@@ -296,7 +310,9 @@ def delete_memory(self, memory_id: str) -> None:
         """
         _require_non_empty(memory_id=memory_id)
         self._request(
-            HttpMethod.DELETE, f"{MEMORIES}({memory_id})", tenant_subdomain=self._tenant
+            HttpMethod.DELETE,
+            f"{MEMORIES}({_encode_entity_id(memory_id)})",
+            tenant_subdomain=self._tenant,
         )
 
     @record_metrics(Module.AGENT_MEMORY, Operation.AGENT_MEMORY_LIST_MEMORIES)
@@ -497,7 +513,9 @@ def get_message(self, message_id: str) -> Message:
         """
         _require_non_empty(message_id=message_id)
         data = self._request(
-            HttpMethod.GET, f"{MESSAGES}({message_id})", tenant_subdomain=self._tenant
+            HttpMethod.GET,
+            f"{MESSAGES}({_encode_entity_id(message_id)})",
+            tenant_subdomain=self._tenant,
         )
         return Message.from_dict(data)
 
@@ -516,7 +534,7 @@ def delete_message(self, message_id: str) -> None:
         _require_non_empty(message_id=message_id)
         self._request(
             HttpMethod.DELETE,
-            f"{MESSAGES}({message_id})",
+            f"{MESSAGES}({_encode_entity_id(message_id)})",
             tenant_subdomain=self._tenant,
         )
 

tests/agent_memory/unit/test_url_identifiers.py

@@ -0,0 +1,102 @@
+"""Regression tests for Agent Memory entity-key URL encoding."""
+
+import json
+from unittest.mock import Mock, patch
+
+import pytest
+from requests import Response, Session
+
+from sap_cloud_sdk.agent_memory.client import AgentMemoryClient
+from sap_cloud_sdk.agent_memory.exceptions import AgentMemoryHttpError
+from sap_cloud_sdk.core.protocol.http import AuthProvider, HttpClient
+
+
+ENTITY_ID = "4056bb6c-f544-41d7-87e1-ffe818573e6e"
+OPERATIONS = [
+    ("get_memory", "GET", "memories"),
+    ("update_memory", "PATCH", "memories"),
+    ("delete_memory", "DELETE", "memories"),
+    ("get_message", "GET", "messages"),
+    ("delete_message", "DELETE", "messages"),
+]
+
+
+def _response():
+    response = Response()
+    response.status_code = 200
+    response._content = json.dumps(
+        {
+            "id": ENTITY_ID,
+            "agentID": "agent",
+            "invokerID": "invoker",
+            "content": "memory",
+            "messageGroup": "group",
+            "role": "USER",
+        }
+    ).encode()
+    return response
+
+
+@pytest.mark.parametrize("operation,method,entity", OPERATIONS)
+@pytest.mark.parametrize(
+    "identifier,encoded",
+    [
+        (
+            "x)/../OtherEntity?injected=yes#fragment",
+            "x%29%2F..%2FOtherEntity%3Finjected%3Dyes%23fragment",
+        ),
+        ("%29%2F%2e%2e%2Fother", "%2529%252F%252e%252e%252Fother"),
+    ],
+)
+def test_prepared_urls_contain_ids_without_changing_odata_syntax(
+    operation, method, entity, identifier, encoded
+):
+    with Session() as session:
+        auth = Mock(spec=AuthProvider)
+        auth.get_session.return_value = session
+        http = HttpClient("https://memory.example.invalid/agent-memory", auth)
+        client = AgentMemoryClient(http, tenant="subscriber")
+        kwargs = {"content": "updated"} if operation == "update_memory" else {}
+        with patch.object(session, "send", return_value=_response()) as send:
+            getattr(client, operation)(identifier, **kwargs)
+
+        prepared = send.call_args.args[0]
+        assert prepared.method == method
+        assert (
+            prepared.url
+            == f"https://memory.example.invalid/agent-memory/v1/{entity}({encoded})"
+        )
+        if operation == "update_memory":
+            assert json.loads(prepared.body) == {"content": "updated"}
+        else:
+            assert prepared.body is None
+
+
+def test_uuid_spelling_is_preserved():
+    identifier = ENTITY_ID.upper()
+    http = Mock(spec=HttpClient)
+    http.request.return_value = _response()
+    client = AgentMemoryClient(http, tenant="subscriber")
+
+    client.get_memory(identifier)
+
+    assert http.request.call_args.args[1] == f"/v1/memories({identifier})"
+
+
+def test_dot_values_remain_inside_odata_keys():
+    http = Mock(spec=HttpClient)
+    http.request.return_value = _response()
+    client = AgentMemoryClient(http, tenant="subscriber")
+
+    client.delete_memory("..")
+
+    assert http.request.call_args.args[1] == "/v1/memories(..)"
+
+
+def test_unencodable_ids_raise_the_existing_http_exception():
+    http = Mock(spec=HttpClient)
+    client = AgentMemoryClient(http, tenant="subscriber")
+    with pytest.raises(AgentMemoryHttpError) as error:
+        client.delete_memory("\ud800")
+    assert isinstance(error.value.__cause__, UnicodeEncodeError)
+    http.request.assert_not_called()