Build an AI Agent in Elixir: Multi-Step Tool Loops with OTP

Most “AI feature” tutorials stop at a single round trip. You send a prompt, the model answers, you render the answer. That covers a surprising amount of ground, and if it covers your use case you should stop there.

An agent is a different animal. An agent runs a loop: the model looks at the task, picks a tool, you execute it, you feed the result back, and the model decides what to do next. It might do that twice. It might do it fifteen times. It might get stuck in a circle and burn forty dollars of tokens before anyone notices.

That loop is a long-lived, stateful, failure-prone, concurrent process that needs supervision, cancellation, budgets and observability. Which is a fairly precise description of the thing the BEAM was built for. If you want to build an AI agent in Elixir, you are starting from an unusually good position, and this post walks through the whole thing: tool definitions, the loop, a supervised runner, live progress in LiveView, hard budget caps, human approval for dangerous actions, and how to test all of it without spending a cent on API calls.

If you have not wired up a model client yet, start with Claude API in Elixir, which covers the request shape, streaming and structured output that this post assumes.

An agent is a loop, not a prompt

Strip away the marketing and an agent is about twelve lines of pseudocode:

messages = [user_task]
loop do
  reply = model(messages, tools)
  messages = messages ++ [reply]
  if reply.stop_reason != "tool_use", do: return reply.text
  results = for call <- reply.tool_calls, do: execute(call)
  messages = messages ++ [results]
end

Everything hard about agents lives in the words “loop” and “execute”. The loop needs a termination condition that does not depend on the model behaving. The execute step reaches into your database and your payment provider, so it needs authorisation that the model cannot influence. Get those two right and the rest is plumbing.

This is genuinely different from the chatbot pattern. A LiveView chatbot is turn-based: a human is in the loop after every response, which is a very effective safety mechanism. An agent removes the human from most iterations, so you have to replace them with code.

Why OTP fits this shape

Four things an agent run needs, and what the BEAM already gives you:

1. A place to live. An agent run is state that outlives a single request and is not tied to a browser tab. That is a GenServer under a DynamicSupervisor, with a Registry so any part of the app can find a run by id.
2. Cancellation. Users close tabs and change their minds. Killing a process is a first-class operation here, not a cooperative flag you have to check in a while loop.
3. Isolation. One run raising in a tool call should not touch the other forty runs in flight. Separate processes give you that for free.
4. Concurrency without ceremony. Two hundred concurrent agent runs, each blocked on a slow HTTP call, is a boring workload for the BEAM. In a threaded runtime it is an architecture decision.

The part OTP does not solve is that agent state is in memory, so a deploy loses it. We will come back to that.

Step 1: describe your tools once, as data

Every tool needs a JSON schema for the model and an implementation for you. Keep them in the same module so they cannot drift apart.

defmodule MyApp.Agent.Tools do
  @moduledoc "Every tool the agent may call, described once."

  def schemas do
    [
      %{
        name: "search_orders",
        description:
          "Find orders belonging to a customer, by email address. " <>
            "Returns at most 20 rows, newest first.",
        input_schema: %{
          type: "object",
          properties: %{
            email: %{type: "string", description: "Customer email address"},
            status: %{type: "string", enum: ["open", "shipped", "refunded"]}
          },
          required: ["email"]
        }
      },
      %{
        name: "refund_order",
        description:
          "Refund an order in full. This moves real money and cannot be undone. " <>
            "Only call this when the customer has explicitly asked for a refund.",
        input_schema: %{
          type: "object",
          properties: %{
            order_id: %{type: "integer"},
            reason: %{type: "string", description: "Short human-readable reason"}
          },
          required: ["order_id", "reason"]
        }
      }
    ]
  end

  @doc "Tools that need a human to sign off before they run."
  def approval("refund_order"), do: :always
  def approval(_name), do: :never

  def run("search_orders", %{"email" => email} = args, ctx) do
    orders =
      ctx.account_id
      |> MyApp.Orders.search(email, args["status"])
      |> Enum.take(20)
      |> Enum.map(&Map.take(&1, [:id, :status, :total_cents, :placed_at]))

    {:ok, orders}
  end

  def run("refund_order", %{"order_id" => id, "reason" => reason}, ctx) do
    case MyApp.Orders.refund(ctx.account_id, id, reason) do
      {:ok, order} -> {:ok, %{refunded: true, order_id: order.id}}
      {:error, changeset} -> {:error, "refund failed: #{inspect(changeset.errors)}"}
    end
  end

  def run(name, _args, _ctx), do: {:error, "unknown tool: #{name}"}
