imp_checkout? = fn path ->
is_binary(path) and File.regular?(Path.join(path, "mix.exs")) and
File.regular?(Path.join(path, "lib/imp.ex"))
end
explicit_repo = System.get_env("IMP_PATH")
if explicit_repo && not imp_checkout?.(Path.expand(explicit_repo)) do
raise "IMP_PATH does not point to an Imp source checkout or unpacked package"
end
repo =
[explicit_repo, Path.expand("..", __DIR__), File.cwd!()]
|> Enum.reject(&is_nil/1)
|> Enum.map(&Path.expand/1)
|> Enum.find(imp_checkout?)
if repo do
# Prefer the notebook's own Imp checkout even when Livebook was launched
# from an unrelated Mix project. Unpacked package archives may omit
# mix.lock, so pin it only when the source checkout actually provides it.
install_opts =
if File.regular?(Path.join(repo, "mix.lock")),
do: [lockfile: Path.join(repo, "mix.lock")],
else: []
Mix.install([{:imp, path: repo}], install_opts)
else
# Standalone notebook: install the released package from Hex.
Mix.install([{:imp, "~> 0.5"}])
end
# Live cells run when LIVE_PROVIDER=1 and a key is set: OPENAI_API_KEY, or
# OPENROUTER_API_KEY for the same model through OpenRouter. In your own code,
# any ReqLLM model string works.
live_lm = fn opts ->
model = System.get_env("OPENAI_MODEL", "gpt-5.4-mini")
cond do
System.get_env("LIVE_PROVIDER") != "1" ->
{:skip, "Set LIVE_PROVIDER=1 and OPENAI_API_KEY to run this cell."}
key = System.get_env("OPENAI_API_KEY") ->
{:ok, Imp.req_llm("openai:" <> model, Keyword.put(opts, :api_key, key))}
key = System.get_env("OPENROUTER_API_KEY") ->
{:ok, Imp.req_llm("openrouter:openai/" <> model, Keyword.put(opts, :api_key, key))}
true ->
{:skip, "Set OPENAI_API_KEY to run this cell."}
end
endTools
Livebook 03 handled improvement loops. This chapter adds action boundaries: tools, ReAct, MCP imports, and RLM all let an Imp program do controlled work outside a single LM completion.
Tools are ordinary named functions with metadata. ReAct and RLM use the same tool struct, which keeps policy and tracing consistent.
lookup =
Imp.tool(:lookup, "lookup a fact", fn %{"query" => _query} -> "Paris" end)
Imp.Tool.call(lookup, %{"query" => "capital-france"})Owning the loop yourself
The packaged agent surface is the react-family spectrum above. When you want a loop the framework does not ship, compose the same pieces in ordinary Elixir: tools are values, and calling one is just a function call your own supervised process can orchestrate.
Imp.Tool.call(lookup, %{"query" => "capital-france"})Policy lives where the model chooses actions: any react-family program takes
tool_policy:, and a tool outside the policy becomes a recorded denial
instead of an execution. The ReAct section below shows the action loop; use a
deny policy when you want to exercise this boundary directly.
Tools from an MCP server
MCP (the Model Context Protocol) is a standard way for a program to offer
tools to a model client. Imp.MCP.connect/2 starts or dials the servers you
list and imports their tools as ordinary Imp.Tool values.
A real server would be a package you install. To keep this notebook offline,
the next cell writes a small one: an Elixir script that answers MCP over its
standard input and output and offers one tool, charter.
server_source = ~S"""
defmodule Charters do
@charters %{
"atlas" => "atlas owns money: charges, refunds, invoices, plans.",
"harbor" => "harbor owns the platform: outages, errors, latency.",
"beacon" => "beacon owns identity: accounts, credentials, sessions.",
"quill" => "quill owns the product: feature requests, how-to questions."
}
def loop do
case IO.read(:stdio, :line) do
:eof -> :ok
line -> line |> JSON.decode!() |> reply() |> write(); loop()
end
end
defp reply(%{"id" => id, "method" => "initialize", "params" => params}),
do: {id, %{"protocolVersion" => params["protocolVersion"], "capabilities" => %{"tools" => %{}},
"serverInfo" => %{"name" => "charters", "version" => "1.0.0"}}}
defp reply(%{"id" => id, "method" => "tools/list"}),
do: {id, %{"tools" => [%{"name" => "charter", "description" => "The charter of one squad.",
"inputSchema" => %{"type" => "object", "properties" => %{"team" => %{"type" => "string"}}, "required" => ["team"]}}]}}
defp reply(%{"id" => id, "method" => "tools/call", "params" => %{"arguments" => %{"team" => team}}}),
do: {id, %{"content" => [%{"type" => "text", "text" => Map.get(@charters, team, "No squad named #{team}.")}]}}
defp reply(%{"id" => id}), do: {:error, id}
defp reply(_notification), do: nil
defp write(nil), do: :ok
defp write({:error, id}),
do: IO.puts(JSON.encode!(%{"jsonrpc" => "2.0", "id" => id, "error" => %{"code" => -32601, "message" => "Method not found"}}))
defp write({id, result}), do: IO.puts(JSON.encode!(%{"jsonrpc" => "2.0", "id" => id, "result" => result}))
end
Charters.loop()
"""
script = Path.join(System.tmp_dir!(), "charters_mcp_server.exs")
File.write!(script, server_source)A descriptor names the command to start. Imp starts only servers you
authorize, here with trusted_servers::
server = %{"name" => "charters", "command" => System.find_executable("elixir"), "args" => [script]}
{:ok, imported} = Imp.MCP.connect([server], trusted_servers: [server])
[charter] = imported.tools
{charter.name, charter.metadata.mcp.server_name,
Imp.Tool.call(charter, %{"team" => "harbor"})}The server runs until its connection closes: imported.cleanup.() stops it,
and so does the end of the process that connected.
imported.cleanup.()Tools and MCP covers remote servers, credentials, and what happens when a server does not answer.
ReAct
Imp.react/3 builds a tool-using agent: the model calls tools natively, one
step per request, and the turn ends when it answers in text. A signature a
text answer cannot fill (several outputs, or a typed one) gets a reserved
submit tool whose arguments are validated against the signature instead.
{:ok, script} =
Agent.start_link(fn ->
[%{tool_calls: [%{name: :lookup, arguments: %{query: "capital-france"}}]}, "Paris"]
end)
lm =
Imp.LM.Static.new(
handler: fn _messages, _opts ->
Agent.get_and_update(script, fn [next | rest] -> {next, rest} end)
end
)
react = Imp.react("question -> answer", [lookup], lm: lm, max_iters: 2, tool_policy: [:lookup])
{:ok, pred} = Imp.call(react, %{question: "Capital of France?"})
Imp.get(pred, :answer)Run ReAct with a live provider
This cell proves the provider tool-call path with a real LM only when
LIVE_PROVIDER=1 and provider credentials are present.
case live_lm.(max_tokens: 180) do
{:ok, lm} ->
live_react =
Imp.react(
Imp.signature(
"question -> answer",
"""
Use the lookup tool first with query "capital-france".
If the history already contains a lookup result of Paris, stop calling lookup and answer "Paris".
Do not answer directly without using lookup.
"""
),
[lookup],
lm: lm,
tool_policy: [:lookup],
max_iters: 4
)
{:ok, live_prediction} =
Enum.reduce_while(1..3, {:error, :not_run}, fn _attempt, _last ->
case Imp.call(live_react, %{question: "What is the capital of France?"}) do
{:ok, prediction} -> {:halt, {:ok, prediction}}
{:error, _reason} = error -> {:cont, error}
end
end)
answer = Imp.get(live_prediction, :answer)
unless is_binary(answer) and answer =~ "Paris" do
raise "live ReAct call returned an invalid result: #{inspect(Imp.to_map(live_prediction))}"
end
Imp.to_map(live_prediction)
skip ->
skip
endRLM
RLM explores large or awkward context through persistent, constrained Elixir code rather than stuffing the entire context into one prompt.
actions = [
%{reasoning: "Keep an exact symbolic value.", code: ~S|scratch = "Paris"|},
%{
reasoning: "Exercise the registered tool inside the environment.",
code: ~S|fact = lookup(%{"query" => "capital-france"})|
},
%{reasoning: "Submit the persisted value.", code: ~S|submit(%{answer: scratch})|}
]
controller_lm =
Imp.LM.Static.new(
handler: fn _messages, _opts ->
[action | rest] = Process.get(:rlm_actions)
Process.put(:rlm_actions, rest)
action
end
)
Process.put(:rlm_actions, actions)
rlm =
Imp.rlm("context, question -> answer",
lm: controller_lm,
tools: [lookup],
max_iterations: 5,
max_llm_calls: 5,
max_preview_chars: 20
)
{:ok, pred} =
Imp.call(rlm, %{
context: String.duplicate("large context ", 200),
question: "What is the capital?"
})
Process.delete(:rlm_actions)
{Imp.to_map(pred), pred.metadata.rlm_trace}RLM lazy context loading
Use a serializable handle when a value is too large or expensive to show in the first controller prompt. The controller loads it only if needed.
lazy_context =
Imp.rlm_serializable(:context, fn ->
"large private context"
end,
metadata: %{source: "demo"}
)
Process.put(:rlm_lazy_actions, [
%{reasoning: "Materialize the lazy handle.", code: ~S|context = load("context")|},
%{reasoning: "Inspect it without copying it into controller JSON.", code: "print(String.length(context))"},
%{reasoning: "Submit.", code: ~S|submit(%{answer: "loaded"})|}
])
lazy_controller =
Imp.LM.Static.new(
handler: fn _messages, _opts ->
[action | rest] = Process.get(:rlm_lazy_actions)
Process.put(:rlm_lazy_actions, rest)
action
end
)
lazy_rlm = Imp.rlm("context, question -> answer", lm: lazy_controller)
{:ok, lazy_pred} = Imp.call(lazy_rlm, %{context: lazy_context, question: "q"})
Process.delete(:rlm_lazy_actions)
{Imp.to_map(lazy_pred), lazy_pred.metadata.rlm_trace}RLM batched subqueries
RLM code can ask a sub-LM several ordered questions and retain the results as
ordinary environment values. Each batch item counts against max_llm_calls.
parent = self()
batch_controller =
Imp.LM.Static.new(
handler: fn _messages, _opts ->
[action | rest] = Process.get(:rlm_batch_actions)
Process.put(:rlm_batch_actions, rest)
action
end
)
batch_sub_lm =
Imp.LM.Static.new(
handler: fn messages, _opts ->
prompt = Enum.map_join(messages, "\n", & &1.content)
send(parent, {:rlm_batch_prompt, prompt})
%{answer: if(prompt =~ "first", do: "one", else: "two")}
end
)
Process.put(:rlm_batch_actions, [
%{
reasoning: "Run two semantic calls from inside the environment.",
code: ~S|results = llm_query_batched(["first", "second"])|
},
%{reasoning: "Results persist for later computation.", code: ~S|submit(%{answer: "batched"})|}
])
batch_rlm =
Imp.rlm("question -> answer",
lm: batch_controller,
sub_lm: batch_sub_lm,
max_iterations: 3,
max_llm_calls: 2
)
{:ok, batch_pred} = Imp.call(batch_rlm, %{question: "parent"})
Process.delete(:rlm_batch_actions)
{Imp.to_map(batch_pred), batch_pred.metadata.rlm_trace}RLM budget failure
loop_lm =
Imp.LM.Static.new(
handler: fn messages, _opts ->
if Enum.map_join(messages, "\n", & &1.content) =~ "RLM extract pass" do
%{answer: "recovered by extract"}
else
%{reasoning: "Inspect without submitting.", code: "1 + 1"}
end
end
)
rlm = Imp.rlm("question -> answer", lm: loop_lm, max_iterations: 1)
Imp.call(rlm, %{question: "loop?"})Run RLM with a live provider
This example asks the controller to emit constrained Elixir and submit from the persistent environment.
case live_lm.(max_tokens: 100, response_format: %{type: "json_object"}) do
{:ok, lm} ->
live_rlm =
Imp.rlm(
Imp.signature(
"question -> answer",
"Return reasoning and Elixir code that assigns Paris to a variable and submits it as answer."
),
lm: lm,
max_iterations: 2,
max_llm_calls: 2
)
{:ok, live_prediction} =
Imp.call(live_rlm, %{
question: "Capital of France?"
})
unless Imp.get(live_prediction, :answer) == "Paris" do
raise "live RLM call returned an invalid result: #{inspect(Imp.to_map(live_prediction))}"
end
{Imp.to_map(live_prediction), live_prediction.metadata.rlm_trace}
skip ->
skip
endNext: open livebooks/05_operating_imp.livemd to bound calls under a
supervisor, check what traces and saved files hold, and run a live check.