Amp SDK for Elixir

An idiomatic Elixir SDK for Amp (by Sourcegraph) -- the agentic coding assistant. Wraps the Amp CLI with streaming JSON output, multi-turn conversations, thread management, MCP server integration, and fine-grained permission control.

Note: This SDK requires the Amp CLI to be installed on the host machine. The SDK communicates with Amp exclusively through its --execute --stream-json interface -- no direct API calls are made.

Documentation Menu

• README.md - installation, quick start, and runtime boundaries
• guides/getting-started.md - first prompts and streams
• guides/configuration.md - execution options and environment shaping
• guides/streaming.md - event flow and result handling
• guides/threads.md - thread lifecycle and continuation
• guides/testing.md - local validation workflow

What You Can Build

• Automated code review and refactoring pipelines
• CI/CD integrations that use Amp to fix failing tests or lint issues
• Multi-agent orchestration with Amp as a coding sub-agent
• Chat interfaces backed by Amp's coding capabilities
• Batch processing across repositories with thread continuity
• Custom developer tools with approval hooks and permission policies

Installation

Add amp_sdk to your dependencies in mix.exs:

def deps do
  [
    {:amp_sdk, "~> 0.5.0"}
  ]
end

Then fetch dependencies:

mix deps.get

Prerequisites

Amp CLI

Install the Amp CLI binary:

curl -fsSL https://ampcode.com/install.sh | bash

Or via npm:

npm install -g @sourcegraph/amp

Verify the installation:

amp --version

Authentication

Log in to your Amp account (required for execution):

amp login

Or set the AMP_API_KEY environment variable:

export AMP_API_KEY=your-api-key

CLI Discovery

The SDK locates the Amp CLI automatically by checking, in order:

When execution_surface targets SSH, amp_sdk does not forward a local AMP_CLI_PATH into remote execution. Remote surfaces resolve the provider command as amp and expect that binary to exist on the remote PATH.

Quick Start

1. Run a Simple Query

{:ok, result} = AmpSdk.run("What files are in this directory?")
IO.puts(result)

AmpSdk.run/2 blocks until the agent finishes, returning the final result text.

2. Stream Responses in Real Time

alias AmpSdk.Types.{AssistantMessage, ResultMessage, SystemMessage}

"Explain the architecture of this project"
|> AmpSdk.execute()
|> Enum.each(fn
  %SystemMessage{tools: tools} ->
    IO.puts("Session started with #{length(tools)} tools")

  %AssistantMessage{message: %{content: content}} ->
    for %{type: "text", text: text} <- content do
      IO.write(text)
    end

  %ResultMessage{result: result, duration_ms: ms, num_turns: turns} ->
    IO.puts("\n--- Done in #{ms}ms (#{turns} turns) ---")

  _other ->
    :ok
end)

AmpSdk.execute/2 returns a lazy Stream -- messages arrive as the agent works, and the stream halts automatically when a result or error is received. Under the hood, the public stream surface is projected from a shared core session runtime, while cleanup still drains internal runtime messages so finished or timed-out streams do not leave residual events in the caller mailbox.

3. Continue a Thread

alias AmpSdk.Types.Options

# First interaction
"Add input validation to the User module"
|> AmpSdk.execute(%Options{visibility: "private"})
|> Enum.each(&handle_message/1)

# Continue the same thread
"Now add tests for the validation we just added"
|> AmpSdk.execute(%Options{continue_thread: true})
|> Enum.each(&handle_message/1)

# Or continue a specific thread by ID
"Review the changes"
|> AmpSdk.execute(%Options{continue_thread: "T-abc123-def456"})
|> Enum.each(&handle_message/1)

Core API

AmpSdk.execute/2

Streams messages from the Amp agent as a lazy Enumerable.

@spec execute(String.t() | [AmpSdk.Types.UserInputMessage.t() | map()], Options.t()) ::
        Enumerable.t(stream_message())

Messages are yielded in order as the agent works:

1. SystemMessage -- session init with available tools and MCP server status
2. AssistantMessage -- agent responses (text blocks and/or tool calls)
3. UserMessage -- tool results fed back to the agent
4. ResultMessage or ErrorResultMessage -- final outcome (stream halts)

AmpSdk.run/2

Convenience wrapper that collects the stream and returns the final result:

@spec run(String.t(), Options.t()) :: {:ok, String.t()} | {:error, AmpSdk.Error.t()}

{:ok, answer} = AmpSdk.run("How many modules are in lib/?")
{:error, reason} = AmpSdk.run("Do something impossible")

AmpSdk.create_user_message/1

Creates a UserInputMessage struct for JSON-input streaming:

