Droid Python SDK

The Droid SDK lets you run the same agent harness that powers Factory's CLI, desktop application, and web platform from your own code. It manages conversation context, tool execution, permissions, streaming, model selection, and multi-step execution.

Your application decides when Droid runs, which tools and models it can use, and what happens to the result.

Overview

The Python SDK runs a local droid subprocess and provides one-shot runs, persistent sessions, streaming, and session discovery.

Choose an API

Use Session when prompts need shared history or session operations such as interruption, compaction, or rewind. There is no session.run(); every session turn goes through stream().

What you can build

Invoke Droid when a pull request opens, a support ticket is escalated, an incident is created, or scheduled repository maintenance is due. The agent does not need a chat interface.

Install and authenticate

Install the droid-sdk package and import it as droid_sdk:

pip install droid-sdk

The package provides an async API for running Droid locally.

Requirements:

• Python 3.10 or later
• droid on PATH
• an authenticated Droid CLI session or a Factory API key

The SDK uses the Droid CLI's local authentication by default. It reads FACTORY_API_KEY when present; pass api_key= only when the key comes from application configuration. The SDK never places keys in command arguments, logs, or exception messages.

Quick start

Run one prompt

import asyncio
 
from droid_sdk import run
 
 
async def main() -> None:
    result = await run("Summarize this repository.")
    if result.success:
        print(result.text)
    else:
        print(result.subtype)
 
 
asyncio.run(main())

run() starts a session, runs one turn, and closes everything it created. The saved session remains resumable. See Result types for every terminal outcome.

Continue a conversation

import asyncio
from pathlib import Path
 
from droid_sdk import Session
 
 
async def run_turn(session: Session, prompt: str) -> str:
    async with session.stream(prompt) as stream:
        async for _ in stream:
            pass
 
    result = stream.result
    if not result.success:
        raise RuntimeError(result.subtype)
    return result.text
 
 
async def main() -> None:
    async with Session(cwd=Path.cwd()) as session:
        print(await run_turn(session, "What does this project do?"))
        print(await run_turn(session, "What should I test first?"))
 
 
asyncio.run(main())

The second turn includes context from the first. Exiting the session closes the subprocess and session-owned resources. See Default stream types for every yielded value.

Core concepts

Session

A Session owns conversation history, settings, a working directory, and one Droid connection. It accepts one active turn at a time.

Turn

A turn begins with session.stream(prompt) and ends when the stream yields a RunResult. Create separate sessions for parallel work.

Result

A turn's outcome is a value, not an exception: RunSuccess, RunInterrupted, or RunFailure. Interrupted turns, execution failures, and structured-output failures do not raise. Failures in the machinery around a turn (setup, connection, process, protocol, timeout, and cancellation) raise exceptions.

Ownership and cleanup

Typing

The package includes py.typed, and the public API type-checks under strict Pyright and mypy. Public unions narrow with isinstance(). High-level value models are immutable data classes.

Sessions

Use a session when turns share history.

Create and use a session

Session is lazy. It creates the subprocess and session on entry:

async with Session(
    cwd=Path.cwd(),
    model="auto",
) as session:
    async with session.stream("Review the project.") as stream:
        async for message in stream:
            handle(message)

model="auto" selects the Factory Router.

Configure behavior and attach handlers with SessionConfig and InteractionHandlers:

from droid_sdk import (
    Autonomy,
    InteractionHandlers,
    PermissionRequest,
    PermissionResponse,
    SessionConfig,
    ToolConfirmationOutcome,
)
 
 
def approve(request: PermissionRequest) -> PermissionResponse:
    return request.respond(ToolConfirmationOutcome.PROCEED_ONCE)
 
 
config = SessionConfig(
    autonomy=Autonomy.LOW,
    disabled_tools={"Execute"},  # any iterable of tool IDs works here
    disable_builtin_skills=True,
)
 
async with Session(
    config=config,
    interactions=InteractionHandlers(on_permission=approve),
) as session:
    ...

