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
• guides/migrating-to-0.6.md - compatibility and migration notes for 0.6
• guides/provider_behavior_manifest.md - evidence for Amp-native feature translation

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.6.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 pass an API key explicitly to the launched CLI environment:

AmpSdk.run("Summarize this repository", %AmpSdk.Types.Options{
  env: %{"AMP_API_KEY" => "your-api-key"}
})

Host applications that intentionally source credentials from the OS environment should materialize those values in config/runtime.exs or another configuration boundary before calling the SDK.

These authentication paths are standalone direct-use compatibility. Governed execution does not treat amp login, AMP_API_KEY, native Amp config, or MCP OAuth state as authority; callers must pass an explicit governed authority that materializes command, cwd, env, target, credential lease, command reference, and redaction reference for the effect.

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.

SDK-direct live verification lives in examples/promotion_path/sdk_direct_amp.exs. It uses the Amp SDK API without importing ASM, passes keyword execution_surface input, and demonstrates Amp-native permission settings and UI suppression flags as SDK-owned behavior:

mix run examples/promotion_path/sdk_direct_amp.exs \
  --prompt "Reply with exactly: amp sdk direct ok"

Provider-native feature evidence is tracked in guides/provider_behavior_manifest.md. Add or update that manifest before translating any new Amp-specific setting, permission, MCP, task, skill, review, thread, or management behavior.

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: explicit Options.env overrides, optional :amp_sdk, :base_env application config or caller-supplied base env maps, and automatic AMP_SDK_VERSION tagging. Provider behavior keys such as AMP_API_KEY, AMP_URL, and AMP_TOOLBOX are not ambiently inherited by the runtime path; pass them explicitly through Options.env or the typed option when applicable.

In governed execution, Options.env, AMP_CLI_PATH, AMP_API_KEY, AMP_URL, settings files, permissions settings, MCP config, and MCP OAuth credentials are rejected as unmanaged authority inputs. The launched process receives only the authority-materialized env with clear_env?: true.

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
• Migrating to 0.6 — compatibility and required changes
• Provider Behavior Manifest — evidence for Amp-native feature translation

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.