msgs = [
  AmpSdk.create_user_message("Summarize the last change and suggest next steps.")
]

msgs
|> AmpSdk.execute()
|> Enum.to_list()

AmpSdk.create_permission/3

Creates a Permission struct for tool access control:

perm = AmpSdk.create_permission("Bash", "allow")
perm = AmpSdk.create_permission("Bash", "delegate", to: "bash -c")
perm = AmpSdk.create_permission("Read", "ask", matches: %{"path" => "/secret/*"})

AmpSdk.threads_new/1 and AmpSdk.threads_markdown/1

Manage threads directly:

{:ok, thread_id} = AmpSdk.threads_new(visibility: :private)
{:ok, markdown}   = AmpSdk.threads_markdown(thread_id)

Centralized Model Selection

amp_sdk now renders model arguments from the shared cli_subprocess_core model-selection payload. It no longer infers local model defaults or fallback policy inside the Amp repo.

Authoritative policy surface:

• CliSubprocessCore.ModelRegistry.resolve/3
• CliSubprocessCore.ModelRegistry.validate/2
• CliSubprocessCore.ModelRegistry.default_model/2
• CliSubprocessCore.ModelRegistry.build_arg_payload/3

Amp-side responsibility is transport-only:

• carry the resolved payload through AmpSdk.Types.Options
• render model flags from that resolved payload only
• never emit blank or placeholder model values

Amp intentionally does not expose a second raw model-selection surface in this repo today. AmpSdk.Types.Options.validate!/1 canonicalizes a supplied payload when present, but if no payload was provided the Amp SDK does not invent an extra model fallback path of its own.

When callers already have a serialized selection map, Map.from_struct(payload) is normalized back into the canonical CliSubprocessCore.ModelRegistry.Selection while preserving forward-compatible extra fields.

Typed Management List APIs

Management list functions return typed data for programmatic use:

{:ok, threads} = AmpSdk.threads_list()
{:ok, rules} = AmpSdk.permissions_list()
{:ok, servers} = AmpSdk.mcp_list()

AmpSdk.Types.PermissionRule and AmpSdk.Types.MCPServer are schema-backed: known JSON fields are normalized through Zoi, forward-compatible unknown fields are preserved in extra, and to_map/1 projects them back to wire form.

The same schema-backed contract now applies to the public stream message structs and their nested content blocks.

Configuration Options

All execution behavior is controlled through AmpSdk.Types.Options:

%AmpSdk.Types.Options{
  cwd: "/path/to/project",           # Working directory (default: cwd)
  mode: "smart",                      # Agent mode (see table below)
  dangerously_allow_all: false,       # Skip all permission prompts
  visibility: "workspace",            # Thread visibility
  continue_thread: nil,               # true | "thread-id" | nil
  settings_file: nil,                 # Path to settings.json
  log_level: nil,                     # "debug" | "info" | "warn" | "error" | "audit"
  log_file: nil,                      # Log file path
  env: %{},                           # Extra environment variables
  mcp_config: nil,                    # MCP server configuration (map or JSON string)
  toolbox: nil,                       # Path to toolbox scripts
  skills: nil,                        # Path to custom skills
  permissions: nil,                   # List of Permission structs
  labels: nil,                        # Thread labels (max 20, alphanumeric + hyphens)
  thinking: false,                    # Use --stream-json-thinking when prompt is a string
  model_payload: nil,                 # Shared core Selection (or a canonicalizable map form)
  execution_surface: nil,             # Optional ExecutionSurface struct, map, or keyword
  stream_timeout_ms: 300_000,         # Receive timeout for stream events
  no_ide: false,                      # Disable IDE context injection
  no_notifications: false,            # Disable notification sounds
  no_color: false,                    # Disable ANSI colors
  no_jetbrains: false                 # Disable JetBrains integration
}

Agent Modes

Note: Only "smart" mode supports --stream-json, which the SDK requires. Other modes can only be used via the CLI directly.

Thread Visibility

Permissions

Fine-grained control over which tools the agent can use:

alias AmpSdk.Types.{Options, Permission}

permissions = [
  # Allow file reads without prompting
  AmpSdk.create_permission("Read", "allow"),

  # Ask before running shell commands
  AmpSdk.create_permission("Bash", "ask"),

  # Block file deletion entirely
  AmpSdk.create_permission("Bash", "reject",
    matches: %{"cmd" => ["rm *", "rmdir *"]}
  ),

  # Only ask in subagent context
  AmpSdk.create_permission("edit_file", "ask", context: "subagent")
]

"Refactor the auth module"
|> AmpSdk.execute(%Options{permissions: permissions, dangerously_allow_all: false})
|> Enum.each(&handle_message/1)