Handlers are plain callables that inspect the request and choose an offered outcome; see Permissions and user input for the full contract.

SessionConfig also accepts mode-specific models, MCP servers, tags, source attribution, machine_id, automatic permission rejection, and native-tool overrides. It does not expose a system-prompt API.

Resume a saved session

async with Session.resume(
    session_id,
    interactions=InteractionHandlers(on_question=answer),
    disabled_tools={"Execute"},
) as session:
    async with session.stream("Continue from the previous prompt.") as stream:
        async for _ in stream:
            pass

Get a session ID from session.id, result.session_id, or list_sessions().

Resume restores conversation history, working directory, title, and settings; runtime concerns such as handlers and MCP servers must be attached again (see the table below). It does not accept a new working directory or model.

Read session state

print(session.id)
print(session.cwd)
print(session.settings.model)
print(session.settings.mode)

These properties are read-only. settings is an immutable snapshot, replaced when Droid reports a settings change.

Update session settings

await session.update_settings(
    model="model-id",
    reasoning_effort=ReasoningEffort.HIGH,
    autonomy=Autonomy.MEDIUM,
    disabled_tools={"Execute"},
)

Only supplied fields change. Set nullable Spec-model fields to None to clear them.

update_settings() accepts:

It returns UpdateSettingsResult. The result currently has no fields.

Rename a session

await session.rename("Authentication review")

rename(title: str) returns None.

List saved sessions

from droid_sdk import list_sessions
 
for saved in await list_sessions(limit=10):
    print(saved.id, saved.title, saved.modified_at)

Pass all_workspaces=True to list sessions across working directories.

list_sessions() reads local files without starting Droid. Results are newest first. Timestamps are timezone-aware.

What persists

Close sessions safely

session = Session()
await session.open()
 
try:
    async with session.stream("Review the project.") as stream:
        async for _ in stream:
            pass
finally:
    await session.close()

open() and close() are idempotent, but a closed session cannot be reopened. Calling an active method before open() raises SessionNotOpenError.

Concurrent open() calls share one startup attempt. If one waiter is cancelled, startup continues for the others. Cancelling the final waiter cancels startup and completes resource cleanup before the session becomes retryable. close() racing startup waits for startup cleanup and leaves the session closed.

Subscribe to raw notifications

unsubscribe = session.on_notification(handle_notification, type="custom_type")
try:
    ...
finally:
    unsubscribe()

High-level streams ignore unknown notifications. on_notification() exposes them without creating a second client.

Streaming and results

Use top-level run() when only the result matters. Session turns always use stream().

session.stream() accepts:

Complete messages

from droid_sdk import AssistantMessage
 
async with session.stream("Run the tests.") as stream:
    async for message in stream:
        if isinstance(message, AssistantMessage):
            print(message.text)

Default stream types

StreamMessage[T] = (
    UserMessage
    | AssistantMessage
    | ToolCall
    | ToolResult
    | HookExecution
    | ErrorEvent
    | RunResult[T]
)

session.stream(prompt) yields:

Values appear in delivery order. RunResult[T] is always last.

Partial events

from droid_sdk import Session, TextDelta
 
async with Session() as session:
    async with session.stream(
        "Explain the failing test.",
        include_partial_messages=True,
    ) as stream:
        async for event in stream:
            if isinstance(event, TextDelta):
                print(event.text, end="", flush=True)
 
    print()
    if not stream.result.success:
        print(f"Turn ended: {stream.result.subtype}")

The default stream yields complete messages. Set include_partial_messages=True to add deltas and operational events. Both modes yield the terminal result and cache it in stream.result. See Partial stream types for the complete event union.

from droid_sdk import RunResult, TextDelta
 
async with session.stream(
    "Explain the failing test.",
    include_partial_messages=True,
) as stream:
    async for event in stream:
        if isinstance(event, TextDelta):
            print(event.text, end="", flush=True)
        elif isinstance(event, RunResult) and not event.success:
            print(f"\nTurn ended: {event.subtype}")

