Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions python/copilot/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -28,6 +28,7 @@
ResumeSessionConfig,
SessionConfig,
SessionEvent,
SessionMetadata,
Tool,
ToolHandler,
ToolInvocation,
Expand Down Expand Up @@ -59,6 +60,7 @@
"ResumeSessionConfig",
"SessionConfig",
"SessionEvent",
"SessionMetadata",
"Tool",
"ToolHandler",
"ToolInvocation",
Expand Down
57 changes: 57 additions & 0 deletions python/copilot/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
ModelInfo,
ResumeSessionConfig,
SessionConfig,
SessionMetadata,
ToolHandler,
ToolInvocation,
ToolResult,
Expand Down Expand Up @@ -633,6 +634,62 @@ async def list_models(self) -> List["ModelInfo"]:
response = await self._client.request("models.list", {})
return response.get("models", [])

async def list_sessions(self) -> List["SessionMetadata"]:
"""
List all available sessions known to the server.

Returns metadata about each session including ID, timestamps, and summary.

Returns:
A list of session metadata dictionaries with keys: sessionId (str),
startTime (str), modifiedTime (str), summary (str, optional),
and isRemote (bool).

Raises:
RuntimeError: If the client is not connected.

Example:
>>> sessions = await client.list_sessions()
>>> for session in sessions:
... print(f"Session: {session['sessionId']}")
"""
if not self._client:
raise RuntimeError("Client not connected")

response = await self._client.request("session.list", {})
return response.get("sessions", [])

async def delete_session(self, session_id: str) -> None:
"""
Delete a session permanently.

This permanently removes the session and all its conversation history.
The session cannot be resumed after deletion.

Args:
session_id: The ID of the session to delete.

Raises:
RuntimeError: If the client is not connected or deletion fails.

Example:
>>> await client.delete_session("session-123")
"""
if not self._client:
raise RuntimeError("Client not connected")

response = await self._client.request("session.delete", {"sessionId": session_id})

success = response.get("success", False)
if not success:
error = response.get("error", "Unknown error")
raise RuntimeError(f"Failed to delete session {session_id}: {error}")

# Remove from local sessions map if present
with self._sessions_lock:
if session_id in self._sessions:
del self._sessions[session_id]

async def _verify_protocol_version(self) -> None:
"""Verify that the server's protocol version matches the SDK's expected version."""
expected_version = get_sdk_protocol_version()
Expand Down
10 changes: 10 additions & 0 deletions python/copilot/types.py
Original file line number Diff line number Diff line change
Expand Up @@ -359,3 +359,13 @@ class GetModelsResponse(TypedDict):
"""Response from models.list"""

models: List[ModelInfo]


class SessionMetadata(TypedDict):
"""Metadata about a session"""

sessionId: str # Session identifier
startTime: str # ISO 8601 timestamp when session was created
modifiedTime: str # ISO 8601 timestamp when session was last modified
summary: NotRequired[str] # Optional summary of the session
isRemote: bool # Whether the session is remote
47 changes: 47 additions & 0 deletions python/e2e/test_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -180,6 +180,53 @@ async def test_should_throw_error_resuming_nonexistent_session(self, ctx: E2ETes
with pytest.raises(Exception):
await ctx.client.resume_session("non-existent-session-id")

async def test_should_list_sessions(self, ctx: E2ETestContext):
# Create a couple of sessions
session1 = await ctx.client.create_session()
session2 = await ctx.client.create_session()

# List sessions and verify they're included
sessions = await ctx.client.list_sessions()
assert isinstance(sessions, list)

session_ids = [s["sessionId"] for s in sessions]
assert session1.session_id in session_ids
assert session2.session_id in session_ids

# Verify session metadata structure
for session_data in sessions:
assert "sessionId" in session_data
assert "startTime" in session_data
assert "modifiedTime" in session_data
assert "isRemote" in session_data
# summary is optional
assert isinstance(session_data["sessionId"], str)
assert isinstance(session_data["startTime"], str)
assert isinstance(session_data["modifiedTime"], str)
assert isinstance(session_data["isRemote"], bool)

async def test_should_delete_session(self, ctx: E2ETestContext):
# Create a session
session = await ctx.client.create_session()
session_id = session.session_id

# Verify session exists in the list
sessions = await ctx.client.list_sessions()
session_ids = [s["sessionId"] for s in sessions]
assert session_id in session_ids

# Delete the session
await ctx.client.delete_session(session_id)

# Verify session no longer exists in the list
sessions_after = await ctx.client.list_sessions()
session_ids_after = [s["sessionId"] for s in sessions_after]
assert session_id not in session_ids_after

# Verify we cannot resume the deleted session
with pytest.raises(Exception):
await ctx.client.resume_session(session_id)

async def test_should_create_session_with_custom_tool(self, ctx: E2ETestContext):
# This test uses the low-level Tool() API to show that Pydantic is optional
def get_secret_number_handler(invocation):
Expand Down
Loading