Permission Actions

Permissions are written to a temporary settings.json that is passed to the CLI via --settings-file and cleaned up after execution.

MCP Server Integration

Configure Model Context Protocol servers to extend the agent's capabilities:

alias AmpSdk.Types.Options

# Stdio-based MCP server
mcp_config = %{
  filesystem: %{
    command: "npx",
    args: ["-y", "@modelcontextprotocol/server-filesystem", "/tmp"],
    env: %{}
  }
}

"List all markdown files using the filesystem MCP tool"
|> AmpSdk.execute(%Options{mcp_config: mcp_config})
|> Enum.each(&handle_message/1)

# HTTP-based MCP server
mcp_config = %{
  remote_api: %{
    url: "https://api.example.com/mcp",
    headers: %{"Authorization" => "Bearer token"}
  }
}

MCP server connection status is reported in the initial SystemMessage:

%SystemMessage{mcp_servers: [%{name: "filesystem", status: "connected"}]}

Possible statuses: "awaiting-approval", "authenticating", "connecting", "reconnecting", "connected", "denied", "failed", "blocked-by-registry".

Thread Management

Threads persist conversation history on Amp's servers. Use them for multi-step workflows:

# Create a new thread
{:ok, thread_id} = AmpSdk.threads_new(visibility: :private)

# Run against it
"Analyze the codebase"
|> AmpSdk.execute(%Options{continue_thread: thread_id})
|> Enum.each(&handle_message/1)

# Continue the same thread later
"Now implement the changes we discussed"
|> AmpSdk.execute(%Options{continue_thread: thread_id})
|> Enum.each(&handle_message/1)

# Export conversation as markdown
{:ok, md} = AmpSdk.threads_markdown(thread_id)
File.write!("thread_export.md", md)

Stream Message Types

Every message from execute/2 is one of these structs:

These message structs and their nested content blocks are schema-backed: known fields are normalized through Zoi, forward-compatible unknown fields are preserved in extra, and to_map/1 projects them back to wire shape.

SystemMessage

First message in every session. Contains session metadata.

%SystemMessage{
  type: "system",
  subtype: "init",
  session_id: "T-...",
  cwd: "/path/to/project",
  tools: ["Bash", "Read", "edit_file", "glob", ...],
  mcp_servers: [%MCPServerStatus{name: "fs", status: "connected"}]
}

AssistantMessage

Agent responses. Content is a list of text blocks and/or tool calls.

%AssistantMessage{
  type: "assistant",
  session_id: "T-...",
  message: %{
    role: "assistant",
    model: "claude-sonnet-4-5-20250929",
    content: [
      %TextContent{type: "text", text: "I'll read the file..."},
      %ToolUseContent{type: "tool_use", id: "tu_1", name: "Read", input: %{"path" => "lib/app.ex"}}
    ],
    stop_reason: "tool_use",
    usage: %Usage{input_tokens: 1024, output_tokens: 256, ...}
  }
}

UserMessage

Tool results fed back to the agent automatically.

%UserMessage{
  type: "user",
  message: %{
    role: "user",
    content: [
      %ToolResultContent{type: "tool_result", tool_use_id: "tu_1", content: "...", is_error: false}
    ]
  }
}

ResultMessage

Successful completion. Includes total usage and timing.

%ResultMessage{
  type: "result",
  subtype: "success",
  is_error: false,
  result: "I've updated the module with...",
  duration_ms: 12450,
  num_turns: 3,
  usage: %Usage{input_tokens: 8192, output_tokens: 2048},
  permission_denials: nil
}

ErrorResultMessage

Execution failed or hit max turns.

%ErrorResultMessage{
  type: "result",
  subtype: "error_during_execution",  # or "error_max_turns"
  is_error: true,
  error: "Failed to complete the task",
  duration_ms: 5000,
  num_turns: 1,
  permission_denials: ["Bash: rm -rf /"]
}

Architecture

┌──────────────────────────────────────────────────────┐
│                  AmpSdk (Public API)                 │
│                                                      │
│  execute/2 ── stream messages from agent             │
│  run/2     ── blocking call, returns final result    │
│  threads_*/N wrappers for thread lifecycle ops        │
│  create_permission/3, create_user_message/1          │
└──────────────────────┬───────────────────────────────┘
                       │
┌──────────────────────▼───────────────────────────────┐
│             AmpSdk.Stream (Stream Engine)            │
│                                                      │
│  - Wraps execution as Stream.resource/3              │
│  - Starts Amp sessions through the shared core lane  │
│  - Projects core events into AmpSdk typed messages   │
│  - Captures stderr tail + cleanup metadata           │
│  - Halts on ResultMessage / ErrorResultMessage       │
└──────────────────────┬───────────────────────────────┘
                       │
