--- lastModified: 2026-09-18 --- # Python SDK API reference for the Amp Python SDK.

Installation

```bash # Install the Amp SDK using pip pip install amp-sdk # Install the Amp CLI with bash curl -fsSL https://ampcode.com/install.sh | bash # Or with npm npm install -g @ampcode/cli ``` If you need to use Amp before Amp Neo, install [`amp-sdk==0.1.5`](https://pypi.org/project/amp-sdk/0.1.5/) with [`@ampcode/cli@0.0.1779896748-g596c49`](https://www.npmjs.com/package/@ampcode/cli/v/0.0.1779896748-g596c49), then set `AMP_CLI_PATH` so the Python SDK uses that CLI binary: ```bash pip install amp-sdk==0.1.5 npm install -g @ampcode/cli@0.0.1779896748-g596c49 export AMP_CLI_PATH="$(command -v amp)" ```

Functions

### execute() The main function for executing Amp CLI commands programmatically. ```python async def execute( prompt: Union[str, AsyncIterator[UserInputMessage]], options: Optional[AmpOptions] = None ) -> AsyncIterator[StreamMessage] ``` **Parameters** - `prompt` (`str | AsyncIterator[UserInputMessage]`) - The user prompt as a string or async iterator of user input messages for multi-turn conversations - `options` ([`AmpOptions`](#ampoptions), optional) - Configuration options for Amp execution **Returns** - `AsyncIterator[StreamMessage]` - Stream of messages from the Amp CLI **Example** ```python import asyncio from amp_sdk import execute, AmpOptions async def main(): async for message in execute( "Analyze this codebase", AmpOptions( cwd="./my-project" ) ): if message.type == "assistant": print("Assistant:", message.message.content) elif message.type == "result" and not message.is_error: print("Final result:", message.result) break asyncio.run(main()) ``` ### create_user_message() Helper function to create properly formatted user input messages for streaming conversations. ```python def create_user_message(text: str, request_id: str | None = None) -> UserInputMessage ``` **Parameters** - `text` (`str`) - The text content for the user message - `request_id` (`str | None`) - An optional 1–256 character ID. Reusing it for a retry on the same thread prevents Amp from creating another user message. **Returns** - [`UserInputMessage`](#userinputmessage) - A formatted user input message **Example** ```python from amp_sdk import create_user_message message = create_user_message("Analyze this code", request_id="analysis-123") print(message) # Output: UserInputMessage(type='user', request_id='analysis-123', message={'role': 'user', 'content': [{'type': 'text', 'text': 'Analyze this code'}]}) ```

Types

### AmpOptions Configuration options for the `execute()` function. ```python class AmpOptions(BaseModel): cwd: Optional[str] = None mode: str = "medium" # Prefer "low", "medium", "high", or "ultra" effort: Optional[Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"]] = None no_archive_after_execute: Optional[bool] = None visibility: Optional[Literal["private", "unlisted", "workspace", "group"]] = "workspace" settings_file: Optional[str] = None log_level: Optional[Literal["debug", "info", "warn", "error", "audit"]] = None log_file: Optional[str] = None env: dict[str, str] = Field(default_factory=dict) continue_thread: Union[bool, str, None] = None mcp_config: Optional[Union[MCPConfig, str]] = None skills: Optional[str] = None enabled_tools: Optional[list[str]] = None labels: Optional[list[str]] = None thinking: bool = False executor: Optional[Literal["local", "orb", "runner"]] = None runner_id: Optional[str] = None runner_dir: Optional[str] = None project: Optional[str] = None title: Optional[str] = None ``` The built-in modes are `"low"`, `"medium"`, `"high"`, and `"ultra"`. You can also pass a string for a custom plugin-defined mode. `executor` defaults to local execution. `executor="orb"` runs the agent in an [orb](/docs/orbs); `executor="runner"` runs it on one of your [runners](/docs/cli/runners), identified by `runner_id`. Set `runner_dir` to the absolute path of one directory the runner serves, or omit it to use its starting directory. Each thread has one working directory; use separate `execute()` calls for multiple directories. Continuing a thread ignores this option and keeps its existing directory. `cwd` only sets the local CLI subprocess directory, not the remote runner directory. With either remote executor, `prompt` must be a string rather than streaming input, and local-only options such as `enabled_tools`, `skills`, and `mcp_config` are ignored with a warning; configure those in the Amp project or on the runner instead. Plugin agent modes must be loaded on the runner. `project` requires orb execution; when it is omitted, Amp may infer a project from the Git remotes under `cwd`, or start the orb without a repository if none match. **Properties** | Property | Type | Default | Description | | -------------------------- | ----------------------------------------------------------------------------- | ------------- | -------------------------------------------------------------------------------------------------------- | | `cwd` | `str \| None` | `None` | Working directory of the local CLI subprocess | | `mode` | `str` | `"medium"` | Prefer `"low"`, `"medium"`, `"high"`, or `"ultra"`; custom plugin-defined mode strings are also accepted | | `effort` | `Literal["none", "minimal", "low", "medium", "high", "xhigh", "max"] \| None` | `None` | Reasoning effort for supported modes | | `no_archive_after_execute` | `bool \| None` | `None` | Leave new execute threads unarchived after execution completes | | `visibility` | `Literal["private", "unlisted", "workspace", "group"] \| None` | `"workspace"` | Thread visibility level | | `settings_file` | `str \| None` | `None` | Path to custom settings file | | `log_level` | `Literal["debug", "info", "warn", "error", "audit"] \| None` | `None` | Logging verbosity level | | `log_file` | `str \| None` | `None` | Path to write logs | | `continue_thread` | `bool \| str \| None` | `None` | Continue most recent thread (`True`) or specific thread by ID (`str`) | | `mcp_config` | `MCPConfig \| str \| None` | `None` | MCP server configuration as JSON string, dict, or config object | | `env` | `dict[str, str]` | `{}` | Additional environment variables | | `skills` | `str \| None` | `None` | Folder path with custom skills | | `enabled_tools` | `list[str] \| None` | `None` | Tool name patterns to enable (maps to `amp.tools.enable`) | | `labels` | `list[str] \| None` | `None` | Labels to add to the thread | | `thinking` | `bool` | `False` | Include thinking blocks in the result stream | | `executor` | `Literal["local", "orb", "runner"] \| None` | `None` | Run in the local CLI process (the default), a remote orb, or one of your runners | | `runner_id` | `str \| None` | `None` | Runner ID for a new runner thread; requires `executor="runner"` | | `runner_dir` | `str \| None` | `None` | Absolute path to one served directory for a new runner thread; requires `executor="runner"` | | `project` | `str \| None` | Inferred | Amp project for a new orb thread; requires `executor="orb"` | | `title` | `str \| None` | `None` | Title for a new thread, up to 256 characters |

Message Types

The SDK streams various message types during execution. All messages implement the base `StreamMessage` type. ### SystemMessage Initial message containing session information and available tools. ```python class SystemMessage(BaseModel): type: Literal["system"] = "system" subtype: Literal["init"] = "init" session_id: str cwd: str tools: list[str] # Each item exposes name: str and status: MCPConnectionStatus mcp_servers: list ``` Each `mcp_servers` item exposes a `name` string and a `status` of `"awaiting-approval"`, `"authenticating"`, `"connecting"`, `"reconnecting"`, `"connected"`, `"denied"`, `"failed"`, or `"blocked-by-registry"`. **Properties** | Property | Type | Description | | ------------- | ----------- | -------------------------------------------- | | `session_id` | `str` | Unique identifier for this execution session | | `cwd` | `str` | Current working directory | | `tools` | `list[str]` | List of available tool names | | `mcp_servers` | `list` | MCP server items with the inline shape above | ### AssistantMessage AI assistant responses with text content and tool usage. ```python class AssistantMessage(BaseModel): type: Literal["assistant"] = "assistant" session_id: str message: AssistantMessageDetails parent_tool_use_id: Optional[str] = None ``` **Properties** | Property | Type | Description | | --------------------- | ------------------------------------------------------- | ------------------------------------------------ | | `session_id` | `str` | Unique identifier for this execution session | | `message` | `AssistantMessageDetails` | The assistant's message content | | `message.id` | `str \| None` | Message identifier | | `message.role` | `Literal["assistant"]` | Message role (always "assistant") | | `message.model` | `str \| None` | Model used to generate this response | | `message.content` | `list[TextContent \| ToolUseContent]` | Message content blocks | | `message.stop_reason` | `Literal["end_turn", "tool_use", "max_tokens"] \| None` | Why generation stopped | | `message.usage` | `Usage \| None` | Token usage information | | `parent_tool_use_id` | `str \| None` | ID of parent tool use if this is from a subagent | ### UserMessage User input message (echoed back in stream). ```python class UserMessage(BaseModel): type: Literal["user"] = "user" session_id: str message: UserMessageDetails parent_tool_use_id: Optional[str] = None ``` **Properties** | Property | Type | Description | | -------------------- | -------------------- | ------------------------------------------------ | | `session_id` | `str` | Unique identifier for this execution session | | `message` | `UserMessageDetails` | The user's message content | | `parent_tool_use_id` | `str \| None` | ID of parent tool use if this is a tool response | ### ResultMessage Final successful execution result. ```python class ResultMessage(BaseModel): type: Literal["result"] = "result" subtype: Literal["success"] = "success" session_id: str is_error: Literal[False] = False result: str duration_ms: int num_turns: int usage: Optional[Usage] = None permission_denials: Optional[list[str]] = None ``` **Properties** | Property | Type | Description | | -------------------- | ----------------- | -------------------------------------------- | | `session_id` | `str` | Unique identifier for this execution session | | `result` | `str` | The final result from the assistant | | `duration_ms` | `int` | Total execution time in milliseconds | | `num_turns` | `int` | Number of conversation turns | | `usage` | [`Usage`](#usage) | Token usage information | | `permission_denials` | `list[str]` | List of permissions that were denied | ### ErrorResultMessage Final error result indicating execution failure. ```python class ErrorResultMessage(BaseModel): type: Literal["result"] = "result" subtype: Literal["error_during_execution", "error_max_turns"] session_id: str is_error: Literal[True] = True error: str duration_ms: int num_turns: int usage: Optional[Usage] = None permission_denials: Optional[list[str]] = None ``` **Properties** | Property | Type | Description | | -------------------- | ----------------- | -------------------------------------------- | | `session_id` | `str` | Unique identifier for this execution session | | `error` | `str` | Error message describing what went wrong | | `duration_ms` | `int` | Total execution time in milliseconds | | `num_turns` | `int` | Number of conversation turns | | `usage` | [`Usage`](#usage) | Token usage information | | `permission_denials` | `list[str]` | List of permissions that were denied | ### TextContent Plain text content block. ```python class TextContent(BaseModel): type: Literal["text"] = "text" text: str ``` ### ToolUseContent Tool execution request. ```python class ToolUseContent(BaseModel): type: Literal["tool_use"] = "tool_use" id: str name: str input: dict[str, Any] ``` ### ToolResultContent Result from tool execution. ```python class ToolResultContent(BaseModel): type: Literal["tool_result"] = "tool_result" tool_use_id: str content: str is_error: bool = False ``` ### Usage Token usage and billing information from API calls. ```python class Usage(BaseModel): input_tokens: int = 0 output_tokens: int = 0 cache_creation_input_tokens: int = 0 cache_read_input_tokens: int = 0 service_tier: Optional[str] = None ``` **Properties** | Property | Type | Description | | ----------------------------- | ------------- | ---------------------------------- | | `input_tokens` | `int` | Number of input tokens used | | `cache_creation_input_tokens` | `int` | Tokens used for cache creation | | `cache_read_input_tokens` | `int` | Tokens read from cache | | `output_tokens` | `int` | Number of output tokens generated | | `service_tier` | `str \| None` | Service tier used for this request |

Input Types

### UserInputMessage Formatted user input message for streaming conversations. ```python class UserInputMessage(BaseModel): type: Literal["user"] = "user" request_id: Optional[str] = None message: UserInputMessageDetails ``` Where `UserInputMessageDetails` is: ```python class UserInputMessageDetails(BaseModel): role: Literal["user"] = "user" content: list[Union[TextContent, ToolResultContent]] ``` ### MCPConfig Configuration for MCP (Model Context Protocol) servers. Supports both stdio-based and HTTP-based servers. ```python class MCPConfig(BaseModel): servers: dict[str, MCPServer] = Field(default_factory=dict) def to_json_string(self) -> str: """Convert to JSON string for CLI argument.""" ... # MCPServer is a union of stdio and HTTP server configurations ``` `MCPServer` accepts either a stdio server config (with `command`) or an HTTP server config (with `url`). Pass server configs as dicts: ```python mcp_config = MCPConfig(servers={ "playwright": {"command": "npx", "args": ["-y", "@playwright/mcp"]}, "remote": {"url": "https://api.example.com/mcp"} }) ``` **Stdio server properties:** | Property | Type | Required | Description | | ---------- | ---------------- | -------- | ------------------------------------ | | `command` | `str` | Yes | Command to start the MCP server | | `args` | `list[str]` | No | Command line arguments | | `env` | `dict[str, str]` | No | Environment variables for the server | | `disabled` | `bool \| None` | No | Whether this server is disabled | **HTTP server properties:** | Property | Type | Required | Description | | ----------- | ---------------- | -------- | -------------------------------------- | | `url` | `str` | Yes | URL of the HTTP MCP server | | `headers` | `dict[str, str]` | No | HTTP headers to send with requests | | `transport` | `str \| None` | No | Transport type (e.g., "sse") | | `oauth` | `object \| None` | No | OAuth configuration for authentication | | `disabled` | `bool \| None` | No | Whether this server is disabled |

Exceptions

The SDK provides several exception types for error handling: ### AmpError Base exception for all Amp SDK errors. ```python class AmpError(Exception): """Base exception for Amp SDK errors.""" pass ``` ### CLINotFoundError Raised when the Amp CLI cannot be found. ```python class CLINotFoundError(AmpError): """Amp CLI not found.""" pass ``` ### ProcessError Raised when the CLI process fails. ```python class ProcessError(AmpError): """CLI process error.""" exit_code: int stderr: str signal: str = "" ``` ### ValidationError An Amp-specific validation error type. Invalid `AmpOptions` currently raise `pydantic.ValidationError` directly. ```python class ValidationError(AmpError): """Input validation error.""" pass ``` ### JSONParseError Raised when JSON parsing fails. ```python class JSONParseError(AmpError): """JSON parsing error.""" raw_line: str ``` ### AmpTimeoutError An Amp-specific timeout error type. `execute()` does not set a timeout; wrapping stream consumption in `asyncio.wait_for()` raises `asyncio.TimeoutError` when the deadline expires. ```python class AmpTimeoutError(AmpError): """Operation timeout error.""" pass ``` ### CancellationError An Amp-specific cancellation error type. `execute()` propagates `asyncio.CancelledError` when its consumer is cancelled. ```python class CancellationError(asyncio.CancelledError, AmpError): """Operation cancelled.""" pass ```

Requirements

- Python 3.9 or higher - The Amp CLI must be installed ([installation instructions](/docs/cli#install))