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.

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.1.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:

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. Transport-tagged mailbox events are drained during cleanup, so finished/timeout streams do not leave residual transport messages 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)

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
  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:

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)            │
│                                                      │
│  - Builds CLI args from Options struct               │
│  - Creates temp settings.json for permissions/skills │
│  - Wraps execution as Stream.resource/3              │
│  - Parses JSON lines into typed structs              │
│  - Halts on ResultMessage / ErrorResultMessage       │
└──────────────────────┬───────────────────────────────┘
                       │
┌──────────────────────▼───────────────────────────────┐
│     AmpSdk.Transport.Erlexec (GenServer + erlexec)   │
│                                                      │
│  - Spawns `amp --execute --stream-json` subprocess   │
│  - Manages stdin/stdout/stderr via erlexec           │
│  - Splits stdout into JSON lines                     │
│  - Broadcasts lines to subscribers via messages      │
│  - Handles buffer overflow, process exit, cleanup    │
└──────────────────────┬───────────────────────────────┘
                       │
               ┌───────▼───────┐
               │   Amp CLI     │
               │  (headless)   │
               └───────────────┘

Module Overview

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

Exceptional conditions (for example invalid settings JSON) can still raise typed errors such as AmpSdk.Errors.AmpError.

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)

Low-level transport APIs (AmpSdk.Transport.Erlexec) return tagged tuples like {:error, {:transport, reason}}; use AmpSdk.Transport.error_to_error/2 (or AmpSdk.Error.normalize/2) when you want the unified envelope there as well.

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, conflicting atom/string versions of the same key are rejected as :invalid_configuration to avoid ambiguous input maps.

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 — exception types 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 erlexec, the backbone of subprocess management
• Built to complement claude_agent_sdk and codex_sdk for multi-agent Elixir workflows

Related Projects