fix: encode dynamic URL path segments (#5963)

Co-authored-by: kartik-mem0 <kartik.labhshetwar@mem0.ai>
This commit is contained in:
VectorPeak
2026-07-07 18:41:54 +08:00
committed by GitHub
parent 87276ef968
commit cc52f0e367
9 changed files with 374 additions and 39 deletions
+24 -5
View File
@@ -3,6 +3,7 @@
from __future__ import annotations
from typing import Any
from urllib.parse import quote
import httpx
@@ -11,6 +12,10 @@ from mem0_cli.backend.base import Backend
from mem0_cli.config import PlatformConfig
def _encode_path_segment(value: Any) -> str:
return quote(str(value), safe="")
class PlatformBackend(Backend):
"""Backend that talks to the mem0 Platform API."""
@@ -196,7 +201,11 @@ class PlatformBackend(Backend):
)
def get(self, memory_id: str) -> dict:
return self._request("GET", f"/v1/memories/{memory_id}/", params={"source": "CLI"})
return self._request(
"GET",
f"/v1/memories/{_encode_path_segment(memory_id)}/",
params={"source": "CLI"},
)
def list_memories(
self,
@@ -250,7 +259,11 @@ class PlatformBackend(Backend):
if metadata:
payload["metadata"] = metadata
payload["source"] = "CLI"
return self._request("PUT", f"/v1/memories/{memory_id}/", json=payload)
return self._request(
"PUT",
f"/v1/memories/{_encode_path_segment(memory_id)}/",
json=payload,
)
def delete(
self,
@@ -274,7 +287,11 @@ class PlatformBackend(Backend):
params["run_id"] = run_id
return self._request("DELETE", "/v1/memories/", params=params)
elif memory_id:
return self._request("DELETE", f"/v1/memories/{memory_id}/", params={"source": "CLI"})
return self._request(
"DELETE",
f"/v1/memories/{_encode_path_segment(memory_id)}/",
params={"source": "CLI"},
)
else:
raise ValueError("Either memory_id or --all is required")
@@ -302,7 +319,9 @@ class PlatformBackend(Backend):
results: dict = {}
for entity_type, entity_id in entities.items():
results[entity_type] = self._request(
"DELETE", f"/v2/entities/{entity_type}/{entity_id}/", params={"source": "CLI"}
"DELETE",
f"/v2/entities/{_encode_path_segment(entity_type)}/{_encode_path_segment(entity_id)}/",
params={"source": "CLI"},
)
return results
@@ -348,7 +367,7 @@ class PlatformBackend(Backend):
return result if isinstance(result, list) else result.get("results", [])
def get_event(self, event_id: str) -> dict:
return self._request("GET", f"/v1/event/{event_id}/")
return self._request("GET", f"/v1/event/{_encode_path_segment(event_id)}/")
class AuthError(Exception):