Addressing PR comments

python/README.md

@@ -14,6 +14,45 @@ uv pip install -e ".[dev]"
 
 ## Quick Start
 
+### Using Context Managers (Recommended)
+
+The SDK supports Python's async context manager protocol for automatic resource cleanup:
+
+```python
+import asyncio
+from copilot import CopilotClient
+
+async def main():
+    # Client automatically starts on enter and cleans up on exit
+    async with CopilotClient() as client:
+        # Create a session with automatic cleanup
+        async with await client.create_session({"model": "gpt-5"}) as session:
+            # Wait for response using session.idle event
+            done = asyncio.Event()
+
+            def on_event(event):
+                if event.type.value == "assistant.message":
+                    print(event.data.content)
+                elif event.type.value == "session.idle":
+                    done.set()
+
+            session.on(on_event)
+
+            # Send a message and wait for completion
+            await session.send({"prompt": "What is 2+2?"})
+            await done.wait()
+
+        # Session automatically destroyed here
+
+    # Client automatically stopped here
+
+asyncio.run(main())
+```
+
+### Manual Resource Management
+
+You can also manage resources manually:
+
 ```python
 import asyncio
 from copilot import CopilotClient
@@ -56,6 +95,7 @@ asyncio.run(main())
 - ✅ Session history with `get_messages()`
 - ✅ Type hints throughout
 - ✅ Async/await native
+- ✅ Async context manager support for automatic resource cleanup
 
 ## API Reference
 
@@ -140,6 +180,44 @@ unsubscribe()
 - `session.foreground` - A session became the foreground session in TUI
 - `session.background` - A session is no longer the foreground session
 
+### Context Manager Support
+
+Both `CopilotClient` and `CopilotSession` support Python's async context manager protocol for automatic resource cleanup. This is the recommended pattern as it ensures resources are properly cleaned up even if exceptions occur.
+
+**CopilotClient Context Manager:**
+
+```python
+async with CopilotClient() as client:
+    # Client automatically starts on enter
+    session = await client.create_session()
+    await session.send({"prompt": "Hello!"})
+    # Client automatically stops on exit, cleaning up all sessions
+```
+
+**CopilotSession Context Manager:**
+
+```python
+async with await client.create_session() as session:
+    await session.send({"prompt": "Hello!"})
+    # Session automatically destroyed on exit
+```
+
+**Nested Context Managers:**
+
+```python
+async with CopilotClient() as client:
+    async with await client.create_session() as session:
+        await session.send({"prompt": "Hello!"})
+    # Session destroyed here
+# Client stopped here
+```
+
+**Benefits:**
+- Prevents resource leaks by ensuring cleanup even if exceptions occur
+- Eliminates the need to manually call `stop()` and `destroy()`
+- Follows Python best practices for resource management
+- Particularly useful in batch operations and evaluations to prevent process accumulation
+
 ### Tools
 
 Define tools with automatic JSON schema generation using the `@define_tool` decorator and Pydantic models:

python/copilot/client.py

@@ -14,13 +14,15 @@
 
 import asyncio
 import inspect
+import logging
 import os
 import re
 import subprocess
 import sys
 import threading
 from dataclasses import asdict, is_dataclass
 from pathlib import Path
+from types import TracebackType
 from typing import Any, Callable, Optional, cast
 
 from .generated.rpc import ServerRpc
@@ -206,6 +208,52 @@ def __init__(self, options: Optional[CopilotClientOptions] = None):
         self._lifecycle_handlers_lock = threading.Lock()
         self._rpc: Optional[ServerRpc] = None
 
+    async def __aenter__(self) -> "CopilotClient":
+        """
+        Enter the async context manager.
+
+        Automatically starts the CLI server and establishes a connection if not
+        already connected.
+
+        Returns:
+            The CopilotClient instance.
+
+        Example:
+            >>> async with CopilotClient() as client:
+            ...     session = await client.create_session()
+            ...     await session.send({"prompt": "Hello!"})
+        """
+        await self.start()
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> bool:
+        """
+        Exit the async context manager.
+
+        Performs graceful cleanup by destroying all active sessions and stopping
+        the CLI server. If cleanup errors occur, they are logged but do not
+        prevent the context from exiting.
+
+        Args:
+            exc_type: The type of exception that occurred, if any.
+            exc_val: The exception instance that occurred, if any.
+            exc_tb: The traceback of the exception that occurred, if any.
+
+        Returns:
+            False to propagate any exception that occurred in the context.
+        """
+        try:
+            await self.stop()
+        except Exception as e:
+            # Log the error but don't raise - we want cleanup to always complete
+            logging.warning(f"Error during CopilotClient cleanup: {e}")
+        return False
+
     @property
     def rpc(self) -> ServerRpc:
         """Typed server-scoped RPC methods."""

python/copilot/session.py

@@ -7,7 +7,9 @@
 
 import asyncio
 import inspect
+import logging
 import threading
+from types import TracebackType
 from typing import Any, Callable, Optional
 
 from .generated.rpc import SessionRpc
@@ -82,6 +84,50 @@ def __init__(self, session_id: str, client: Any, workspace_path: Optional[str] =
         self._hooks_lock = threading.Lock()
         self._rpc: Optional[SessionRpc] = None
 
+    async def __aenter__(self) -> "CopilotSession":
+        """
+        Enter the async context manager.
+
+        Returns the session instance, ready for use. The session must already be
+        created (via CopilotClient.create_session or resume_session).
+
+        Returns:
+            The CopilotSession instance.
+
+        Example:
+            >>> async with await client.create_session() as session:
+            ...     await session.send({"prompt": "Hello!"})
+        """
+        return self
+
+    async def __aexit__(
+        self,
+        exc_type: Optional[type[BaseException]],
+        exc_val: Optional[BaseException],
+        exc_tb: Optional[TracebackType],
+    ) -> bool:
+        """
+        Exit the async context manager.
+
+        Automatically destroys the session and releases all associated resources.
+        If an error occurs during cleanup, it is logged but does not prevent the
+        context from exiting.
+
+        Args:
+            exc_type: The type of exception that occurred, if any.
+            exc_val: The exception instance that occurred, if any.
+            exc_tb: The traceback of the exception that occurred, if any.
+
+        Returns:
+            False to propagate any exception that occurred in the context.
+        """
+        try:
+            await self.destroy()
+        except Exception as e:
+            # Log the error but don't raise - we want cleanup to always complete
+            logging.warning(f"Error during CopilotSession cleanup: {e}")
+        return False
+
     @property
     def rpc(self) -> SessionRpc:
         """Typed session-scoped RPC methods."""

python/e2e/test_context_managers.py

@@ -0,0 +1,142 @@
+"""E2E Context Manager Tests"""
+
+import pytest
+
+from copilot import CopilotClient
+
+from .testharness import CLI_PATH
+
+pytestmark = pytest.mark.asyncio(loop_scope="module")
+
+
+class TestCopilotClientContextManager:
+    async def test_should_auto_start_and_cleanup_with_context_manager(self):
+        """Test that CopilotClient context manager auto-starts and cleans up."""
+        async with CopilotClient({"cli_path": CLI_PATH}) as client:
+            assert client.get_state() == "connected"
+            # Verify we can use the client
+            pong = await client.ping("test")
+            assert pong.message == "pong: test"
+
+        # After exiting context, client should be disconnected
+        assert client.get_state() == "disconnected"
+
+    async def test_should_create_session_in_context(self):
+        """Test creating and using a session within client context."""
+        async with CopilotClient({"cli_path": CLI_PATH}) as client:
+            session = await client.create_session({"model": "fake-test-model"})
+            assert session.session_id
+
+            # Verify session is usable
+            messages = await session.get_messages()
+            assert len(messages) > 0
+            assert messages[0].type.value == "session.start"
+
+        # After exiting context, verify cleanup happened
+        assert client.get_state() == "disconnected"
+
+    async def test_should_cleanup_multiple_sessions(self):
+        """Test that all sessions are cleaned up when client context exits."""
+        async with CopilotClient({"cli_path": CLI_PATH}) as client:
+            session1 = await client.create_session()
+            session2 = await client.create_session()
+            session3 = await client.create_session()
+
+            assert session1.session_id
+            assert session2.session_id
+            assert session3.session_id
+
+        # All sessions should be cleaned up
+        assert client.get_state() == "disconnected"
+
+    async def test_should_propagate_exceptions(self):
+        """Test that exceptions within context are propagated."""
+        with pytest.raises(ValueError, match="test error"):
+            async with CopilotClient({"cli_path": CLI_PATH}) as client:
+                assert client.get_state() == "connected"
+                raise ValueError("test error")
+
+        # Client should still be cleaned up even after exception
+        assert client.get_state() == "disconnected"
+
+    async def test_should_handle_cleanup_errors_gracefully(self):
+        """Test that cleanup errors don't prevent context from exiting."""
+        async with CopilotClient({"cli_path": CLI_PATH}) as client:
+            await client.create_session()
+
+            # Kill the process to force cleanup to fail
+            if client._process:
+                client._process.kill()
+
+        # Context should still exit successfully despite cleanup errors
+        assert client.get_state() == "disconnected"
+
+
+class TestCopilotSessionContextManager:
+    async def test_should_cleanup_session_with_context_manager(self):
+        """Test that CopilotSession context manager cleans up session."""
+        client = CopilotClient({"cli_path": CLI_PATH})
+        await client.start()
+
+        try:
+            async with await client.create_session() as session:
+                assert session.session_id
+                # Send a message to verify session is working
+                await session.send({"prompt": "Hello!"})
+
+            # After exiting context, session should be destroyed
+            with pytest.raises(Exception, match="Session not found"):
+                await session.get_messages()
+        finally:
+            await client.force_stop()
+
+    async def test_should_propagate_exceptions_in_session_context(self):
+        """Test that exceptions within session context are propagated."""
+        client = CopilotClient({"cli_path": CLI_PATH})
+        await client.start()
+
+        try:
+            with pytest.raises(ValueError, match="test session error"):
+                async with await client.create_session() as session:
+                    assert session.session_id
+                    raise ValueError("test session error")
+
+            # Session should still be cleaned up after exception
+            with pytest.raises(Exception, match="Session not found"):
+                await session.get_messages()
+        finally:
+            await client.force_stop()
+
+    async def test_nested_context_managers(self):
+        """Test using nested context managers for client and session."""
+        async with CopilotClient({"cli_path": CLI_PATH}) as client:
+            async with await client.create_session() as session:
+                assert session.session_id
+                await session.send({"prompt": "Test message"})
+
+            # Session should be cleaned up
+            with pytest.raises(Exception, match="Session not found"):
+                await session.get_messages()
+
+        # Client should be cleaned up
+        assert client.get_state() == "disconnected"
+
+    async def test_multiple_sequential_session_contexts(self):
+        """Test creating multiple sessions sequentially with context managers."""
+        async with CopilotClient({"cli_path": CLI_PATH}) as client:
+            # First session
+            async with await client.create_session() as session1:
+                session1_id = session1.session_id
+                await session1.send({"prompt": "First session"})
+
+            # Second session (after first is cleaned up)
+            async with await client.create_session() as session2:
+                session2_id = session2.session_id
+                await session2.send({"prompt": "Second session"})
+
+            # Both sessions should be different
+            assert session1_id != session2_id
+
+            # First session should be destroyed
+            with pytest.raises(Exception, match="Session not found"):
+                await session1.get_messages()