end

Two details in there are the whole security model.

The account id comes from ctx, never from the model. ctx is built server side from the authenticated session before the run starts. If account_id were a tool parameter, a prompt injection in a customer support email would be able to read any account in your database. Treat every value the model produces as untrusted user input, because that is exactly what it is.

Descriptions are prompt engineering. “Refund an order” gets called speculatively. “This moves real money and cannot be undone. Only call this when the customer has explicitly asked” gets called far less often. The schema is where you spend your tuning effort.

Step 2: one model call

Nothing exotic, just Req against the Messages API.

defmodule MyApp.Agent.Model do
  @endpoint "https://api.anthropic.com/v1/messages"

  def next(messages, tools, opts \\ []) do
    body = %{
      model: Keyword.get(opts, :model, "claude-sonnet-5"),
      max_tokens: 2048,
      system: system_prompt(),
      tools: tools,
      messages: messages
    }

    req =
      Req.new(
        url: @endpoint,
        json: body,
        headers: [
          {"x-api-key", Application.fetch_env!(:my_app, :anthropic_key)},
          {"anthropic-version", "2023-06-01"}
        ],
        receive_timeout: 120_000,
        retry: :transient,
        max_retries: 3
      )
      |> Req.merge(Application.get_env(:my_app, :agent_req_options, []))

    case Req.post(req) do
      {:ok, %{status: 200, body: body}} -> {:ok, body}
      {:ok, %{status: status, body: body}} -> {:error, {:http_error, status, body}}
      {:error, reason} -> {:error, reason}
    end
  end

  defp system_prompt do
    """
    You are a support agent for an online store. Work step by step.
    Use tools to find facts instead of guessing. If a tool returns an
    error, explain it plainly rather than retrying the same call.
    When the task is complete, answer in two sentences or fewer.
    """
  end
end

The Req.merge(Application.get_env(...)) line looks pointless right now. It is the seam that makes this testable without a mocking library, and we use it at the end of the post.

Step 3: the loop, as a function you can call in a test

Keep the loop out of the GenServer. A function that takes state and returns {:continue, state}, {:done, state, answer} or {:halted, reason} is trivial to test. A GenServer that does the same work is not.

defmodule MyApp.Agent.Loop do
  alias MyApp.Agent.{Budget, Model, Tools}

  def step(%{messages: messages, budget: budget} = state) do
    with :ok <- Budget.check(budget),
         {:ok, reply} <- Model.next(messages, Tools.schemas()) do
      state =
        state
        |> Map.update!(:budget, &Budget.spend(&1, reply["usage"]))
        |> append(%{role: "assistant", content: reply["content"]})

      case reply["stop_reason"] do
        "tool_use" -> handle_tool_use(state, reply["content"])
        _other -> {:done, state, text_of(reply["content"])}
      end
    else
      {:error, reason} -> {:halted, reason}
    end
  end

  defp handle_tool_use(state, content) do
    calls = Enum.filter(content, &(&1["type"] == "tool_use"))

    case Enum.find(calls, &(Tools.approval(&1["name"]) == :always)) do
      nil -> {:continue, append(state, %{role: "user", content: run_all(calls, state.ctx)})}
      call -> {:awaiting_approval, state, call}
    end
  end

  defp run_all(calls, ctx) do
    calls
    |> Task.async_stream(&result_block(&1, ctx), timeout: 30_000, on_timeout: :kill_task)
    |> Enum.zip(calls)
    |> Enum.map(fn
      {{:ok, block}, _call} -> block
      {{:exit, :timeout}, call} -> error_block(call, "tool timed out after 30s")
    end)
  end

  defp result_block(%{"id" => id, "name" => name, "input" => args}, ctx) do
    case safe_run(name, args, ctx) do
      {:ok, value} ->
        %{type: "tool_result", tool_use_id: id, content: Jason.encode!(value)}

      {:error, message} ->
        %{type: "tool_result", tool_use_id: id, is_error: true, content: message}
    end
  end

  defp error_block(%{"id" => id}, message),
    do: %{type: "tool_result", tool_use_id: id, is_error: true, content: message}

  defp safe_run(name, args, ctx) do
    Tools.run(name, args, ctx)
  rescue
    e -> {:error, "tool raised: " <> Exception.message(e)}
  end

  defp append(state, message), do: Map.update!(state, :messages, &(&1 ++ [message]))

  defp text_of(content) do
    content
    |> Enum.filter(&(&1["type"] == "text"))
    |> Enum.map_join("\n", & &1["text"])
  end