Partial stream types

StreamEvent[T] = (
    StreamMessage[T]
    | TextDelta
    | TextComplete
    | ThinkingDelta
    | ThinkingComplete
    | ToolCallDelta
    | ToolProgress
    | TokenUsageUpdate
    | WorkingStateChanged
    | PermissionResolved
    | SettingsUpdated
    | SessionTitleUpdated
    | SessionWorkingDirectoryChanged
    | McpStatusChanged
    | McpAuthRequired
    | McpAuthCompleted
)

With include_partial_messages=True, the stream yields every StreamMessage[T] plus:

ToolProgress.update is a ToolProgressUpdate:

ToolProgressUpdate.type is "tool_call", "tool_result", "error", "status", or "message".

Unknown high-level events are ignored. Use session.on_notification() when the application needs raw notifications.

Message model

from droid_sdk import AssistantMessage, ConversationMessage, UserMessage
 
completed: list[ConversationMessage] = []
 
async with session.stream("Explain the failing test.") as stream:
    async for event in stream:
        if isinstance(event, (UserMessage, AssistantMessage)):
            completed.append(event)
            save(event)

ConversationMessage defines the fields shared by user and assistant messages:

ContentBlock is a typed union for text, thinking, tool use, tool result, redacted thinking, images, and documents. Narrow blocks with isinstance().

ContentBlock = (
    TextBlock
    | ThinkingBlock
    | RedactedThinkingBlock
    | ToolUseBlock
    | ToolResultBlock
    | ImageBlock
    | DocumentBlock
)

Message is StreamMessage[T] without RunResult[T]. Partial events are not messages.

TextDelta.message_id and block_index identify the assistant content block that eventually appears in AssistantMessage.content. ToolCall.tool_use_id matches the corresponding ToolUseBlock.id.

A complete message can repeat text already delivered through TextDelta. Use deltas for live rendering and complete messages for persistence. Do not concatenate both.

stream.result.messages is the ordered tuple of complete Message values emitted before the result. It excludes the result and partial events.

Handle the result

The terminal RunResult is yielded by the iterator and cached on the stream:

from droid_sdk import RunResult
 
async with session.stream("Run the tests.") as stream:
    async for message in stream:
        if isinstance(message, RunResult):
            print(message.subtype)
 
result = stream.result
if result.success:
    print(result.text)

Reading stream.result before completion raises StreamIncompleteError.

Result types

RunResult[T] is a union of three terminal states:

RunResult[T] = RunSuccess[T] | RunInterrupted[T] | RunFailure[T]

All result variants retain partial output. RunFailure also exposes error and the server's structured_output_error when available. A successful turn may have no structured output.

RunResult does not define truthiness. Check success or subtype.

Token and context usage

if result.usage:
    print(result.usage.input_tokens)
    print(result.usage.output_tokens)
    print(result.usage.cache_read_tokens)
    print(result.usage.cache_creation_tokens)
    print(result.usage.thinking_tokens)
    print(result.usage.factory_credits)

TokenUsageUpdate contains cumulative committed session usage. Context occupancy is separate:

context = await session.context()
print(context.used, context.remaining, context.limit, context.accuracy)

ContextUsage.updated_at records when Droid measured the value.

Errors

async with session.stream("Run the tests.") as stream:
    async for _ in stream:
        pass
 
result = stream.result
if result.subtype == "interrupted":
    print("The turn was interrupted.")
elif result.subtype in ("error_during_execution", "error_structured_output"):
    print(result.error.message if result.error else result.subtype)

Setup, connection, process, protocol, and timeout failures raise DroidError subclasses. Python programming errors use normal built-in exceptions. asyncio.CancelledError is never wrapped.

Stream concurrency

A session may have one active stream. Starting another raises SessionBusyError.

Different sessions may run concurrently.

Timeout