┌──────────────────────▼───────────────────────────────┐
│            AmpSdk.Runtime.CLI (Session Kit)          │
│                                                      │
│  - Resolves the Amp CLI binary and invocation        │
│  - Builds Amp-compatible args, env, settings files   │
│  - Starts shared core provider sessions              │
│  - Projects provider events back to public structs   │
└──────────────────────┬───────────────────────────────┘
                       │
┌──────────────────────▼───────────────────────────────┐
│              AmpSdk.Command (One-Shot API)          │
│                                                      │
│  - Resolves Amp CLI command specs                    │
│  - Preserves Amp-specific public result/error shape  │
│  - Delegates non-PTY execution to the shared core    │
└──────────────────────┬───────────────────────────────┘
                       │
┌──────────────────────▼───────────────────────────────┐
│      cli_subprocess_core (Shared CLI Runtime)        │
│                                                      │
│  - Session lifecycle and provider parsing            │
│  - Shared non-PTY command execution                  │
│  - Shared raw transport implementation               │
│  - Common task supervision and subprocess handling   │
└──────────────────────┬───────────────────────────────┘
                       │
               ┌───────▼───────┐
               │   Amp CLI     │
               │  (headless)   │
               └───────────────┘

Module Overview

Ownership Boundary

The final Phase 4 boundary for Amp is:

• AmpSdk.Stream, AmpSdk.Runtime.CLI, and AmpSdk.Command now sit above the shared cli_subprocess_core session and command lanes
• no separate Amp-owned transport wrapper or common subprocess runtime remains in this repo

Repo-local ownership is limited to the Amp-facing CLI/runtime wrapper, execution-surface preservation, Amp-specific option and environment shaping, typed message/result projection, and the public thread/permission/MCP management helpers.

The release and composition model is:

• the common Amp profile stays built into cli_subprocess_core
• amp_sdk remains the provider-specific runtime-kit package above that shared core
• no separate ASM extension seam is introduced unless Amp later proves a real richer provider-native surface that should sit above the normalized kernel

If amp_sdk is installed alongside agent_session_manager, ASM reports Amp runtime availability in ASM.Extensions.ProviderSDK.capability_report/0 but keeps namespaces: [] because Amp currently composes through the common ASM surface only.

Error Handling

AmpSdk.run/2 is tuple-based and returns %AmpSdk.Error{} on failures:

case AmpSdk.run("do something") do
  {:ok, result} ->
    IO.puts(result)

  {:error, %AmpSdk.Error{kind: :no_result}} ->
    IO.puts("No result received")

  {:error, %AmpSdk.Error{kind: kind, message: message}} ->
    IO.puts("#{kind}: #{message}")
end

Internal timeout/task helpers also normalize into %AmpSdk.Error{} kinds (for example :task_timeout).

Streaming failures are surfaced inline as ErrorResultMessage structs:

"bad prompt"
|> AmpSdk.execute()
|> Enum.each(fn
  %ErrorResultMessage{error: error, permission_denials: denials} ->
    IO.puts("Error: #{error}")
    if denials, do: IO.puts("Denied: #{inspect(denials)}")

  _msg -> :ok
end)

Execution and transport failures are normalized directly into %AmpSdk.Error{} or inline ErrorResultMessage structs. If you need the shared-core transport envelope itself, use AmpSdk.Error.normalize/2 against the returned reason tuple or exception. Subprocess lifecycle, exit handling, retained-stderr capture, and execution-surface routing are defined in cli_subprocess_core.

Environment Variables

AmpSdk.run/2 and AmpSdk.execute/2 use the same CLI env builder: base system keys (PATH, HOME, etc.), AMP_* keys, Options.env overrides, and automatic AMP_SDK_VERSION tagging.

Additional env vars can be passed per-execution via Options.env:

AmpSdk.run("check env", %Options{env: %{"MY_VAR" => "value"}})

nil values in Options.env and MCP constructor maps (env, headers) are dropped during normalization.

For MCP config constructors that take maps/keywords, use atom keys. String keys are ignored by those constructors.

Examples

Mix Task

# lib/mix/tasks/amp.ex
defmodule Mix.Tasks.Amp do
  use Mix.Task

  @shortdoc "Run an Amp query against the current project"

  def run([prompt | _]) do
    Mix.Task.run("app.start")

    case AmpSdk.run(prompt) do
      {:ok, result} -> Mix.shell().info(result)
      {:error, %AmpSdk.Error{kind: kind, message: message}} ->
        Mix.shell().error("[#{kind}] #{message}")
    end
  end