end

Three choices worth calling out.

Tool errors go back to the model, they do not crash the run. A tool that returns is_error: true gives the model a chance to correct itself, which it is often quite good at. A tool that raises and kills the process gives the user nothing. safe_run/3 turns the second into the first.

Parallel tool calls. Models regularly emit three tool_use blocks in one reply. Task.async_stream runs them concurrently, which is roughly a one-line change in Elixir and a small project in most other runtimes. The on_timeout: :kill_task branch matters: a hanging tool must not hang the run.

Ordering. The results are zipped back against the original calls, because the API requires a tool_result for every tool_use in the preceding message. Drop one and the next request is rejected.

Step 4: budgets, which is the step everyone skips

An agent without a hard budget is an open invoice. The model decides how many iterations to run, and models sometimes loop. Do not rely on the prompt for this.

defmodule MyApp.Agent.Budget do
  defstruct steps: 0,
            max_steps: 12,
            input_tokens: 0,
            output_tokens: 0,
            cents: 0.0,
            max_cents: 50.0

  def check(%__MODULE__{} = b) do
    cond do
      b.steps >= b.max_steps -> {:error, {:budget, :step_limit}}
      b.cents >= b.max_cents -> {:error, {:budget, :cost_limit}}
      true -> :ok
    end
  end

  def spend(%__MODULE__{} = b, %{"input_tokens" => inp, "output_tokens" => out}) do
    %{
      b
      | steps: b.steps + 1,
        input_tokens: b.input_tokens + inp,
        output_tokens: b.output_tokens + out,
        cents: b.cents + cost_cents(inp, out)
    }
  end

  def spend(b, _missing_usage), do: %{b | steps: b.steps + 1}

  # Rates per million tokens, in cents. Keep them in config, not here.
  defp cost_cents(inp, out), do: inp / 1_000_000 * 300 + out / 1_000_000 * 1500
end

Twelve steps and fifty cents are starting points, not answers. Log the distribution of steps and cents across real runs for a week, then set the cap somewhere above the 99th percentile. The point of the cap is to stop pathological runs, not to interrupt normal ones.

Once you are tracking spend per run, you are one short step from tracking it per customer, which is where it stops being an engineering concern and starts being a pricing one. If you plan to bill for agent usage, Aurora Meter meters events on an ETS hot path and reports them to Stripe, so you are not building a metering pipeline as a side quest.

Step 5: a supervised runner

Now wrap the loop. Add these to your supervision tree:

# lib/my_app/application.ex
children = [
  {Registry, keys: :unique, name: MyApp.AgentRegistry},
  {DynamicSupervisor, name: MyApp.AgentSupervisor, strategy: :one_for_one}
]

And the runner itself:

defmodule MyApp.Agent.Run do
  use GenServer, restart: :transient

  alias MyApp.Agent.{Budget, Loop}

  def start(run_id, ctx, task) do
    DynamicSupervisor.start_child(
      MyApp.AgentSupervisor,
      %{id: {__MODULE__, run_id}, start: {__MODULE__, :start_link, [{run_id, ctx, task}]},
        restart: :transient}
    )
  end

  def cancel(run_id), do: GenServer.cast(via(run_id), :cancel)
  def approve(run_id, tool_use_id), do: GenServer.call(via(run_id), {:approve, tool_use_id})

  def start_link({run_id, _ctx, _task} = arg),
    do: GenServer.start_link(__MODULE__, arg, name: via(run_id))

  defp via(run_id), do: {:via, Registry, {MyApp.AgentRegistry, run_id}}

  @impl true
  def init({run_id, ctx, task}) do
    state = %{
      run_id: run_id,
      ctx: ctx,
      budget: %Budget{},
      messages: [%{role: "user", content: task}],
      status: :running
    }

    send(self(), :tick)
    {:ok, state}
  end

  @impl true
  def handle_info(:tick, state) do
    case Loop.step(state) do
      {:continue, state} ->
        emit(state, {:progress, state.budget.steps})
        send(self(), :tick)
        {:noreply, state}

      {:done, state, answer} ->
        emit(state, {:done, answer})
        {:stop, :normal, state}

      {:awaiting_approval, state, call} ->
        emit(state, {:approval_required, call})
        {:noreply, %{state | status: {:awaiting, call}}}

      {:halted, reason} ->
        emit(state, {:halted, reason})
        {:stop, :normal, state}
    end
  end

  @impl true
  def handle_cast(:cancel, state) do
    emit(state, {:halted, :cancelled})
    {:stop, :normal, state}
  end

  @impl true
  def handle_call({:approve, id}, _from, %{status: {:awaiting, %{"id" => id}}} = state) do
    send(self(), :tick)
    {:reply, :ok, %{state | status: :running}}
  end

  def handle_call({:approve, _id}, _from, state), do: {:reply, {:error, :stale}, state}

  defp emit(state, event),
    do: Phoenix.PubSub.broadcast(MyApp.PubSub, "agent:#{state.run_id}", event)