async with session.stream("Perform a long review.", timeout=60) as stream:
    async for event in stream:
        handle(event)

On expiry, the SDK interrupts the turn and raises RunTimeoutError.

Interrupt a turn

Another task may stop active work:

await session.interrupt()

The active stream yields an interrupted result if its consumer remains attached. The session stays open.

Cancellation and interruption

Cancelling the task running a turn sends a best-effort interrupt, releases the subscription, then re-raises asyncio.CancelledError.

Use the stream context manager when iteration may stop early:

async with session.stream("Inspect every test.") as stream:
    async for event in stream:
        if should_stop(event):
            break

Context exit interrupts unfinished work. A bare async iterator cannot guarantee immediate cleanup on break. await stream.aclose() explicitly interrupts and detaches an unfinished stream; it is idempotent.

Models

Model IDs are strings because availability depends on account and organization policy. Omit model to use the configured default, which lives in ~/.factory/settings.json under sessionDefaultSettings. An unknown model ID is rejected by the backend: the turn returns RunFailure(subtype="error_during_execution") with the rejection message in error.message.

Select a model

from droid_sdk import ReasoningEffort, run
 
result = await run(
    "Review this repository.",
    model="model-id",
    reasoning_effort=ReasoningEffort.HIGH,
)

Omit reasoning_effort to use the model default.

Change the model for later turns:

await session.update_settings(
    model="model-id",
    reasoning_effort=ReasoningEffort.HIGH,
)

See Update session settings for the full update_settings() contract.

Use the Factory Router

The model ID auto selects the Factory Router, which routes each prompt to the model with the best balance of quality, latency, and cost. Some product surfaces label it Auto Model; the model ID is auto everywhere.

from droid_sdk import run
 
result = await run("Review this repository.", model="auto")

Pin a session to the router the same way:

async with Session(model="auto") as session:
    ...

Move a live session onto the router:

await session.update_settings(model="auto")

Omit reasoning_effort; the router chooses the effort along with the model and ignores a supplied value. session.settings.model reports auto; the underlying model can differ per response.

The model ID auto is unrelated to Mode.AUTO, the default interaction mode, and to Autonomy, the permission level.

See which model handled a response

Assistant messages in the wire create_message notification carry the underlying model in modelId; routerId is "auto" when the router made the choice. High-level messages omit these fields. Subscribe with on_notification():

from collections.abc import Mapping
 
 
def report_routing(notification: Mapping[str, object]) -> None:
    message = notification.get("message")
    if isinstance(message, Mapping) and message.get("role") == "assistant":
        print(message.get("modelId"), message.get("routerId"))
 
 
unsubscribe = session.on_notification(report_routing, type="create_message")

Wire payloads use camelCase keys and evolve server-side; treat absent keys as normal.

Configure mode-specific models

from droid_sdk import Mode, ReasoningEffort, Session, SessionConfig
 
config = SessionConfig(
    mode=Mode.SPEC,
    spec_model="model-id",
    spec_reasoning_effort=ReasoningEffort.HIGH,
)
 
async with Session(model="model-id", config=config) as session:
    async with session.stream("Draft an implementation plan.") as stream:
        async for _ in stream:
            pass

The primary model handles Auto turns. spec_model handles Spec turns. Switch modes on a live session with enter_spec() and leave_spec(); see Spec mode.

Model configuration contract

None uses the configured default. Setting a Spec field to None through update_settings() clears it.

Inputs and outputs

Images and documents

Attach images and files to a turn with the images and files options:

from droid_sdk import Document, Image, run
 
result = await run(
    "Compare these files.",
    images=[
        Image.from_path("screenshot.png"),
        Image.from_bytes(image_bytes, media_type="image/png"),
    ],
    files=[
        Document.from_path("report.pdf"),
        Document.from_text(source, name="auth.py"),
    ],
)