end

mix amp "What does this project do?"

Streaming with Progress

alias AmpSdk.Types.{AssistantMessage, ResultMessage, ErrorResultMessage, SystemMessage}

defmodule MyApp.AmpRunner do
  def run_with_progress(prompt, opts \\ %AmpSdk.Types.Options{}) do
    prompt
    |> AmpSdk.execute(opts)
    |> Enum.reduce(%{text: "", turns: 0}, fn
      %SystemMessage{session_id: id, tools: tools}, acc ->
        IO.puts("[session #{id}] #{length(tools)} tools available")
        acc

      %AssistantMessage{message: %{content: content}}, acc ->
        text = content
          |> Enum.filter(&match?(%{type: "text"}, &1))
          |> Enum.map(& &1.text)
          |> Enum.join()
        IO.write(text)
        %{acc | text: acc.text <> text}

      %ResultMessage{duration_ms: ms, num_turns: turns}, acc ->
        IO.puts("\nCompleted in #{ms}ms (#{turns} turns)")
        %{acc | turns: turns}

      %ErrorResultMessage{error: error}, acc ->
        IO.puts("\nError: #{error}")
        acc

      _, acc -> acc
    end)
  end
end

Automated Code Review

alias AmpSdk.Types.Options

permissions = [
  AmpSdk.create_permission("Read", "allow"),
  AmpSdk.create_permission("glob", "allow"),
  AmpSdk.create_permission("Grep", "allow"),
  AmpSdk.create_permission("Bash", "reject"),
  AmpSdk.create_permission("edit_file", "reject"),
  AmpSdk.create_permission("create_file", "reject")
]

{:ok, review} = AmpSdk.run(
  "Review the code in lib/ for bugs, security issues, and style problems. Be thorough.",
  %Options{
    mode: "smart",
    permissions: permissions,
    visibility: "private"
  }
)

IO.puts(review)

Multi-Step Workflow with Thread Continuity

alias AmpSdk.Types.Options

opts = %Options{visibility: "private", dangerously_allow_all: true}

# Step 1: Analyze
{:ok, analysis} = AmpSdk.run("Analyze lib/my_app/auth.ex for improvements", opts)
IO.puts(analysis)

# Step 2: Implement (same thread)
{:ok, changes} = AmpSdk.run(
  "Implement the improvements you identified",
  %Options{opts | continue_thread: true}
)
IO.puts(changes)

# Step 3: Test
{:ok, tests} = AmpSdk.run(
  "Write tests for the changes you made",
  %Options{opts | continue_thread: true}
)
IO.puts(tests)

Documentation

Full API documentation is available on HexDocs.

Guides

• Getting Started — installation, authentication, first query
• Configuration — all options, modes, MCP, environment variables
• Streaming — message types, real-time output patterns
• Permissions — tool access control and safety
• Threads — multi-turn conversations and thread management
• Error Handling — error kinds and recovery
• Testing — unit and integration testing strategies

Examples

See examples/ for runnable scripts. Run all with:

./examples/run_all.sh

Generate Docs Locally

mix docs
open doc/index.html

License

MIT -- see LICENSE for details.

Acknowledgments

• Sourcegraph for the Amp coding agent and CLI
• Sasa Juric for the native process-management foundation used underneath cli_subprocess_core
• Built to complement claude_agent_sdk and codex_sdk for multi-agent Elixir workflows

Related Projects

Model Selection Contract

/home/home/p/g/n/amp_sdk now renders model arguments from payloads resolved by /home/home/p/g/n/cli_subprocess_core. The only authoritative model-policy path is CliSubprocessCore.ModelRegistry.resolve/3, CliSubprocessCore.ModelRegistry.validate/2, and CliSubprocessCore.ModelRegistry.default_model/2.

Amp transport code remains responsible for formatting CLI arguments only. It does not implement provider fallback policy and must not emit nil/null/blank --model values.

Thread History And Resume Surfaces

Amp remains thread-oriented, but the runtime layer now publishes that history through the same session-control vocabulary used elsewhere in the CLI stack.

• AmpSdk.Runtime.CLI.capabilities/0 includes :session_history, :session_resume, :session_pause, and :session_intervene
• AmpSdk.Runtime.CLI.list_provider_sessions/1 projects threads_list/1 entries into a common history shape for orchestration layers

This repo does not advertise a synthetic system_prompt surface through that runtime-neutral path. If a higher layer needs prompt intervention, it must use honest Amp thread controls rather than assume Claude/Gemini-style system prompt support exists here.