diff --git a/README.md b/README.md
index fd6a0a7..09d57bb 100644
--- a/README.md
+++ b/README.md
@@ -1,376 +1,566 @@
 # Factory Droid SDK for Python
 
-A Python asyncio SDK for communicating with the [Factory](https://factory.ai) Droid agent via JSON-RPC 2.0 over a subprocess (`droid exec`).
+Run Factory Droid from Python through a local `droid exec` subprocess. The SDK
+uses `asyncio` and streams typed events over JSON-RPC 2.0.
 
-## Requirements
+## Choose an API
+
+| Goal | API |
+| --- | --- |
+| Run one prompt | `query()` |
+| Keep context across prompts | `DroidClient` |
+| Resume a saved session | `DroidClient.load_session()` |
+| Control models, tools, inputs, or lifecycle | `DroidClient` |
+
+`query()` owns one client subprocess and closes it after the turn. Use
+`DroidClient` for everything else.
+
+## Install and authenticate
+
+Requirements:
 
 - Python 3.10+
-- `droid` CLI installed (available at `~/.local/bin/droid`)
+- `droid` on `PATH`
 
-## Installation
+Install the package:
 
 ```bash
 pip install droid-sdk
 ```
 
-Or with [uv](https://docs.astral.sh/uv/):
+Or use [uv](https://docs.astral.sh/uv/):
 
 ```bash
 uv add droid-sdk
 ```
 
-## Quick Start
+Set a Factory API key before starting Python:
+
+```bash
+export FACTORY_API_KEY="your-key"
+```
+
+The `droid` subprocess inherits the Python process environment. You can also
+use an existing authenticated Droid CLI installation. Do not commit API keys or
+pass them in command-line arguments.
 
-The simplest way to use the SDK is with the `query()` convenience function, which handles the full session lifecycle automatically:
+## Quick start
 
 ```python
 import asyncio
-from droid_sdk import query, DroidQueryOptions
-from droid_sdk.stream import AssistantTextDelta, TurnComplete
 
-async def main():
-    async for msg in query("Explain this codebase", cwd="/path/to/project"):
-        if isinstance(msg, AssistantTextDelta):
-            print(msg.text, end="", flush=True)
-        elif isinstance(msg, TurnComplete):
-            print("\nDone!")
+from droid_sdk import AssistantTextDelta, ErrorEvent, query
+
+
+async def main() -> None:
+    error: str | None = None
+    async for event in query(
+        "Summarize this repository.",
+        cwd=".",
+        model_id="auto",
+    ):
+        if isinstance(event, AssistantTextDelta):
+            print(event.text, end="", flush=True)
+        elif isinstance(event, ErrorEvent):
+            error = event.message
+    print()
+    if error is not None:
+        raise RuntimeError(error)
+
 
 asyncio.run(main())
 ```
 
-You can also pass a `DroidQueryOptions` object for more control:
+`query()` yields text, thinking, tool, state, token, error, and completion
+events. It does not return a final result object.
 
-```python
-async def main():
-    options = DroidQueryOptions(
-        cwd="/path/to/project",
-        model_id="claude-sonnet-4",
-        reasoning_effort=ReasoningEffort.High,
-    )
-    async for msg in query("Fix the bug in main.py", options=options):
-        if isinstance(msg, AssistantTextDelta):
-            print(msg.text, end="", flush=True)
-```
+Runnable example: [`examples/query.py`](examples/query.py)
 
-### Using DroidClient directly
+## Sessions
 
-For more control over the session lifecycle, use `DroidClient` directly with `receive_response()`:
+Use one client for one active session.
 
 ```python
 import asyncio
-from droid_sdk import (
-    DroidClient,
-    ProcessTransport,
-    AssistantTextDelta,
-    ThinkingTextDelta,
-    ToolUse,
-    ToolResult,
-    TurnComplete,
-)
+import contextlib
+from pathlib import Path
+
+from droid_sdk import AssistantTextDelta, DroidClient, ErrorEvent
 
-async def main():
-    # Create a transport that spawns a droid exec subprocess
-    transport = ProcessTransport(exec_path="droid", cwd="/path/to/project")
 
-    # Use as an async context manager for automatic cleanup
-    async with DroidClient(transport=transport) as client:
-        # Initialize a new session
+async def receive(client: DroidClient) -> None:
+    error: str | None = None
+    async for event in client.receive_response():
+        if isinstance(event, AssistantTextDelta):
+            print(event.text, end="", flush=True)
+        elif isinstance(event, ErrorEvent):
+            error = event.message
+    print()
+    if error is not None:
+        raise RuntimeError(error)
+
+
+async def send(client: DroidClient, prompt: str) -> None:
+    response = asyncio.create_task(receive(client))
+    await asyncio.sleep(0)
+    try:
+        await client.add_user_message(text=prompt)
+        await response
+    finally:
+        if not response.done():
+            response.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await response
+
+
+async def main() -> None:
+    cwd = str(Path.cwd())
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
         result = await client.initialize_session(
-            machine_id="my-machine",
-            cwd="/path/to/project",
+            machine_id="my-app",
+            cwd=cwd,
+            model_id="auto",
         )
-        print(f"Session ID: {result.session_id}")
-
-        # Send a message and stream the response
-        await client.add_user_message(text="Hello, Droid!")
-
-        async for msg in client.receive_response():
-            if isinstance(msg, AssistantTextDelta):
-                print(msg.text, end="", flush=True)
-            elif isinstance(msg, ThinkingTextDelta):
-                print(f"[thinking] {msg.text}")
-            elif isinstance(msg, ToolUse):
-                print(f"\n🔧 Using tool: {msg.tool_name}")
-            elif isinstance(msg, ToolResult):
-                print(f"   Result: {msg.content}")
-            elif isinstance(msg, TurnComplete):
-                if msg.token_usage:
-                    print(f"\nTokens: {msg.token_usage.input_tokens} in / {msg.token_usage.output_tokens} out")
-                print("Done!")
-
-    # Transport and subprocess are cleaned up automatically
+        print(result.session_id)
+
+        await send(client, "What does this repository do?")
+        await send(client, "What should I test first?")
+
 
 asyncio.run(main())
 ```
 
-## Event Handling
+The second turn includes context from the first. The context manager connects
+the client and always closes its subprocess.
+
+Runnable example:
+[`examples/multi_turn_session.py`](examples/multi_turn_session.py)
 
-Register listeners for real-time notifications from the droid process:
+### Resume a session
 
 ```python
-from droid_sdk import (
-    DroidClient,
-    ProcessTransport,
-    SessionNotificationType,
+async with DroidClient(exec_path="droid") as client:
+    result = await client.load_session(session_id="session-id")
+    print(result.settings.model_id)
+    await send(client, "Continue the previous task.")
+```
+
+`load_session()` restores the saved conversation, working directory, and
+settings. It raises `SessionNotFoundError` when the ID does not exist.
+
+Runnable example:
+[`examples/resume_session.py`](examples/resume_session.py)
+
+### Update settings
+
+```python
+from droid_sdk.schemas import AutonomyLevel, ReasoningEffort
+
+await client.update_session_settings(
+    model_id="auto",
+    reasoning_effort=ReasoningEffort.High,
+    autonomy_level=AutonomyLevel.Low,
 )
+```
 
-async def main():
-    transport = ProcessTransport(exec_path="droid", cwd="/path/to/project")
-    async with DroidClient(transport=transport) as client:
-        # Listen for all notifications
-        def on_notification(notification):
-            params = notification.get("params", {})
-            inner = params.get("notification", {})
-            print(f"Notification type: {inner.get('type')}")
-
-        client.on_notification(on_notification)
-
-        # Or filter by notification type
-        def on_text_delta(notification):
-            params = notification["params"]["notification"]
-            print(params.get("textDelta", ""), end="", flush=True)
-
-        client.on_notification(
-            on_text_delta,
-            notification_type=SessionNotificationType.ASSISTANT_TEXT_DELTA,
-        )
+Updates apply to later turns in the active session.
 
-        result = await client.initialize_session(
-            machine_id="my-machine",
-            cwd="/path/to/project",
-        )
-        await client.add_user_message(text="Explain this codebase")
+### Session lifecycle
+
+```python
+await client.rename_session(title="Authentication review")
 
-        # Keep running to receive streamed notifications
-        import asyncio
-        await asyncio.sleep(60)
+fork = await client.fork_session(title="Alternative approach")
+await client.load_session(session_id=fork.new_session_id)
+
+compacted = await client.compact_session()
+await client.load_session(session_id=compacted.new_session_id)
 ```
 
-## Stream Type Checking
+`fork_session()`, `compact_session()`, and `execute_rewind()` return a new
+session ID. They do not switch the client to that session. Call
+`load_session()` to continue the new session.
+
+Runnable example:
+[`examples/session_lifecycle.py`](examples/session_lifecycle.py)
+
+## Models
+
+Model IDs depend on the account and organization policy. Omit `model_id` to use
+the Droid default.
 
-All stream message types are simple dataclasses that can be used with `isinstance()` for type-safe message handling:
+### Discover models
+
+`initialize_session()` and `load_session()` return `available_models` when the
+CLI provides a model catalog:
 
 ```python
-from droid_sdk import (
-    AssistantTextDelta,
-    ThinkingTextDelta,
-    ToolUse,
-    ToolResult,
-    ToolProgress,
-    WorkingStateChanged,
-    TokenUsageUpdate,
-    TurnComplete,
-    ErrorEvent,
-    StreamMessage,
-)
+result = await client.initialize_session(machine_id="my-app", cwd=".")
 
-def handle_message(msg: StreamMessage) -> None:
-    """Handle a stream message with exhaustive type checking."""
-    if isinstance(msg, AssistantTextDelta):
-        print(msg.text, end="", flush=True)
-    elif isinstance(msg, ThinkingTextDelta):
-        print(f"[thinking] {msg.text}")
-    elif isinstance(msg, ToolUse):
-        print(f"Tool call: {msg.tool_name}({msg.tool_input})")
-    elif isinstance(msg, ToolResult):
-        status = "❌" if msg.is_error else "✅"
-        # tool_use_id correlates the result with its ToolUse; tool_name is
-        # backfilled from that call (None if the call was never seen).
-        print(f"{status} [{msg.tool_use_id}] {msg.tool_name}: {msg.content}")
-    elif isinstance(msg, ToolProgress):
-        print(f"  ⏳ {msg.tool_name}: {msg.content}")
-    elif isinstance(msg, WorkingStateChanged):
-        print(f"State: {msg.state.value}")
-    elif isinstance(msg, TokenUsageUpdate):
-        print(f"Tokens: {msg.input_tokens} in / {msg.output_tokens} out")
-    elif isinstance(msg, TurnComplete):
-        print("\n--- Turn complete ---")
-    elif isinstance(msg, ErrorEvent):
-        # error_type is often the unhelpful "Error"; error_name exposes the
-        # nested error.name (e.g. "LLMInvalidRequestError") to branch on.
-        print(f"Error [{msg.error_name or msg.error_type}]: {msg.message}")
+for model in result.available_models or []:
+    print(model.id, model.supported_reasoning_efforts)
 ```
 
-## Permission Handler
+The server may add fields that are not yet typed. Read them from
+`model.model_extra`. Check `disabled` before presenting a model as selectable,
+and use `disabledReason` when it is present.
+
+Runnable example:
+[`examples/model_discovery.py`](examples/model_discovery.py)
 
-Handle permission requests when Droid needs approval to execute tools:
+### Select a model
 
 ```python
-from droid_sdk import DroidClient, ProcessTransport, ToolConfirmationOutcome
+from droid_sdk.schemas import ReasoningEffort
 
-async def main():
-    transport = ProcessTransport(exec_path="droid", cwd="/path/to/project")
-    async with DroidClient(transport=transport) as client:
+result = await client.initialize_session(
+    machine_id="my-app",
+    cwd=".",
+    model_id="model-id",
+    reasoning_effort=ReasoningEffort.High,
+)
+```
 
-        def handle_permission(params):
-            tool_uses = params.get("toolUses", [])
-            for tool in tool_uses:
-                tool_use = tool.get("toolUse", {})
-                print(f"Permission requested for: {tool_use.get('name')}")
-            # Approve the action
-            return ToolConfirmationOutcome.ProceedOnce.value
+Use a reasoning effort listed in the model's
+`supported_reasoning_efforts`. Otherwise, omit it to use the model default.
 
-        client.set_permission_handler(handle_permission)
+Runnable example:
+[`examples/model_selection.py`](examples/model_selection.py)
 
-        result = await client.initialize_session(
-            machine_id="my-machine",
-            cwd="/path/to/project",
-        )
-        await client.add_user_message(text="Create a hello.py file")
+### Auto Router
+
+Set `model_id="auto"` to let Factory choose the model:
+
+```python
+async for event in query("Find the failing test.", model_id="auto"):
+    ...
 ```
 
-## Error Handling
+The selected model can change between turns. Use a fixed model ID when every
+turn must use the same model.
+
+### Spec mode model
+
+```python
+from droid_sdk.schemas import DroidInteractionMode, ReasoningEffort
+
+result = await client.initialize_session(
+    machine_id="my-app",
+    cwd=".",
+    interaction_mode=DroidInteractionMode.Spec,
+    spec_mode_model_id="model-id",
+    spec_mode_reasoning_effort=ReasoningEffort.High,
+)
+```
+
+Use `update_session_settings()` to change interaction mode or spec-mode model
+settings later. The Python SDK does not yet provide `enter_spec_mode()` or
+`exit_spec_mode()` helpers.
+
+### Custom models
+
+Configure custom models in Droid, then pass the configured ID as `model_id`.
+Custom IDs use `custom:<model>`. See
+[Custom Models (BYOK)](https://docs.factory.ai/cli/byok/overview).
+
+## Streaming and errors
+
+`receive_response()` yields:
+
+| Event | Meaning |
+| --- | --- |
+| `AssistantTextDelta` | Assistant text |
+| `ThinkingTextDelta` | Reasoning text |
+| `ToolUse` | Tool call |
+| `ToolProgress` | Tool progress |
+| `ToolResult` | Tool result |
+| `WorkingStateChanged` | Agent state |
+| `TokenUsageUpdate` | Cumulative session usage |
+| `ErrorEvent` | Turn error |
+| `TurnComplete` | Turn finished |
+
+Use `isinstance()` to narrow events:
 
-The SDK provides a typed error hierarchy:
+```python
+from droid_sdk import AssistantTextDelta, ErrorEvent, TurnComplete
+
+error: str | None = None
+async for event in client.receive_response():
+    if isinstance(event, AssistantTextDelta):
+        print(event.text, end="", flush=True)
+    elif isinstance(event, ErrorEvent):
+        error = event.message
+    elif isinstance(event, TurnComplete) and event.token_usage:
+        print(event.token_usage.input_tokens)
+
+if error is not None:
+    raise RuntimeError(error)
+```
+
+An `ErrorEvent` reports a problem during the turn. SDK setup, transport, and
+protocol failures raise exceptions:
 
 ```python
 from droid_sdk import (
-    DroidClient,
+    ConnectionError as DroidConnectionError,
     DroidClientError,
-    ConnectionError,
-    TimeoutError,
+    ProcessExitError,
     ProtocolError,
-    SessionError,
     SessionNotFoundError,
-    ProcessExitError,
+    TimeoutError,
 )
-
-async def main():
-    # ... setup client ...
-    try:
-        result = await client.load_session(session_id="nonexistent")
-    except SessionNotFoundError as e:
-        print(f"Session not found: {e.session_id}")
-    except TimeoutError as e:
-        print(f"Request timed out after {e.timeout_duration}s")
-    except ConnectionError as e:
-        print(f"Connection failed: {e}")
-    except ProtocolError as e:
-        print(f"Protocol error (code={e.code}): {e.message}")
-    except DroidClientError as e:
-        print(f"SDK error: {e}")
 ```
 
-**Error hierarchy:**
+All SDK exceptions inherit from `DroidClientError`.
 
-- `DroidClientError` — base for all SDK errors
-  - `ConnectionError` — transport/connection failures
-  - `TimeoutError` — request timeout
-  - `ProtocolError` — JSON-RPC protocol errors
-  - `SessionError` — session-related errors
-    - `SessionNotFoundError` — session does not exist
-  - `ProcessExitError` — subprocess exited unexpectedly
+### Stop a turn
 
-## API Reference
+Call `interrupt_session()` from another task:
 
-### `DroidClient`
+```python
+async def consume_response() -> None:
+    async for _ in client.receive_response():
+        pass
+
+
+consumer = asyncio.create_task(consume_response())
+await asyncio.sleep(0)
+try:
+    await client.add_user_message(text="Perform a long review.")
+    await client.interrupt_session()
+    await consumer
+finally:
+    if not consumer.done():
+        consumer.cancel()
+        with contextlib.suppress(asyncio.CancelledError):
+            await consumer
+```
 
-The main client class. Wraps a transport and provides typed async methods for all `droid.*` RPC methods.
+Breaking out of `receive_response()` only stops the local iterator. It does not
+interrupt Droid. Call `interrupt_session()` first when work should stop.
 
-**Session methods:**
-- `initialize_session(...)` — Create a new session (supports `enabled_tool_ids` and `disabled_tool_ids`)
-- `load_session(session_id=...)` — Load an existing session
-- `add_user_message(text=..., output_format=...)` — Send a user message, optionally with a structured-output (JSON Schema) contract
-- `interrupt_session()` — Interrupt the current session
-- `kill_worker_session(worker_session_id=...)` — Kill a worker session
-- `update_session_settings(...)` — Update session settings (supports `enabled_tool_ids`/`disabled_tool_ids`)
-- `close_session(reason=...)` — Close the active session
-- `compact_session(custom_instructions=...)` — Compact the conversation to reclaim context
-- `fork_session(title=..., tags=...)` — Fork the session into a new one
-- `rename_session(title=...)` — Rename the session
+`query()` is an async generator. If you stop it early, close it explicitly:
 
-**Discovery methods:**
-- `list_tools(...)` — List native CLI tools with `default_allowed`/`currently_allowed` (useful for locking the tool set down)
-- `list_commands()` — List custom slash commands
+```python
+stream = query("Inspect the repository.")
+try:
+    async for event in stream:
+        break
+finally:
+    await stream.aclose()
+```
 
-**Context and rewind methods:**
-- `get_context_stats()` — Context-window usage (used/remaining/limit)
-- `get_context_breakdown()` — Per-category/skill/MCP/droid token breakdown
-- `get_rewind_info(message_id=...)` — Restorable/created/evicted files for a rewind point
-- `execute_rewind(...)` — Rewind to a message, forking the session
+## Inputs and structured output
 
-**Locking the tool set down:**
+Inputs are available through `DroidClient.add_user_message()`.
+
+### Images and documents
 
 ```python
-# enabled_tool_ids is additive, so pass an explicit disable list to
-# actually restrict native tools. list_tools() lets you verify the result.
-catalog = await client.list_tools()
-tool_ids = [t.id for t in catalog.tools]
-await client.update_session_settings(enabled_tool_ids=[], disabled_tool_ids=tool_ids)
+await client.add_user_message(
+    text="Summarize the attachments.",
+    images=[
+        {
+            "type": "base64",
+            "data": base64_png,
+            "mediaType": "image/png",
+        }
+    ],
+    files=[
+        {
+            "type": "text",
+            "mediaType": "text/plain",
+            "data": report,
+            "name": "report.txt",
+        }
+    ],
+)
 ```
 
-**Structured output:**
+Supported image types are JPEG, PNG, GIF, and WebP. PDF data must be
+base64-encoded with `mediaType` set to `application/pdf`.
+
+Runnable example:
+[`examples/attachment.py`](examples/attachment.py)
+
+### Structured output
 
 ```python
 await client.add_user_message(
-    text="Return an answer.",
+    text="Return the repository name.",
     output_format={
         "type": "json_schema",
         "schema": {
             "type": "object",
-            "properties": {"answer": {"type": "integer"}},
-            "required": ["answer"],
+            "properties": {"name": {"type": "string"}},
+            "required": ["name"],
+            "additionalProperties": False,
         },
     },
 )
 ```
 
-**MCP methods:**
-- `toggle_mcp_server(...)` — Enable/disable an MCP server
-- `authenticate_mcp_server(...)` — Authenticate an MCP server (OAuth)
-- `cancel_mcp_auth(...)` / `clear_mcp_auth(...)` — Cancel/clear MCP auth
-- `submit_mcp_auth_code(...)` — Submit an MCP auth code
-- `add_mcp_server(...)` / `remove_mcp_server(...)` — Add/remove MCP servers
-- `list_mcp_registry()` / `list_mcp_tools()` / `list_mcp_servers()` — List MCP resources
-- `toggle_mcp_tool(...)` — Enable/disable an MCP tool
+The response still arrives as streamed text. Collect the text, then parse it
+with `json.loads()`.
+
+Runnable example:
+[`examples/structured_output.py`](examples/structured_output.py)
+
+## Permissions and user input
+
+Without handlers, the SDK rejects permission requests and declines AskUser
+questions.
+
+### Permission handler
 
-**Other methods:**
-- `list_skills()` — List available skills
-- `submit_bug_report(...)` — Submit a bug report
+```python
+from typing import Any
+
+from droid_sdk import ToolConfirmationOutcome
+
+
+def handle_permission(params: dict[str, Any]) -> str:
+    offered = {option["value"] for option in params["options"]}
+    desired = ToolConfirmationOutcome.ProceedOnce.value
+    if desired in offered:
+        return desired
+    cancel = ToolConfirmationOutcome.Cancel.value
+    if cancel in offered:
+        return cancel
+    raise ValueError("No supported permission outcome was offered.")
 
-**Event system:**
-- `on_notification(callback, notification_type=None)` — Register a notification listener
-- `set_permission_handler(handler)` / `clear_permission_handler()` — Permission handling
-- `set_ask_user_handler(handler)` / `clear_ask_user_handler()` — Ask-user handling
 
-**Lifecycle:**
-- `connect()` / `close()` — Manual connection management
-- `async with DroidClient(...) as client:` — Context manager (recommended)
+client.set_permission_handler(handle_permission)
+```
+
+Return only an outcome present in `params["options"]`. Handlers may be
+synchronous or asynchronous.
 
-### `ProcessTransport`
+Runnable example:
+[`examples/permission_handler.py`](examples/permission_handler.py)
 
-Spawns a `droid exec` subprocess and manages JSONL communication over stdin/stdout.
+### AskUser handler
 
-### `DroidClientTransport`
+```python
+def handle_ask_user(params: dict[str, Any]) -> dict[str, Any]:
+    answers = [
+        {
+            "index": question["index"],
+            "question": question["question"],
+            "answer": (question["options"] or ["none"])[0],
+        }
+        for question in params["questions"]
+    ]
+    return {"cancelled": False, "answers": answers}
+
+
+client.set_ask_user_handler(handle_ask_user)
+```
 
-Protocol (interface) that all transport implementations must satisfy. Use this to create custom transports for testing or alternative communication channels.
+To decline, return `{"cancelled": True, "answers": []}`.
+
+## Tools, skills, and MCP
+
+### Control native tools
+
+```python
+catalog = await client.list_tools()
+tool_ids = [tool.id for tool in catalog.tools]
+
+await client.initialize_session(
+    machine_id="my-app",
+    cwd=".",
+    enabled_tool_ids=[],
+    disabled_tool_ids=tool_ids,
+)
+```
+
+`enabled_tool_ids` is additive. To restrict tools, pass an explicit disable
+list. `list_tools()` can run before session initialization.
+
+### List skills
+
+```python
+skills = await client.list_skills()
+for skill in skills.skills:
+    print(skill.name, skill.enabled)
+```
+
+The Python SDK can list skills but cannot enable or disable them.
+
+### External MCP servers
+
+Pass MCP server configuration to `initialize_session()` or `load_session()`.
+After initialization, use:
+
+- `add_mcp_server()` and `remove_mcp_server()`
+- `toggle_mcp_server()` and `toggle_mcp_tool()`
+- `list_mcp_servers()`, `list_mcp_tools()`, and `list_mcp_registry()`
+- `authenticate_mcp_server()`, `submit_mcp_auth_code()`,
+  `cancel_mcp_auth()`, and `clear_mcp_auth()`
+
+The Python SDK does not yet provide in-process SDK MCP tools.
+
+## API reference
+
+### Top-level imports
+
+`droid_sdk` exports the main client, transport, query API, stream events, and
+errors. Protocol models and enums are under `droid_sdk.schemas`.
+
+### `DroidClient`
+
+| Method | Purpose |
+| --- | --- |
+| `initialize_session()` | Create a session |
+| `load_session()` | Resume a saved session |
+| `add_user_message()` | Start a turn |
+| `receive_response()` | Stream the turn |
+| `interrupt_session()` | Stop active work |
+| `update_session_settings()` | Change active-session settings |
+| `list_tools()` / `list_commands()` | Discover tools and commands |
+| `get_context_stats()` / `get_context_breakdown()` | Inspect context usage |
+| `rename_session()` | Rename the active session |
+| `fork_session()` | Fork the active session |
+| `compact_session()` | Compact into a new session |
+| `get_rewind_info()` / `execute_rewind()` | Inspect and perform rewind |
+| `close_session()` | End the active session |
+| `list_skills()` | List skills |
+| `on_notification()` | Subscribe to raw notifications |
+| `set_permission_handler()` | Handle tool approvals |
+| `set_ask_user_handler()` | Handle AskUser questions |
+| `connect()` / `close()` | Manage the subprocess connection |
+
+### TypeScript SDK differences
+
+The Python SDK currently exposes the subprocess JSON-RPC client. It does not
+yet expose the TypeScript SDK's:
+
+- daemon or browser client
+- local session listing
+- final `DroidResult` object
+- in-process SDK MCP tools
+- hook stream events
+- observability sinks
+- Factory REST helpers
 
 ## Development
 
 ```bash
-# Install dependencies
 uv sync
-
-# Run tests
 uv run --group dev python -m pytest
+uv run mypy --strict src/ examples/
+uv run ruff check src/ tests/ examples/
+uv run ruff format --check src/ tests/ examples/
+```
 
-# Run opt-in tests against the installed, authenticated droid exec CLI.
-# These create real sessions and consume model usage.
+Live tests create real sessions and consume model usage:
+
+```bash
 DROID_LIVE_TESTS=1 uv run --group dev python -m pytest \
   tests/test_live_droid_exec.py -v
-
-# Override the executable path when droid is not on PATH.
-DROID_LIVE_TESTS=1 DROID_EXEC_PATH=/path/to/droid \
-  uv run --group dev python -m pytest tests/test_live_droid_exec.py -v
-
-# Type check (strict mode)
-uv run mypy --strict src/
-
-# Lint and format
-uv run ruff check src/ tests/
-uv run ruff format --check src/ tests/
 ```
 
 ## License
 
-Apache 2.0 — see [LICENSE](LICENSE) for details.
+Apache 2.0. See [LICENSE](LICENSE).
diff --git a/examples/README.md b/examples/README.md
new file mode 100644
index 0000000..9379a46
--- /dev/null
+++ b/examples/README.md
@@ -0,0 +1,25 @@
+# Examples
+
+Run examples from the repository root:
+
+```bash
+uv sync
+export FACTORY_API_KEY="..."
+uv run python examples/query.py
+```
+
+| Example | Purpose |
+| --- | --- |
+| [`query.py`](query.py) | Run one prompt with Auto Router |
+| [`multi_turn_session.py`](multi_turn_session.py) | Preserve context across turns |
+| [`model_discovery.py`](model_discovery.py) | List available model IDs |
+| [`model_selection.py`](model_selection.py) | Select Auto Router or a fixed model |
+| [`resume_session.py`](resume_session.py) | Resume a saved session |
+| [`structured_output.py`](structured_output.py) | Request JSON Schema output |
+| [`permission_handler.py`](permission_handler.py) | Handle tool approval requests |
+| [`attachment.py`](attachment.py) | Send an image, PDF, or text file |
+| [`session_lifecycle.py`](session_lifecycle.py) | Fork and continue a session |
+| [`interactive_session.py`](interactive_session.py) | Run an interactive terminal session |
+
+Each example starts a local `droid exec` subprocess. It uses your current
+directory unless its usage says otherwise.
diff --git a/examples/_helpers.py b/examples/_helpers.py
new file mode 100644
index 0000000..24e2d46
--- /dev/null
+++ b/examples/_helpers.py
@@ -0,0 +1,85 @@
+"""Shared helpers for the runnable examples."""
+
+from __future__ import annotations
+
+import asyncio
+import contextlib
+from typing import Any
+
+from droid_sdk import (
+    AssistantTextDelta,
+    DroidClient,
+    ErrorEvent,
+    ThinkingTextDelta,
+    TokenUsageUpdate,
+    ToolProgress,
+    ToolResult,
+    ToolUse,
+    TurnComplete,
+)
+from droid_sdk.schemas import Base64ImageSource, DocumentSource
+
+ImageInput = Base64ImageSource | dict[str, Any]
+DocumentInput = DocumentSource | dict[str, Any]
+
+
+async def run_turn(
+    client: DroidClient,
+    prompt: str,
+    *,
+    images: list[ImageInput] | None = None,
+    files: list[DocumentInput] | None = None,
+    output_format: dict[str, Any] | None = None,
+    show_activity: bool = False,
+    collect_text: bool = False,
+) -> str | None:
+    """Send one prompt, print its stream, and optionally return its text."""
+
+    async def consume() -> str | None:
+        chunks: list[str] = []
+        error_message: str | None = None
+
+        async for event in client.receive_response():
+            if isinstance(event, AssistantTextDelta):
+                print(event.text, end="", flush=True)
+                if collect_text:
+                    chunks.append(event.text)
+            elif show_activity and isinstance(event, ThinkingTextDelta):
+                print(event.text, end="", flush=True)
+            elif show_activity and isinstance(event, ToolUse):
+                print(f"\n[tool] {event.tool_name}")
+            elif show_activity and isinstance(event, ToolProgress):
+                print(f"\n[progress] {event.tool_name}: {event.content}")
+            elif show_activity and isinstance(event, ToolResult):
+                status = "error" if event.is_error else "done"
+                print(f"\n[{status}] {event.tool_name or event.tool_use_id}")
+            elif show_activity and isinstance(event, TokenUsageUpdate):
+                print(
+                    f"\n[tokens] input={event.input_tokens} "
+                    f"output={event.output_tokens}"
+                )
+            elif isinstance(event, ErrorEvent):
+                error_message = event.message
+            elif isinstance(event, TurnComplete):
+                print()
+                if error_message is not None:
+                    raise RuntimeError(error_message)
+                return "".join(chunks) if collect_text else None
+
+        raise RuntimeError("Droid stopped before the turn completed.")
+
+    consumer = asyncio.create_task(consume())
+    await asyncio.sleep(0)
+    try:
+        await client.add_user_message(
+            text=prompt,
+            images=images,
+            files=files,
+            output_format=output_format,
+        )
+        return await consumer
+    finally:
+        if not consumer.done():
+            consumer.cancel()
+            with contextlib.suppress(asyncio.CancelledError):
+                await consumer
diff --git a/examples/attachment.py b/examples/attachment.py
new file mode 100644
index 0000000..9501828
--- /dev/null
+++ b/examples/attachment.py
@@ -0,0 +1,81 @@
+#!/usr/bin/env python3
+"""Send one image, PDF, or text file.
+
+Usage:
+    uv run python examples/attachment.py PATH
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+import base64
+from pathlib import Path
+
+from _helpers import DocumentInput, ImageInput, run_turn
+
+from droid_sdk import DroidClient
+
+_IMAGE_TYPES = {
+    ".gif": "image/gif",
+    ".jpeg": "image/jpeg",
+    ".jpg": "image/jpeg",
+    ".png": "image/png",
+    ".webp": "image/webp",
+}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("path", type=Path)
+    return parser.parse_args()
+
+
+def encode(path: Path) -> tuple[list[ImageInput], list[DocumentInput]]:
+    suffix = path.suffix.lower()
+    if suffix in _IMAGE_TYPES:
+        image = {
+            "type": "base64",
+            "data": base64.b64encode(path.read_bytes()).decode("ascii"),
+            "mediaType": _IMAGE_TYPES[suffix],
+        }
+        return [image], []
+
+    if suffix == ".pdf":
+        document = {
+            "type": "base64",
+            "mediaType": "application/pdf",
+            "data": base64.b64encode(path.read_bytes()).decode("ascii"),
+            "name": path.name,
+        }
+        return [], [document]
+
+    document = {
+        "type": "text",
+        "mediaType": "text/plain",
+        "data": path.read_text(),
+        "name": path.name,
+    }
+    return [], [document]
+
+
+async def main(path: Path) -> None:
+    images, files = encode(path)
+    cwd = str(Path.cwd())
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+        )
+        await run_turn(
+            client,
+            "Describe the attached file.",
+            images=images,
+            files=files,
+        )
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    asyncio.run(main(args.path))
diff --git a/examples/interactive_session.py b/examples/interactive_session.py
index eed251c..8f9d32b 100644
--- a/examples/interactive_session.py
+++ b/examples/interactive_session.py
@@ -1,433 +1,93 @@
 #!/usr/bin/env python3
-"""Interactive session example for the Factory Droid Python SDK.
-
-Demonstrates the SDK's full capabilities by rendering ALL intermediate activity
-from a Droid session with ANSI-colored output:
-
-  - Assistant text deltas (streamed char-by-char)
-  - Thinking text deltas (gray, with [Thinking] prefix)
-  - Tool calls (yellow: tool name + input summary)
-  - Tool results (green for success, red for errors)
-  - Tool progress updates
-  - Token usage updates
-  - Error events (red)
-  - Permission requests (auto-approve or prompt user)
-  - Turn completion with final token stats
-  - MCP status changes
-  - Session title updates
+"""Run an interactive Droid session.
 
 Usage:
-    python examples/interactive_session.py [--cwd /path/to/project]
-
-Requires the ``droid`` CLI to be installed and available on PATH
-(or at ``~/.local/bin/droid``).
-
-Simpler alternative using query():
-
-    import asyncio
-    from droid_sdk import query, AssistantTextDelta, TurnComplete
-
-    async def main() -> None:
-        async for msg in query("Hello!", cwd="."):
-            if isinstance(msg, AssistantTextDelta):
-                print(msg.text, end="", flush=True)
-            elif isinstance(msg, TurnComplete):
-                print()
-
-    asyncio.run(main())
+    uv run python examples/interactive_session.py [--cwd PATH] [--model MODEL_ID]
 """
 
 from __future__ import annotations
 
 import argparse
 import asyncio
-import json
-import os
-import sys
-
-from droid_sdk import (
-    AssistantTextDelta,
-    ConnectionError,
-    DroidClient,
-    DroidClientError,
-    ErrorEvent,
-    ProcessExitError,
-    ProcessTransport,
-    StreamMessage,
-    ThinkingTextDelta,
-    TimeoutError,
-    TokenUsageUpdate,
-    ToolConfirmationOutcome,
-    ToolProgress,
-    ToolResult,
-    ToolUse,
-    TurnComplete,
-    WorkingStateChanged,
-)
-
-# ---------------------------------------------------------------------------
-# ANSI color helpers
-# ---------------------------------------------------------------------------
-
-# Standard ANSI escape codes
-_RESET = "\033[0m"
-_BOLD = "\033[1m"
-_DIM = "\033[2m"
-_RED = "\033[31m"
-_GREEN = "\033[32m"
-_YELLOW = "\033[33m"
-_BLUE = "\033[34m"
-_MAGENTA = "\033[35m"
-_CYAN = "\033[36m"
-_GRAY = "\033[90m"
-
-
-def _colored(text: str, *codes: str) -> str:
-    """Wrap text in ANSI escape codes."""
-    prefix = "".join(codes)
-    return f"{prefix}{text}{_RESET}"
-
-
-# ---------------------------------------------------------------------------
-# Tracking state for deduplication and clean output
-# ---------------------------------------------------------------------------
-
-# Track whether we're in the middle of streaming text (to manage newlines)
-_streaming_text = False
-
-# Track the last MCP summary string to deduplicate notifications
-_last_mcp_summary: str | None = None
-
-# Track the last token usage to only print at turn end
-_last_token_usage: TokenUsageUpdate | None = None
-
-
-def _ensure_newline() -> None:
-    """Print a newline if we were streaming text inline."""
-    global _streaming_text
-    if _streaming_text:
-        print()
-        _streaming_text = False
-
-
-# ---------------------------------------------------------------------------
-# Message rendering
-# ---------------------------------------------------------------------------
-
-
-def render_message(msg: StreamMessage) -> None:
-    """Render a single StreamMessage with ANSI colors."""
-    global _streaming_text, _last_token_usage
-
-    if isinstance(msg, AssistantTextDelta):
-        # Stream assistant text char-by-char (inline, no newline)
-        print(msg.text, end="", flush=True)
-        _streaming_text = True
-
-    elif isinstance(msg, ThinkingTextDelta):
-        # Thinking text in gray with prefix (inline, no newline)
-        if not _streaming_text:
-            print(_colored("[thinking] ", _GRAY, _DIM), end="", flush=True)
-            _streaming_text = True
-        print(_colored(msg.text, _GRAY, _DIM), end="", flush=True)
-
-    elif isinstance(msg, ToolUse):
-        _ensure_newline()
-        print()  # blank line before tool call
-        input_summary = json.dumps(msg.tool_input, ensure_ascii=False)
-        if len(input_summary) > 120:
-            input_summary = input_summary[:117] + "..."
-        print(_colored("[tool] ", _YELLOW, _BOLD) + _colored(msg.tool_name, _YELLOW))
-        print(_colored(f"  {input_summary}", _DIM))
-
-    elif isinstance(msg, ToolResult):
-        _ensure_newline()
-        content_preview = _content_preview(msg.content)
-        if msg.is_error:
-            tag = "[error] "
-            name = msg.tool_name or "unknown"
-            print(_colored(tag, _RED, _BOLD) + _colored(f"{name} (error)", _RED))
-            print(_colored(f"  {content_preview}", _RED))
-        else:
-            tag = "[result] "
-            name = msg.tool_name or "unknown"
-            print(_colored(tag, _GREEN, _BOLD) + _colored(f"{name} (success)", _GREEN))
-            print(_colored(f"  {content_preview}", _DIM))
-        print()  # blank line after tool result
-
-    elif isinstance(msg, ToolProgress):
-        _ensure_newline()
-        print(_colored("[progress] ", _CYAN, _BOLD) + _colored(msg.tool_name, _CYAN))
-        print(_colored(f"  {msg.content}", _CYAN))
-
-    elif isinstance(msg, WorkingStateChanged):
-        # Suppress state changes entirely — tool calls and results already
-        # indicate what's happening, and "Streaming Assistant Message" is
-        # obvious from the text appearing on screen.
-        pass
-
-    elif isinstance(msg, TokenUsageUpdate):
-        # Stash the latest token usage; we only print it at turn end.
-        _last_token_usage = msg
-
-    elif isinstance(msg, ErrorEvent):
-        _ensure_newline()
-        print()
-        print(_colored(f"[error] {msg.error_type}: {msg.message}", _RED, _BOLD))
-        print()
-
-    elif isinstance(msg, TurnComplete):
-        _ensure_newline()
-        print()
-        # Print token usage (from stashed update or TurnComplete payload)
-        tu = msg.token_usage
-        if tu is not None:
-            total = tu.input_tokens + tu.output_tokens
-            print(
-                _colored(
-                    f"[tokens] {total:,} total "
-                    f"(in={tu.input_tokens:,}, out={tu.output_tokens:,}, "
-                    f"cache_read={tu.cache_read_tokens:,}, "
-                    f"cache_write={tu.cache_write_tokens:,})",
-                    _DIM,
-                )
-            )
-        elif _last_token_usage is not None:
-            t = _last_token_usage
-            total = t.input_tokens + t.output_tokens
-            print(
-                _colored(
-                    f"[tokens] {total:,} total "
-                    f"(in={t.input_tokens:,}, out={t.output_tokens:,}, "
-                    f"cache_read={t.cache_read_tokens:,}, "
-                    f"cache_write={t.cache_write_tokens:,})",
-                    _DIM,
-                )
-            )
-        print(_colored("[done] Turn complete", _GREEN, _BOLD))
-        print()  # blank line before next prompt
-        # Reset stashed token usage for next turn
-        _last_token_usage = None
-
-    else:
-        # Future-proof: render unknown message types
-        _ensure_newline()
-        print(_colored(f"  [?] {type(msg).__name__}: {msg}", _DIM))
+import contextlib
+from pathlib import Path
+from typing import Any
 
+from _helpers import run_turn
 
-def _content_preview(content: str | list[object], max_len: int = 100) -> str:
-    """Create a truncated preview of tool result content."""
-    if isinstance(content, list):
-        text = json.dumps(content, ensure_ascii=False)
-    else:
-        text = str(content)
-    # Collapse whitespace for display
-    text = " ".join(text.split())
-    if len(text) > max_len:
-        return text[: max_len - 3] + "..."
-    return text
+from droid_sdk import DroidClient, ToolConfirmationOutcome
 
 
-# ---------------------------------------------------------------------------
-# Notification listener for events not in receive_response()
-# ---------------------------------------------------------------------------
-
-
-def on_raw_notification(notification: dict[str, object]) -> None:
-    """Handle notifications not covered by receive_response().
-
-    Renders MCP status changes, session title updates, and other
-    notification types that are not mapped to StreamMessage.
-    """
-    global _last_mcp_summary
-
-    params = notification.get("params")
-    if not isinstance(params, dict):
-        return
-    inner = params.get("notification")
-    if not isinstance(inner, dict):
-        return
-
-    notif_type = inner.get("type")
-
-    if notif_type == "mcp_status_changed":
-        servers = inner.get("servers", [])
-        summary = inner.get("summary", {})
-        connected = summary.get("connected", 0)
-        total = summary.get("total", 0)
-
-        # Build a summary string to detect duplicates
-        server_lines: list[str] = []
-        if isinstance(servers, list):
-            for srv in servers:
-                if isinstance(srv, dict):
-                    name = srv.get("name", "?")
-                    status = srv.get("status", "?")
-                    server_lines.append(f"  {name}: {status}")
-        current_summary = f"{connected}/{total}|" + "|".join(server_lines)
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--cwd", type=Path, default=Path.cwd())
+    parser.add_argument("--model")
+    parser.add_argument("--exec-path", default="droid")
+    return parser.parse_args()
 
-        # Only print if the summary actually changed
-        if current_summary == _last_mcp_summary:
-            return
-        _last_mcp_summary = current_summary
 
-        print()
-        print(
-            _colored(
-                f"[mcp] servers ({connected}/{total} connected):",
-                _MAGENTA,
-            )
+async def ask(prompt: str) -> str:
+    return await asyncio.to_thread(input, prompt)
+
+
+async def handle_permission(params: dict[str, Any]) -> str:
+    offered = {
+        option.get("value"): option.get("label", option.get("value"))
+        for option in params.get("options", [])
+    }
+    tool_names = [
+        item.get("toolUse", {}).get("name", "unknown")
+        for item in params.get("toolUses", [])
+    ]
+    answer = (await ask(f"Allow {', '.join(tool_names)} once? [y/N] ")).lower()
+    desired = ToolConfirmationOutcome.ProceedOnce.value
+    if answer == "y" and desired in offered:
+        return desired
+    cancel = ToolConfirmationOutcome.Cancel.value
+    if cancel in offered:
+        return cancel
+    raise ValueError("No supported permission outcome was offered.")
+
+
+async def handle_ask_user(params: dict[str, Any]) -> dict[str, Any]:
+    answers = []
+    for question in params.get("questions", []):
+        options = ", ".join(question.get("options", []))
+        answer = await ask(f"{question['question']} [{options}] ")
+        answers.append(
+            {
+                "index": question["index"],
+                "question": question["question"],
+                "answer": answer,
+            }
         )
-        for line in server_lines:
-            print(_colored(line, _DIM))
-
-    elif notif_type == "session_title_updated":
-        title = inner.get("title", "")
-        print(_colored(f"[title] {title}", _BLUE, _BOLD))
-
-    elif notif_type == "mcp_auth_required":
-        server_name = inner.get("serverName", "?")
-        auth_url = inner.get("authUrl", "?")
-        print(_colored(f"[mcp] auth required for {server_name}", _YELLOW, _BOLD))
-        print(_colored(f"  URL: {auth_url}", _DIM))
-
-    elif notif_type == "mcp_auth_completed":
-        server_name = inner.get("serverName", "?")
-        outcome = inner.get("outcome", "?")
-        print(_colored(f"[mcp] auth {outcome} for {server_name}", _YELLOW))
-
-    elif notif_type == "mission_state_changed":
-        state = inner.get("state", "?")
-        print(_colored(f"[state] mission: {state}", _MAGENTA, _BOLD))
-
-    elif notif_type == "permission_resolved":
-        option = inner.get("selectedOption", "?")
-        print(_colored(f"[state] permission resolved: {option}", _DIM))
-
-
-# ---------------------------------------------------------------------------
-# Main session loop
-# ---------------------------------------------------------------------------
-
+    return {"cancelled": False, "answers": answers}
 
-async def main(cwd: str, exec_path: str) -> None:
-    """Run an interactive Droid session with full activity rendering."""
-    transport = ProcessTransport(exec_path=exec_path, cwd=cwd)
 
-    async with DroidClient(transport=transport) as client:
-        # Register raw notification listener for events outside receive_response()
-        client.on_notification(on_raw_notification)
-
-        # Register permission handler (auto-approve all)
-        def handle_permission(params: dict[str, object]) -> str:
-            """Auto-approve all permission requests."""
-            tool_uses = params.get("toolUses")
-            if isinstance(tool_uses, list):
-                for tool in tool_uses:
-                    if isinstance(tool, dict):
-                        tool_use = tool.get("toolUse")
-                        if isinstance(tool_use, dict):
-                            name = tool_use.get("name", "unknown")
-                            print(
-                                _colored(
-                                    f"[state] auto-approving: {name}",
-                                    _YELLOW,
-                                    _BOLD,
-                                )
-                            )
-            return ToolConfirmationOutcome.ProceedOnce.value
+async def main(cwd: Path, model_id: str | None, exec_path: str) -> None:
+    resolved_cwd = str(cwd.resolve())
 
+    async with DroidClient(exec_path=exec_path, cwd=resolved_cwd) as client:
         client.set_permission_handler(handle_permission)
+        client.set_ask_user_handler(handle_ask_user)
 
-        # Initialize session
-        print(f"Executable: {exec_path}")
-        print(f"Working directory: {cwd}")
-        print()
-
-        try:
-            result = await client.initialize_session(
-                machine_id="interactive-example",
-                cwd=cwd,
-            )
-            print(f"Session ID: {result.session_id}")
-            print(f"Model: {result.settings.model_id}")
-            if result.available_models:
-                print(f"Available models: {len(result.available_models)}")
-        except TimeoutError:
-            print(_colored("ERROR: Session initialization timed out.", _RED, _BOLD))
-            return
-        except ConnectionError as e:
-            print(_colored(f"ERROR: Could not connect: {e}", _RED, _BOLD))
-            return
-
-        # Brief pause to let MCP notifications arrive before first prompt
-        await asyncio.sleep(1.0)
-
-        # Interactive loop
-        print()
-        print("Type your messages below. Press Ctrl+C or type 'exit' to quit.")
+        result = await client.initialize_session(
+            machine_id="python-sdk-interactive-example",
+            cwd=resolved_cwd,
+            model_id=model_id,
+        )
+        print(f"Session: {result.session_id}")
+        print("Enter a prompt. Type exit to stop.")
 
         while True:
-            try:
-                # Small delay for pending notification flush
-                await asyncio.sleep(0.1)
-                user_input = await asyncio.get_event_loop().run_in_executor(
-                    None, lambda: input(_colored("\n> ", _CYAN, _BOLD))
-                )
-            except (EOFError, KeyboardInterrupt):
-                print("\nGoodbye!")
-                break
-
-            user_input = user_input.strip()
-            if not user_input:
-                continue
-            if user_input.lower() in ("exit", "quit", ":q"):
-                print("Goodbye!")
-                break
-
-            try:
-                await client.add_user_message(text=user_input)
-
-                # Stream the response using receive_response()
-                async for msg in client.receive_response():
-                    render_message(msg)
-
-            except ProcessExitError as e:
-                print(_colored(f"\n[error] Droid process exited: {e}", _RED, _BOLD))
-                break
-            except DroidClientError as e:
-                print(_colored(f"\n[error] {e}", _RED, _BOLD))
-                break
-
-    print(_colored("\nSession closed.", _DIM))
-
-
-# ---------------------------------------------------------------------------
-# CLI entry point
-# ---------------------------------------------------------------------------
-
-
-def parse_args() -> argparse.Namespace:
-    """Parse command-line arguments."""
-    parser = argparse.ArgumentParser(
-        description="Interactive Factory Droid session with full activity rendering",
-    )
-    parser.add_argument(
-        "--cwd",
-        default=os.getcwd(),
-        help="Working directory for the droid session (default: current directory)",
-    )
-    parser.add_argument(
-        "--exec-path",
-        default="droid",
-        help="Path to the droid executable (default: 'droid')",
-    )
-    return parser.parse_args()
+            prompt = (await ask("> ")).strip()
+            if prompt.lower() in {"exit", "quit"}:
+                return
+            if prompt:
+                await run_turn(client, prompt, show_activity=True)
 
 
 if __name__ == "__main__":
     args = parse_args()
-    try:
-        asyncio.run(main(cwd=args.cwd, exec_path=args.exec_path))
-    except KeyboardInterrupt:
-        sys.exit(0)
+    with contextlib.suppress(EOFError, KeyboardInterrupt):
+        asyncio.run(main(args.cwd, args.model, args.exec_path))
diff --git a/examples/model_discovery.py b/examples/model_discovery.py
new file mode 100644
index 0000000..78534fd
--- /dev/null
+++ b/examples/model_discovery.py
@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+"""List models returned during session initialization.
+
+Usage:
+    uv run python examples/model_discovery.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from droid_sdk import DroidClient
+
+
+async def main() -> None:
+    cwd = str(Path.cwd())
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        result = await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+        )
+
+        for model in result.available_models or []:
+            extra = model.model_extra or {}
+            state = f"disabled: {extra.get('disabledReason', 'unavailable')}"
+            if not extra.get("disabled"):
+                state = "available"
+            efforts = ", ".join(
+                effort.value for effort in model.supported_reasoning_efforts
+            )
+            print(f"{model.id}\t{state}\t{efforts}")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/model_selection.py b/examples/model_selection.py
new file mode 100644
index 0000000..c457103
--- /dev/null
+++ b/examples/model_selection.py
@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""Run a prompt with Auto Router or a fixed model.
+
+Usage:
+    uv run python examples/model_selection.py
+    uv run python examples/model_selection.py --model MODEL_ID
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+from pathlib import Path
+
+from _helpers import run_turn
+
+from droid_sdk import DroidClient
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument(
+        "--model",
+        default="auto",
+        help="Model ID returned by model_discovery.py (default: auto)",
+    )
+    return parser.parse_args()
+
+
+async def main(model_id: str) -> None:
+    cwd = str(Path.cwd())
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        result = await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+            model_id=model_id,
+        )
+        print(f"Configured model: {result.settings.model_id}")
+        await run_turn(client, "Find the main entry point in this repository.")
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    asyncio.run(main(args.model))
diff --git a/examples/multi_turn_session.py b/examples/multi_turn_session.py
new file mode 100644
index 0000000..6f008cc
--- /dev/null
+++ b/examples/multi_turn_session.py
@@ -0,0 +1,34 @@
+#!/usr/bin/env python3
+"""Run two prompts in the same session.
+
+Usage:
+    uv run python examples/multi_turn_session.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from _helpers import run_turn
+
+from droid_sdk import DroidClient
+
+
+async def main() -> None:
+    cwd = str(Path.cwd())
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        result = await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+            model_id="auto",
+        )
+        print(f"Session: {result.session_id}")
+
+        await run_turn(client, "What does this repository do?")
+        await run_turn(client, "What should I test first?")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/permission_handler.py b/examples/permission_handler.py
new file mode 100644
index 0000000..bff2dc4
--- /dev/null
+++ b/examples/permission_handler.py
@@ -0,0 +1,49 @@
+#!/usr/bin/env python3
+"""Approve one offered tool request at a time.
+
+Usage:
+    uv run python examples/permission_handler.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+from typing import Any
+
+from _helpers import run_turn
+
+from droid_sdk import DroidClient, ToolConfirmationOutcome
+from droid_sdk.schemas import AutonomyLevel
+
+
+def handle_permission(params: dict[str, Any]) -> str:
+    offered = {option.get("value") for option in params.get("options", [])}
+    desired = ToolConfirmationOutcome.ProceedOnce.value
+    if desired in offered:
+        return desired
+    cancel = ToolConfirmationOutcome.Cancel.value
+    if cancel in offered:
+        return cancel
+    raise ValueError("No supported permission outcome was offered.")
+
+
+async def main() -> None:
+    cwd = str(Path.cwd())
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        client.set_permission_handler(handle_permission)
+        await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+            autonomy_level=AutonomyLevel.Off,
+        )
+        await run_turn(
+            client,
+            "List the top-level files. Do not modify anything.",
+            show_activity=True,
+        )
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/query.py b/examples/query.py
new file mode 100644
index 0000000..c8d7a88
--- /dev/null
+++ b/examples/query.py
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+"""Run one prompt with query().
+
+Usage:
+    uv run python examples/query.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from droid_sdk import AssistantTextDelta, ErrorEvent, query
+
+
+async def main() -> None:
+    error: str | None = None
+    async for event in query(
+        "Summarize this repository.",
+        cwd=str(Path.cwd()),
+        model_id="auto",
+    ):
+        if isinstance(event, AssistantTextDelta):
+            print(event.text, end="", flush=True)
+        elif isinstance(event, ErrorEvent):
+            error = event.message
+    print()
+    if error is not None:
+        raise RuntimeError(error)
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/resume_session.py b/examples/resume_session.py
new file mode 100644
index 0000000..640e1c0
--- /dev/null
+++ b/examples/resume_session.py
@@ -0,0 +1,33 @@
+#!/usr/bin/env python3
+"""Resume a saved session.
+
+Usage:
+    uv run python examples/resume_session.py SESSION_ID
+"""
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+
+from _helpers import run_turn
+
+from droid_sdk import DroidClient
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("session_id")
+    return parser.parse_args()
+
+
+async def main(session_id: str) -> None:
+    async with DroidClient(exec_path="droid") as client:
+        result = await client.load_session(session_id=session_id)
+        print(f"Model: {result.settings.model_id}")
+        await run_turn(client, "Continue from the previous conversation.")
+
+
+if __name__ == "__main__":
+    args = parse_args()
+    asyncio.run(main(args.session_id))
diff --git a/examples/session_lifecycle.py b/examples/session_lifecycle.py
new file mode 100644
index 0000000..c5cdbc5
--- /dev/null
+++ b/examples/session_lifecycle.py
@@ -0,0 +1,37 @@
+#!/usr/bin/env python3
+"""Fork a session, load the fork, and continue it.
+
+Usage:
+    uv run python examples/session_lifecycle.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+from pathlib import Path
+
+from _helpers import run_turn
+
+from droid_sdk import DroidClient
+
+
+async def main() -> None:
+    cwd = str(Path.cwd())
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        result = await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+        )
+        print(f"Source session: {result.session_id}")
+
+        await run_turn(client, "Remember the value ORANGE.")
+        fork = await client.fork_session(title="Python SDK fork example")
+        print(f"Forked session: {fork.new_session_id}")
+
+        await client.load_session(session_id=fork.new_session_id)
+        await run_turn(client, "What value did I ask you to remember?")
+
+
+if __name__ == "__main__":
+    asyncio.run(main())
diff --git a/examples/structured_output.py b/examples/structured_output.py
new file mode 100644
index 0000000..8937a0c
--- /dev/null
+++ b/examples/structured_output.py
@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""Request and parse JSON that follows a JSON Schema.
+
+Usage:
+    uv run python examples/structured_output.py
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+from pathlib import Path
+from typing import Any
+
+from _helpers import run_turn
+
+from droid_sdk import DroidClient
+
+
+async def main() -> None:
+    cwd = str(Path.cwd())
+    output_format: dict[str, Any] = {
+        "type": "json_schema",
+        "schema": {
+            "type": "object",
+            "properties": {
+                "name": {"type": "string"},
+                "language": {"type": "string"},
+            },
+            "required": ["name", "language"],
+            "additionalProperties": False,
+        },
+    }
+
+    async with DroidClient(exec_path="droid", cwd=cwd) as client:
+        await client.initialize_session(
+            machine_id="python-sdk-example",
+            cwd=cwd,
+            model_id="auto",
+        )
+        text = await run_turn(
+            client,
+            "Return this repository's name and primary language.",
+            output_format=output_format,
+            collect_text=True,
+        )
+
+    if text is None:
+        raise RuntimeError("No structured output was returned.")
+    value = json.loads(text)
+    print(json.dumps(value, indent=2))
+
+
+if __name__ == "__main__":
+    asyncio.run(main())