The constructors read local data and encode it for the turn. Supported image types are PNG, JPEG, GIF, and WebP; image URLs are unsupported. Invalid local input raises InvalidAttachmentError before the turn starts. Attachments are limited to MAX_ATTACHMENT_BYTES (5 MiB), and PDFs to MAX_PDF_ATTACHMENT_BYTES (3 MiB).

Input schemas

images accepts Sequence[Image]. files accepts Sequence[Document]. Wire payloads use canonical mediaType fields. The optional mime hint on text documents is forwarded with the document.

Structured output

Return a Pydantic model

from typing import Literal
 
from pydantic import BaseModel
 
from droid_sdk import RunSuccess, run
 
 
class Finding(BaseModel):
    severity: Literal["low", "medium", "high"]
    message: str
 
 
class Review(BaseModel):
    summary: str
    findings: list[Finding]
 
 
result = await run(
    "Review the authentication code.",
    output=Review,
)
 
if isinstance(result, RunSuccess) and result.output is not None:
    print(result.output.summary)  # success guarantees output when requested
elif result.output_validation_error is not None:
    print(result.output_validation_error)

output accepts a BaseModel subclass, not an arbitrary class. Unsupported types raise TypeError before the turn starts. With output=Review, the return type is RunResult[Review].

When structured output arrives, the SDK validates it against the model. If output was requested, RunSuccess guarantees result.output is set. Missing or invalid output turns the result into RunFailure with subtype error_structured_output; the failure keeps the text, messages, usage, raw structured_output, and output_validation_error for inspection.

Use raw JSON Schema

from droid_sdk import JsonObject, JsonSchema, RunResult, run
 
schema = JsonSchema(
    {
        "type": "object",
        "properties": {"summary": {"type": "string"}},
        "required": ["summary"],
    }
)
 
result: RunResult[JsonObject] = await run(
    "Summarize the repository.",
    output=schema,
)
 
if result.output is not None:
    print(result.output["summary"])

JsonSchema accepts an object-shaped schema. Droid reports unsupported or invalid schemas through the normal result subtype.

Output contract

JsonSchema.schema is FrozenJsonObject. Schema mappings must contain only JSON-compatible values; the SDK validates this and freezes them recursively. Raw schema output remains JsonObject.

JsonValue = bool | int | float | str | list["JsonValue"] | dict[str, "JsonValue"] | None
JsonObject = dict[str, JsonValue]
FrozenJsonValue = (
    bool
    | int
    | float
    | str
    | tuple["FrozenJsonValue", ...]
    | Mapping[str, "FrozenJsonValue"]
    | None
)
FrozenJsonObject = Mapping[str, FrozenJsonValue]

Both local and Droid-side output failures produce RunFailure with subtype error_structured_output. To tell them apart, check structured_output_error.code: local failures use local_validation_failed (with the Pydantic error in output_validation_error) or local_output_missing; Droid-side failures use Droid's own codes.

Permissions and user input

Autonomy

Autonomy controls which actions require approval in Auto mode:

Configure it through SessionConfig:

config = SessionConfig(autonomy=Autonomy.LOW)

Autonomy does not select Auto or Spec mode. Mode controls that.

Permission handler

from droid_sdk import (
    InteractionHandlers,
    PermissionRequest,
    PermissionResponse,
    Session,
    ToolConfirmationOutcome,
)
from droid_sdk.permissions import CreateFile
 
 
def approve(request: PermissionRequest) -> PermissionResponse:
    if request.actions and all(
        isinstance(action, CreateFile) for action in request.actions
    ):
        return request.respond(ToolConfirmationOutcome.PROCEED_ONCE)
    return request.respond(ToolConfirmationOutcome.CANCEL)
 
 
session = Session(
    interactions=InteractionHandlers(on_permission=approve),
)

Droid decides which outcomes are on offer. respond() accepts only an outcome listed in request.options, plus optional comment or edited_spec_content.

An absent handler, an invalid response, or a handler exception cancels the request. Handler failures appear as ErrorEvent values; they do not raise through the active stream.

AskUser handler

