· SDK Documentation

Python Reference

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 Core using npm
npm install -g @sourcegraph/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",
            dangerously_allow_all=True
        )
    ):
        if message.type == "assistant":
            print("Assistant:", message.message.content)
        elif message.type == "result":
            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) -> UserInputMessage

Parameters

• text (str) - The text content for the user message

Returns

• UserInputMessage - A formatted user input message

Example

from amp_sdk import create_user_message

message = create_user_message("Analyze this code")
print(message)
# Output: UserInputMessage(type='user', message={'role': 'user', 'content': [{'type': 'text', 'text': 'Analyze this code'}]})

create_permission()

Helper function to create permission 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 before allowing Read operations on 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: Literal["smart", "rush", "deep"] = "smart"
    dangerously_allow_all: bool = False
    archive: Optional[bool] = None
    visibility: Optional[Literal["private", "public", "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
    toolbox: Optional[str] = None
    skills: Optional[str] = None
    permissions: Optional[list[Permission]] = None
    labels: Optional[list[str]] = None

Properties

Property	Type	Default	Description
cwd	str \| None	None	Current working directory for execution
mode	Literal["smart", "rush", "deep"]	"smart"	Agent mode - controls model, system prompt, and tool selection
dangerously_allow_all	bool	False	Allow all tool usage without permission prompts
archive	bool \| None	None	Archive the thread after execution completes
visibility	Literal["public", "private", "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
toolbox	str \| None	None	Folder path with toolbox scripts
skills	str \| None	None	Folder path with custom skills
permissions	list[Permission] \| None	None	Permission rules for tool usage
labels	list[str] \| None	None	Labels to add to the thread

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]
    mcp_servers: list[MCPServerStatus]

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[MCPServerStatus]	Status of MCP servers

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

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).

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.

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	Token usage information
permission_denials	list[str]	List of permissions that were denied

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

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	Token usage information
permission_denials	list[str]	List of permissions that were denied

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

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.

class UserInputMessage(BaseModel):
    type: Literal["user"] = "user"
    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:

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

Permission

Individual permission 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

Property	Type	Required	Description
tool	str	Yes	Tool name (supports glob patterns like Bash or mcp__*)
matches	dict[str, PermissionMatchCondition] \| None	No	Match conditions for tool arguments
action	Literal["allow", "reject", "ask", "delegate"]	Yes	How Amp should proceed when the rule matches
context	Literal["thread", "subagent"] \| None	No	Apply rule only in main thread or sub-agents
to	str \| None	No	Command to delegate to (required when action is "delegate")

Example

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: Optional[str] = None

ValidationError

Raised when input validation fails.

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

JSONParseError

Raised when JSON parsing fails.

class JSONParseError(AmpError):
    """JSON parsing error."""
    raw_line: str

AmpTimeoutError

Raised when an operation times out.

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

CancellationError

Raised when an operation is cancelled.

class CancellationError(AmpError):
    """Operation cancelled."""
    pass

Requirements

• Python 3.9 or higher
• The Amp CLI must be installed (npm install -g @sourcegraph/amp)