end

One deliberate choice: the loop advances with send(self(), :tick) rather than {:continue, :tick}. handle_continue runs before any queued messages, so a run that chains continues is deaf to a cancel cast until it finishes. Sending yourself a normal message puts the tick at the back of the mailbox, so cancellations and approvals land between iterations. It is a small thing that turns cancellation from “eventually” into “immediately”.

restart: :transient means a crashed run restarts, and a run that stops with :normal does not. Note that a restarted run loses its message history and starts from init/1. Fixing that properly means persisting messages, covered below.

Step 6: streaming progress into LiveView

The runner broadcasts, the LiveView subscribes. The browser tab is now just a viewer, so closing it does not kill the run and reopening it does not start a second one.

defmodule MyAppWeb.AgentLive do
  use MyAppWeb, :live_view

  alias MyApp.Agent.Run

  @impl true
  def mount(%{"id" => run_id}, _session, socket) do
    if connected?(socket), do: Phoenix.PubSub.subscribe(MyApp.PubSub, "agent:#{run_id}")

    {:ok, assign(socket, run_id: run_id, steps: 0, answer: nil, pending: nil, error: nil)}
  end

  @impl true
  def handle_info({:progress, steps}, socket), do: {:noreply, assign(socket, steps: steps)}
  def handle_info({:done, answer}, socket), do: {:noreply, assign(socket, answer: answer)}
  def handle_info({:approval_required, call}, socket), do: {:noreply, assign(socket, pending: call)}
  def handle_info({:halted, reason}, socket), do: {:noreply, assign(socket, error: reason)}

  @impl true
  def handle_event("approve", _params, socket) do
    Run.approve(socket.assigns.run_id, socket.assigns.pending["id"])
    {:noreply, assign(socket, pending: nil)}
  end

  def handle_event("cancel", _params, socket) do
    Run.cancel(socket.assigns.run_id)
    {:noreply, socket}
  end
end

The approval gate is worth taking seriously. Read-only tools can run unattended all day. Anything that spends money, sends email, or deletes rows should pause and show a human exactly which tool is about to run with exactly which arguments. That is one Tools.approval/1 clause per dangerous tool, and it is the difference between a demo and something you can point at production data.

Step 7: surviving a deploy

In-memory state is the honest weakness of the design above. Deploy mid-run and the run is gone.

If runs are short and users can retry, accept it and move on. If runs are long or expensive, persist. The cheapest version that works: write every message to a table as it is appended, and rehydrate in init/1.

def init({run_id, ctx, task}) do
  messages =
    case MyApp.Agents.load_messages(run_id) do
      [] -> [%{role: "user", content: task}]
      existing -> existing
    end

  send(self(), :tick)
  {:ok, %{run_id: run_id, ctx: ctx, budget: MyApp.Agents.load_budget(run_id), messages: messages, status: :running}}
end

Then run the whole thing from an Oban job instead of a bare DynamicSupervisor child, so a node going away means the job is retried on another node rather than lost. Oban gives you uniqueness (one live run per id), retries with backoff, and a record that the run happened. If Oban is new to you, the practical Oban guide covers the setup this assumes.

One caveat that catches people: tools with side effects are not automatically idempotent across a retry. If the run crashed after refund_order succeeded but before the result was written, a naive replay refunds twice. Write the tool result inside the same transaction as the side effect, or give each tool call an idempotency key.

Testing the loop without spending money

This is where keeping Loop.step/1 a plain function pays off. Req ships a test adapter, and the Req.merge seam from step 2 is how you reach it.

# config/test.exs
config :my_app, :agent_req_options, plug: {Req.Test, MyApp.Agent.Model}