from droid_sdk import QuestionRequest, QuestionResponse
 
 
def answer(request: QuestionRequest) -> QuestionResponse:
    answers = [
        question.answer(question.options[0] if question.options else "none")
        for question in request.questions
    ]
    return request.submit(answers)

Cancel the questionnaire:

return request.cancel()

Answers are always strings on the wire. Without a handler, or when a handler fails or returns an invalid shape, the questionnaire is cancelled and the stream receives an ErrorEvent.

question.answer(value) answers one value. question.answer_multiple(values) joins multi-select values with ", " to produce the wire-compatible string.

Handlers run inside the active turn. Use async handlers for I/O, and do not start another turn on the same session from a handler.

Interaction contracts

PermissionHandler = Callable[
    [PermissionRequest],
    PermissionResponse | Awaitable[PermissionResponse],
]
QuestionHandler = Callable[
    [QuestionRequest],
    QuestionResponse | Awaitable[QuestionResponse],
]

PermissionAction is a discriminated union:

PermissionAction = (
    EditAction
    | ExecuteAction
    | CreateFile
    | AskUserAction
    | ExitSpecModeAction
    | ApplyPatchAction
    | McpToolAction
    | SandboxViolationAction
    | DroidShieldViolationAction
)

Every action includes its tool_use, confirmation type, and typed details:

AskUserParseError(message: str, line: int | None = None) records malformed questionnaire details. ApplyPatchFile requires file_path, file_name, and operation ("create", "update", or "delete"); move_to, old_content, and new_content default to None.

ToolConfirmationOutcome defines:

Only outcomes present in PermissionRequest.options are valid.

Tool controls

config = SessionConfig(
    additional_tools={"CustomTool"},
    enabled_tools={"Read"},
    disabled_tools={"Execute", "Edit"},
    restrict_tools={"Read", "Grep"},
)

Each of the four sets has a distinct role:

• additional_tools adds IDs to the catalog.
• enabled_tools enables otherwise available tools.
• disabled_tools removes tools.
• restrict_tools is a restrictive allowlist and never elevates permission.

for tool in await session.list_tools(model="model-id", mode=Mode.AUTO):
    print(tool.id, tool.allowed)

update_settings() accepts the same override fields for later turns. The four override parameters accept any iterable of tool IDs, such as a set, list, or tuple. Passing a bare string raises TypeError.

list_tools() returns list[ToolInfo].

Extensions

Skills

skills = await session.list_skills()
for skill in skills.skills:
    print(skill.name, skill.enabled)

skills.project_available reports whether a project skill scope is available; it is None when Droid does not report it.

Enable or disable a skill:

await session.enable_skill("review", scope="project")
await session.disable_skill("legacy", scope="user")

Skills may come from project, personal, built-in, or automation settings.

Skill schemas

scope is "user" or "project". SkillResource.type is "reference" or "asset". SkillInfo.location is "project", "personal", "builtin", or "automation".

In-process MCP tools

In-process servers require the mcp extra (pip install "droid-sdk[mcp]"). Use @tool to expose an annotated Python function:

from droid_sdk.mcp import create_sdk_mcp_server, tool
 
 
@tool("lookup_owner", "Return the owner of a repository file.")
async def lookup_owner(path: str) -> str:
    return f"Owner for {path}: platform-team"
 
 
server = create_sdk_mcp_server(
    name="review-tools",
    tools=[lookup_owner],
    version="1.0.0",
)
config = SessionConfig(mcp_servers=[server])

The decorator derives JSON Schema from type annotations. Invalid input is returned to Droid as a tool error and is not passed to the function.

Tool functions may be synchronous or asynchronous. They may return text or a typed ToolResponse.

The SDK starts an authenticated loopback server and closes it with the session. Attach in-process servers again when resuming.

In-process MCP schemas

tool(name, description) is a decorator; tool(name, description, function) is the equivalent direct call. A return annotation that describes an object is validated at call time, returned as structured content, and advertised through the MCP outputSchema field.

