Python SDK

Complete API reference for the Amp Python SDK. This document provides detailed information about all functions, types, and interfaces.

Installation

# 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 with @ampcode/cli@0.0.1779896748-g596c49, then set AMP_CLI_PATH so the Python SDK uses that CLI binary:

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.

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, optional) - Configuration options for Amp execution

Returns

• AsyncIterator[StreamMessage] - Stream of messages from the Amp CLI

Example

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.

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 - A formatted user input message

Example

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'}]})

create_permission()

Helper function to create permissions plugin rule objects for controlling tool usage.

def create_permission(
    tool: str,
    action: Literal["allow", "reject", "ask", "delegate"],
    options: Optional[dict[str, Any]] = None
) -> Permission

Parameters

• tool (str) - The name of the tool to which this permission applies (supports glob patterns)
• action (Literal["allow", "reject", "ask", "delegate"]) - How Amp should proceed when matched
• options (dict, optional) - Additional configuration for the permission
  • matches (dict[str, PermissionMatchCondition]) - Match conditions for tool arguments
  • context (Literal["thread", "subagent"]) - Only apply this rule in specific context
  • to (str) - Command to delegate to (required when action is "delegate")

Returns

• Permission - A permission object that can be used in the permissions list

Examples

from amp_sdk import create_permission

# Allow all Bash commands
create_permission("Bash", "allow")

# Allow specific git commands
create_permission("Bash", "allow", {
    "matches": {"cmd": "git *"}
})

# Ask for approval on Read operations for sensitive paths
create_permission("Read", "ask", {
    "matches": {"path": "/etc/*"}
})

# Delegate web browsing to a custom command
create_permission("mcp__playwright__*", "delegate", {
    "to": "node browse.js"
})

# Only apply in subagent context
create_permission("Bash", "reject", {
    "context": "subagent"
})

Types

AmpOptions

Configuration options for the execute() function.

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
    dangerously_allow_all: Optional[bool] = 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
    permissions: Optional[list[Permission]] = None
    labels: Optional[list[str]] = None
    thinking: bool = False
    executor: Optional[Literal["local", "orb"]] = 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. With executor="orb", prompt must be a string rather than streaming input. 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. The orb executor ignores local-only options (permissions, enabled_tools, skills, mcp_config, and dangerously_allow_all) with a warning; configure those in the Amp project instead.

Properties

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.

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

AssistantMessage

AI assistant responses with text content and tool usage.

class AssistantMessage(BaseModel):
    type: Literal["assistant"] = "assistant"
    session_id: str
    message: AssistantMessageDetails
    parent_tool_use_id: Optional[str] = None

Properties

UserMessage

User input message (echoed back in stream).

class UserMessage(BaseModel):
    type: Literal["user"] = "user"
    session_id: str
    message: UserMessageDetails
    parent_tool_use_id: Optional[str] = None

Properties

ResultMessage

Final successful execution result.

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

ErrorResultMessage

Final error result indicating execution failure.

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

TextContent

Plain text content block.

class TextContent(BaseModel):
    type: Literal["text"] = "text"
    text: str

ToolUseContent

Tool execution request.

class ToolUseContent(BaseModel):
    type: Literal["tool_use"] = "tool_use"
    id: str
    name: str
    input: dict[str, Any]

ToolResultContent

Result from tool execution.

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.

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

Input Types

UserInputMessage

Formatted user input message for streaming conversations.

class UserInputMessage(BaseModel):
    type: Literal["user"] = "user"
    request_id: Optional[str] = None
    message: UserInputMessageDetails

Where UserInputMessageDetails is:

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.

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:

mcp_config = MCPConfig(servers={
    "playwright": {"command": "npx", "args": ["-y", "@playwright/mcp"]},
    "remote": {"url": "https://api.example.com/mcp"}
})

Stdio server properties:

HTTP server properties:

Permission

Individual permissions plugin rule for controlling tool usage.

class Permission(BaseModel):
    tool: str
    matches: Optional[dict[str, Any]] = None
    action: Literal["allow", "reject", "ask", "delegate"]
    context: Optional[Literal["thread", "subagent"]] = None
    to: Optional[str] = None

Properties

Example

import asyncio
from amp_sdk import execute, AmpOptions, create_permission

async def main():
    async for message in execute(
        "Deploy the application",
        AmpOptions(
            permissions=[
                # Allow git commands
                create_permission("Bash", "allow", {"matches": {"cmd": "git *"}}),
                # Allow reading files
                create_permission("Read", "allow"),
            ]
        )
    ):
        # Handle messages
        pass

asyncio.run(main())

PermissionMatchCondition

Match condition for tool arguments. Supports strings (with glob patterns or regex), lists (OR logic), booleans, numbers, None, and nested dicts.

PermissionMatchCondition = Union[
    str,
    bool,
    int,
    float,
    None,
    list['PermissionMatchCondition'],
    dict[str, 'PermissionMatchCondition']
]

Examples

# String pattern with wildcard
{"cmd": "npm *"}

# List for OR logic
{"cmd": ["npm install", "npm test", "npm run build"]}

# Regex pattern
{"cmd": "/^git (status|log|diff)$/"}

# Nested dict matching
{"env": {"NODE_ENV": "production"}}

Exceptions

The SDK provides several exception types for error handling:

AmpError

Base exception for all Amp SDK errors.

class AmpError(Exception):
    """Base exception for Amp SDK errors."""
    pass

CLINotFoundError

Raised when the Amp CLI cannot be found.

class CLINotFoundError(AmpError):
    """Amp CLI not found."""
    pass

ProcessError

Raised when the CLI process fails.

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.

class ValidationError(AmpError):
    """Input validation error."""
    pass

JSONParseError

Raised when JSON parsing fails.

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.

class AmpTimeoutError(AmpError):
    """Operation timeout error."""
    pass

CancellationError

An Amp-specific cancellation error type. execute() propagates asyncio.CancelledError when its consumer is cancelled.

class CancellationError(asyncio.CancelledError, AmpError):
    """Operation cancelled."""
    pass

Requirements

• Python 3.9 or higher
• The Amp CLI must be installed (installation instructions)