feat(sdk): add list_sessions to Python SDK

examples/claude/telegram_support.py

@@ -567,18 +567,11 @@ def register_agent(mt: Memtrace, name: str, description: str) -> str:
 
 
 def find_active_session(mt: Memtrace, agent_id: str, account_id: str) -> str | None:
-    """Find an active session for a customer by account_id in session metadata.
-    Searches across all agents' sessions to find one matching this customer."""
-    resp = mt._client.get("/api/v1/sessions", params={"agent_id": agent_id})
-    if resp.status_code != 200:
-        return None
-    data = resp.json()
-    for s in data.get("sessions", []):
-        if (
-            s.get("status") == "active"
-            and s.get("metadata", {}).get("account_id") == account_id
-        ):
-            return s["id"]
+    """Find an active session for a customer by account_id in session metadata."""
+    result = mt.list_sessions(agent_id=agent_id)
+    for s in result.sessions:
+        if s.status == "active" and (s.metadata or {}).get("account_id") == account_id:
+            return s.id
     return None
 
 

sdks/python/src/memtrace/__init__.py

@@ -18,6 +18,7 @@
     SearchResult,
     Session,
     SessionContext,
+    SessionList,
 )
 
 __all__ = [
@@ -44,4 +45,5 @@
     "SearchResult",
     "Session",
     "SessionContext",
+    "SessionList",
 ]

sdks/python/src/memtrace/async_client.py

@@ -19,6 +19,7 @@
     SearchResult,
     Session,
     SessionContext,
+    SessionList,
 )
 
 
@@ -162,6 +163,15 @@ async def get_session_context(
         handle_error(resp)
         return SessionContext.model_validate(resp.json())
 
+    async def list_sessions(self, agent_id: str | None = None) -> SessionList:
+        """List sessions, optionally filtered by agent."""
+        params = {}
+        if agent_id:
+            params["agent_id"] = agent_id
+        resp = await self._client.get("/api/v1/sessions", params=params)
+        handle_error(resp)
+        return SessionList.model_validate(resp.json())
+
     async def close_session(self, session_id: str) -> Session:
         """Close a session."""
         resp = await self._client.put(

sdks/python/src/memtrace/client.py

@@ -19,6 +19,7 @@
     SearchResult,
     Session,
     SessionContext,
+    SessionList,
 )
 
 
@@ -152,6 +153,15 @@ def get_session_context(
         handle_error(resp)
         return SessionContext.model_validate(resp.json())
 
+    def list_sessions(self, agent_id: str | None = None) -> SessionList:
+        """List sessions, optionally filtered by agent."""
+        params = {}
+        if agent_id:
+            params["agent_id"] = agent_id
+        resp = self._client.get("/api/v1/sessions", params=params)
+        handle_error(resp)
+        return SessionList.model_validate(resp.json())
+
     def close_session(self, session_id: str) -> Session:
         """Close a session."""
         resp = self._client.put(f"/api/v1/sessions/{session_id}", json={"status": "closed"})

sdks/python/src/memtrace/models.py

@@ -142,6 +142,13 @@ class Session(BaseModel):
     closed_at: datetime | None = None
 
 
+class SessionList(BaseModel):
+    """Response for listing sessions."""
+
+    sessions: list[Session]
+    count: int
+
+
 class CreateSessionRequest(BaseModel):
     """Request body for creating a session."""
 

sdks/python/tests/test_async_client.py

@@ -194,6 +194,28 @@ async def test_context_no_opts(self, client, mock_api):
         assert ctx.session_id == "sess_1"
 
 
+class TestListSessions:
+    async def test_list_all(self, client, mock_api):
+        mock_api.get("/api/v1/sessions").mock(
+            return_value=httpx.Response(
+                200, json={"sessions": [SESSION_JSON], "count": 1}
+            )
+        )
+        result = await client.list_sessions()
+        assert result.count == 1
+        assert result.sessions[0].id == "sess_1"
+
+    async def test_list_with_agent_id(self, client, mock_api):
+        route = mock_api.get("/api/v1/sessions").mock(
+            return_value=httpx.Response(
+                200, json={"sessions": [], "count": 0}
+            )
+        )
+        await client.list_sessions(agent_id="agent_1")
+        url = str(route.calls[0].request.url)
+        assert "agent_id=agent_1" in url
+
+
 class TestCloseSession:
     async def test_close(self, client, mock_api):
         closed = {**SESSION_JSON, "status": "closed", "closed_at": "2026-02-08T13:00:00Z"}

sdks/python/tests/test_client.py

@@ -224,6 +224,38 @@ def test_context_no_opts(self, client, mock_api):
         assert ctx.session_id == "sess_1"
 
 
+class TestListSessions:
+    def test_list_all(self, client, mock_api):
+        mock_api.get("/api/v1/sessions").mock(
+            return_value=httpx.Response(
+                200, json={"sessions": [SESSION_JSON], "count": 1}
+            )
+        )
+        result = client.list_sessions()
+        assert result.count == 1
+        assert result.sessions[0].id == "sess_1"
+
+    def test_list_with_agent_id(self, client, mock_api):
+        route = mock_api.get("/api/v1/sessions").mock(
+            return_value=httpx.Response(
+                200, json={"sessions": [], "count": 0}
+            )
+        )
+        client.list_sessions(agent_id="agent_1")
+        url = str(route.calls[0].request.url)
+        assert "agent_id=agent_1" in url
+
+    def test_list_empty(self, client, mock_api):
+        mock_api.get("/api/v1/sessions").mock(
+            return_value=httpx.Response(
+                200, json={"sessions": [], "count": 0}
+            )
+        )
+        result = client.list_sessions()
+        assert result.count == 0
+        assert result.sessions == []
+
+
 class TestCloseSession:
     def test_close(self, client, mock_api):
         closed = {**SESSION_JSON, "status": "closed", "closed_at": "2026-02-08T13:00:00Z"}