defmodule MyApp.Agent.LoopTest do
  use MyApp.DataCase, async: true

  alias MyApp.Agent.{Budget, Loop}

  defp reply(body), do: fn conn -> Req.Test.json(conn, body) end

  defp tool_use(name, input) do
    %{
      "stop_reason" => "tool_use",
      "usage" => %{"input_tokens" => 100, "output_tokens" => 20},
      "content" => [%{"type" => "tool_use", "id" => "t1", "name" => name, "input" => input}]
    }
  end

  defp final(text) do
    %{
      "stop_reason" => "end_turn",
      "usage" => %{"input_tokens" => 120, "output_tokens" => 30},
      "content" => [%{"type" => "text", "text" => text}]
    }
  end

  test "runs a tool, then answers" do
    account = insert(:account)
    insert(:order, account: account, email: "ada@example.com", status: "shipped")

    Req.Test.expect(MyApp.Agent.Model, reply(tool_use("search_orders", %{"email" => "ada@example.com"})))
    Req.Test.expect(MyApp.Agent.Model, reply(final("Your order shipped on Tuesday.")))

    state = %{ctx: %{account_id: account.id}, budget: %Budget{}, messages: [], run_id: "r1"}

    assert {:continue, state} = Loop.step(state)
    assert {:done, _state, "Your order shipped on Tuesday."} = Loop.step(state)
  end

  test "stops at the step limit instead of looping forever" do
    state = %{ctx: %{}, budget: %Budget{steps: 12}, messages: [], run_id: "r2"}
    assert {:halted, {:budget, :step_limit}} = Loop.step(state)
  end
end

That second test is the one that matters. It is a two-line assertion that your agent cannot run away, and it costs nothing to run on every commit. Add a third that stubs a tool raising, and assert the loop returns an is_error block rather than crashing.

For the handful of tests that should hit a real model, tag them @tag :live and exclude the tag by default in test_helper.exs. Run them nightly, not on every push.

Honest tradeoffs

Agents are oversold, so here is the counter case.

Most tasks do not need one. If the sequence of steps is known in advance, write the sequence. A three-step pipeline of ordinary function calls with one model call in the middle is cheaper, faster, deterministic and debuggable. Reach for an agent when the path genuinely varies with the input, not because agents are interesting.

Latency compounds. Six iterations at four seconds each is a twenty-four second wait. Streaming progress into LiveView makes that bearable rather than fast. If the answer must arrive in two seconds, an agent is the wrong tool.

Non-determinism is permanent. The same input can take a different path tomorrow. Your test suite pins behaviour with stubbed replies, which verifies your code and says nothing about the model’s choices. Log every run’s full message history: when something goes wrong in production, the transcript is the only debugger you have.

Frameworks are optional. Everything above is a few hundred lines of ordinary Elixir. If you would rather have chains, provider fallbacks and callbacks handled for you, Elixir LangChain is a good library and its mode: :while_needs_response implements this loop. Writing it yourself is worth doing once, so that you know what the framework is doing when it misbehaves.

Tools are an attack surface. Anything the model reads can contain instructions aimed at your agent, including a customer’s support email or a scraped web page. Scope every tool to the authenticated account, gate the destructive ones behind approval, and never build SQL from model output.

Skip the plumbing

If you want the supervised runner, the tool registry, the approval gate and the LiveView already assembled, phx_ai ships a Phoenix app with the model client, streaming and tool calling wired in, so you start at “which tools does my product need” instead of “how do I stream a token”. If your agent’s tools should be reachable by Claude Desktop or another host, the same pieces expose cleanly over an MCP server in Elixir.

Building more than one thing this year? The Builder Pass is lifetime access to every template in the catalog, which works out cheaper than buying two.

Summary

An Elixir AI agent is a loop plus the guardrails that keep the loop honest:

• Tools as data, with schemas and implementations in one module, and the account scope taken from the session rather than from the model.
• A pure step/1 function returning :continue, :done, :awaiting_approval or :halted, so the interesting logic is testable without a network.
• Hard budgets on steps and cost, checked before every model call, because the prompt is not a spending limit.
• A supervised GenServer per run, advanced with send(self(), :tick) so cancels and approvals land between iterations.
• PubSub into LiveView for progress, which decouples the run from the browser tab.
• Approval gates on anything that spends money or deletes data.
• Persistence plus Oban when runs are long enough that losing one to a deploy actually hurts.

The loop is the easy part. The budget, the approval gate and the account scoping are what make it something you can leave running.