create_sdk_mcp_server(name, tools, version="1.0.0") returns SdkMcpServer. SdkMcpServer.config is the active HttpMcpServerConfig or None; await server.start() returns that config, and await server.close() stops the server. Sessions normally own these calls.

External MCP servers

External server configs do not need the mcp extra:

from droid_sdk import HttpMcpServerConfig, StdioMcpServerConfig
 
config = SessionConfig(
    mcp_servers=[
        HttpMcpServerConfig(name="docs", url="https://example.com/mcp"),
        StdioMcpServerConfig(
            name="search",
            command="python",
            args=["-m", "search_server"],
        ),
    ],
)

HTTP, SSE, and stdio transports are supported. HTTP and SSE configurations support headers and OAuth settings.

Inspect connected servers and tools:

servers = await session.list_mcp_servers()
tools = await session.list_mcp_tools()
 
print(servers.summary)
for server in servers.servers:
    print(server.name, server.status)

Session methods can add, remove, enable, disable, and authenticate external servers. Configuration mutations affect the user's Droid settings. Servers passed through SessionConfig are session-scoped.

External MCP schemas

McpServerConfig = (
    StdioMcpServerConfig | HttpMcpServerConfig | SseMcpServerConfig | SdkMcpServer
)

oauth accepts McpOAuthOptions or False.

Remove, enable, and disable mutate user-scoped MCP configuration only. These methods do not accept a project-scope argument.

Hooks

There is no Python API for defining hooks; configure them in .factory/hooks.json. Hook execution appears as HookExecution values in the run stream. HookExecution.status is "started", "completed", or "error".

Session lifecycle

Fork, compact, and rewind create successor sessions. The successor is an opened Session that takes over the existing Droid connection; fork() returns it directly, while compact() and rewind() return it on their outcome objects.

Fork

Fork copies the conversation into a new session and continues there.

fork = await session.fork(
    title="Alternative approach",
    tags=[SessionTag(name="experiment")],
)
 
async with fork:
    async with fork.stream("Try the other strategy.") as stream:
        async for _ in stream:
            pass

Compact

Compaction summarizes older conversation history to free context-window space.

outcome = await session.compact(instructions="Keep decisions and unresolved failures.")
 
async with outcome.session as compacted:
    print(outcome.removed_count)

Rewind

Rewind returns the conversation to an earlier message and can restore or delete files changed since.

info = await session.rewind_info(message_id)
 
outcome = await session.rewind(
    message_id,
    restore=info.available_files,
    delete=info.created_files,
    title="Before the failed change",
)
 
async with outcome.session as rewound:
    print(outcome.restored_count, outcome.deleted_count)
    print(outcome.failed_restore_count, outcome.failed_delete_count)

info.evicted_files explains files that cannot be restored.

Lifecycle contracts

Successor ownership

After a successful replacement:

• the returned successor owns the connection and runtime resources
• the source session becomes retired
• id, cwd, and settings remain readable on the source
• active methods on the source raise SessionReplacedError
• closing the source is a no-op

Replacing a session with an active turn raises SessionBusyError.

Only one replacement may run at a time. open() and another replacement raise SessionBusyError while replacement is active. close() racing a replacement waits; on success it closes the successor, and on rollback it closes the source.

Cancelling a replacement before handoff restores the source to open state. If cancellation arrives after a successor is created, the SDK reloads the source with its attached policies and retires the detached successor. If source restoration fails, the SDK closes the connection rather than leaving an ambiguous owner.

Spec mode

Spec mode lets Droid inspect a codebase and propose a plan without changing files.

Enter Spec Mode

await session.enter_spec(
    model="model-id",
    reasoning_effort=ReasoningEffort.HIGH,
)

Start directly in Spec mode with SessionConfig(mode=Mode.SPEC).

Leave without approving a plan

await session.leave_spec()

This changes the mode only. It does not approve the plan or start implementation.

Approve a plan

Plan approval arrives through the permission handler:

def approve(request: PermissionRequest) -> PermissionResponse:
    if request.plan:
        print(request.plan.text)
        return request.respond(ToolConfirmationOutcome.PROCEED_ONCE)
    return request.respond(ToolConfirmationOutcome.CANCEL)

Return any offered PROCEED_NEW_SESSION* outcome to hand implementation to a new session. Return PROCEED_EDIT with edited_spec_content to revise the plan. CANCEL ends the turn with an interrupted result.

Mode contract

Entering or leaving Spec mode only changes settings. Plan approval is an ExitSpecModeAction permission request, and only an offered ToolConfirmationOutcome is valid.

Observability and advanced APIs

Observability

from droid_sdk import Runtime, run
from droid_sdk.observability import LogEvent, Observability
 
 
class PrintLogger:
    def log(self, event: LogEvent) -> None:
        print(event.level, event.name, event.message)
 
 
runtime = Runtime(observability=Observability(logger=PrintLogger()))
 
result = await run("Check repository status.", runtime=runtime)

Logger, metric, and trace-context sinks are synchronous and best-effort. Sink failures never fail a Droid operation.

Events exclude prompts, messages, thinking, tool inputs and results, file contents, raw process output, stack traces, and credentials.

Observability schemas

attributes is Mapping[str, str | int | float | bool | None]. Log levels are "debug", "info", "warn", and "error". Metric kinds are "counter" and "histogram". TraceContext is deliberately mutable so tracing providers can inject values into it.

Custom runtime

Runtime holds process configuration:

runtime = Runtime(
    executable=Path("/opt/factory/bin/droid"),
    args=["--flag"],
    env={"EXAMPLE": "value"},
)

Environment entries extend the current process environment when the SDK starts Droid.

Runtime schema

Examples

Run commands from the repository root:

Live examples require an authenticated local Droid CLI or FACTORY_API_KEY; every model call uses a finite timeout.

API reference

Root constants and supporting types

Top-level functions

Session

RunStream

Main enums

ReasoningEffort defines NONE, DYNAMIC, OFF, MINIMAL, LOW, MEDIUM, HIGH, EXTRA_HIGH, and MAX. Model and tool IDs are plain strings, not enums.

On the wire, these enums use the following values:

ToolCategory: read, edit, execute, other
ToolConfirmationType: edit, exec, create, ask_user, exit_spec_mode,
  apply_patch, mcp_tool, sandbox_violation, droid_shield_violation
OAuthTokenEndpointAuthMethod: none, client_secret_basic, client_secret_post
SandboxOperation: read, write, network, tool
SandboxViolationType: filesystem-read, filesystem-write, network, tool
SandboxViolationReason: deny-list, not-allowed
SessionPlatform: slack, web, api, sessions_api, jira, linear,
  microsoft-teams, readiness-remediation, readiness-evaluation, automation,
  wiki-generation, wiki-ci-setup, tui, desktop, acp, unknown

Exceptions

All SDK-defined exceptions derive from DroidError. Python built-ins and asyncio.CancelledError do not.

Exception constructor metadata is public:

Detailed session contracts

Session construction contract

Session(...) accepts:

SessionConfig fields:

Session.resume(session_id, ...) accepts only values that can be reattached:

Session state schemas

Every SessionSettingsUpdate field defaults to None. Its field types match the update_settings() table above.

list_sessions() returns list[SavedSession]. Its filters are cwd, all_workspaces, and limit.

on_notification(callback, type=None) passes Mapping[str, object] to the callback and returns an unsubscribe function.

SessionSource requires platform: SessionPlatform; every other attribution field is optional and defaults to None. Required combinations are validated when converted to the wire protocol.

Known limitations

• The SDK runs local Droid subprocess sessions only.
• The API is asyncio-only.
• Hooks are configured through Droid files, not Python callbacks.
• Image URLs are not supported by the local runtime.
• One session can run one turn at a time.