Control Claude Code ACP from Livebook

Mix.install([
  {:acpex, "~> 0.1.0"},
  {:jason, "~> 1.4"},
  {:kino, "~> 0.14"}
])

🚀 Introduction

Welcome to an interactive demonstration of ACPex controlling Claude Code ACP agent!

In this livebook, you’ll build a complete AI coding assistant interface that:

• 🤖 Connects to Claude Code ACP via the Agent Client Protocol (ACP)
• 💬 Sends prompts and receives intelligent responses
• 📁 Allows the agent to read and write files on your system
• 🖥️ Enables the agent to execute terminal commands
• 🎨 Provides a beautiful interactive UI powered by Kino

This is a real, working implementation - not a simulation! You’ll be controlling an actual AI agent that can help you write code, analyze projects, and perform complex tasks.

📋 Prerequisites

Before running this livebook, ensure you have:

1. Claude Code ACP Installed

Option A: NPM (Node.js 20+)

npm install -g @zed-industries/claude-code-acp

2. Authentication

Claude Code ACP requires an Anthropic API key. Set the ANTHROPIC_API_KEY environment variable:

export ANTHROPIC_API_KEY="your-api-key-here"

Get your API key from: https://console.anthropic.com/settings/keys

3. Set up Environment

Livebook needs the PATH environment variable to find node (required for the claude-code-acp shebang):

Kino.Markdown.new("""
**Environment configured**

PATH includes: `node` at `#{System.find_executable("node") || "not found"}`

ANTHROPIC_API_KEY: is `#{System.get_env("ANTHROPIC_API_KEY", "not api key :(") |> String.slice(0..11)}`
""")

4. Verify Installation

Check that Claude Code ACP is available:

claude_code_acp_path = "claude-code-acp"

claude_acp_path = System.find_executable(claude_code_acp_path)

if claude_acp_path do
  Kino.Markdown.new("""
  ✅ **Claude Code ACP found!**

  Path: `#{claude_acp_path}`
  """)
else
  Kino.Markdown.new("""
  ❌ **Claude Code ACP not found!**

  Please install it using the NPM command above, then restart the runtime.
  """)
end

🎨 Building the Interactive UI

Let’s create a beautiful dashboard for controlling the agent!

# Create UI components
defmodule UI do
  def create_dashboard do
    %{
      # Status indicator
      status_frame: Kino.Frame.new(),

      # Agent output area
      output_frame: Kino.Frame.new(),

      # File operations log
      file_log_frame: Kino.Frame.new(),

      # Terminal output
      terminal_frame: Kino.Frame.new(),

      # Input controls
      prompt_form:
        Kino.Control.form(
          [
            prompt:
              Kino.Input.textarea("💭 Your Prompt",
                default: "Analyze the structure of this ACPex library and suggest improvements."
              )
          ],
          submit: "📤 Send Prompt",
          reset_on_submit: []
        ),
      model_selector:
        Kino.Input.select(
          "Model",
          [
            {"claude-opus-4", "Claude Opus 4 (Most Capable)"},
            {"claude-sonnet-4", "Claude Sonnet 4 (Balanced)"},
            {"claude-haiku-4", "Claude Haiku 4 (Fastest)"}
          ],
          default: "claude-sonnet-4"
        ),

      # Action buttons
      start_button: Kino.Control.button("🚀 Start Agent"),
      stop_button: Kino.Control.button("🛑 Stop Agent")
    }
  end

  def render_dashboard(ui) do
    Kino.Layout.grid(
      [
        # Header
        Kino.Markdown.new("## 🤖 Claude Code ACP Agent Controller"),

        # Status Section
        Kino.Layout.grid([
          Kino.Markdown.new("### Status"),
          ui.status_frame
        ]),

        # Control Section
        Kino.Layout.grid([
          Kino.Markdown.new("### Controls"),
          Kino.Layout.grid([ui.start_button, ui.stop_button], columns: 2),
          ui.model_selector
        ]),

        # Prompt Section
        Kino.Layout.grid([
          Kino.Markdown.new("### Prompt"),
          ui.prompt_form
        ]),

        # Output Section
        Kino.Layout.grid([
          Kino.Markdown.new("### 💬 Agent Response"),
          ui.output_frame
        ]),

        # File Operations
        Kino.Layout.grid([
          Kino.Markdown.new("### 📁 File Operations"),
          ui.file_log_frame
        ]),

        # Terminal Output
        Kino.Layout.grid([
          Kino.Markdown.new("### 🖥️ Terminal Output"),
          ui.terminal_frame
        ])
      ],
      columns: 1
    )
  end

  def update_status(frame, status, color \\ :blue) do
    icon =
      case status do
        :disconnected -> "⭕"
        :connecting -> "🔄"
        :connected -> "✅"
        :error -> "❌"
        _ -> "ℹ️"
      end

    color_code =
      case color do
        :green -> "#22c55e"
        :red -> "#ef4444"
        :blue -> "#3b82f6"
        :yellow -> "#eab308"
        _ -> "#6b7280"
      end

    Kino.Frame.render(
      frame,
      Kino.Markdown.new("""
      
        #{icon} #{status |> to_string() |> String.upcase()}
      
      """)
    )
  end

  def log_file_operation(frame, operation, path, success \\ true) do
    icon = if success, do: "✅", else: "❌"

    emoji =
      case operation do
        :read -> "📖"
        :write -> "✍️"
        _ -> "📄"
      end

    timestamp = DateTime.utc_now() |> DateTime.to_string()

    Kino.Frame.append(
      frame,
      Kino.Markdown.new("""
      `#{timestamp}` #{icon} #{emoji} **#{operation}** `#{path}`
      """)
    )
  end

  def log_terminal(frame, command, output \\ nil) do
    content =
      if output do
        """
        ```bash
        $ #{command}
        #{output}
        ```
        """
      else
        """
        ```bash
        $ #{command}
        ```
        """
      end

    Kino.Frame.append(frame, Kino.Markdown.new(content))
  end

  def render_agent_update(frame, update) do
    content =
      case update do
        %{"kind" => "thought", "content" => text} ->
          "💭 **Thinking:** #{text}"

        %{"kind" => "message", "content" => text} ->
          "💬 **Response:**\n\n#{text}"

        %{"kind" => "tool_call", "tool" => tool, "args" => args} ->
          "🔧 **Tool Call:** `#{tool}`\n```json\n#{Jason.encode!(args, pretty: true)}\n```"

        %{"kind" => "plan", "steps" => steps} ->
          steps_text =
            steps
            |> Enum.with_index(1)
            |> Enum.map(fn {step, i} -> "#{i}. #{step}" end)
            |> Enum.join("\n")

          "📋 **Plan:**\n\n#{steps_text}"

        _ ->
          "ℹ️ **Update:** #{inspect(update)}"
      end

    Kino.Frame.append(frame, Kino.Markdown.new(content <> "\n\n---\n"))
  end
end

# Create and render the dashboard
ui = UI.create_dashboard()
dashboard = UI.render_dashboard(ui)

# Initialize status
UI.update_status(ui.status_frame, :disconnected, :red)

dashboard

🔌 Implementing the ACPex Client

Now let’s create a full-featured client that implements all ACP callbacks:

defmodule ClaudeClient do
  @moduledoc """
  A complete ACPex client implementation for controlling Claude Code ACP.

  This client handles:
  - Session updates and streams them to the UI
  - File system operations (read/write)
  - Terminal operations (create, execute, monitor)

  All callbacks use typed structs from ACPex.Schema for type safety and
  automatic camelCase ↔ snake_case conversion.
  """

  @behaviour ACPex.Client

  # Import schema types for typed struct usage
  alias ACPex.Schema.Session.UpdateNotification
  alias ACPex.Schema.Client.{FsReadTextFileResponse, FsWriteTextFileResponse}
  alias ACPex.Schema.Client.Terminal

  def init(args) do
    ui = Keyword.fetch!(args, :ui)

    state = %{
      ui: ui,
      sessions: %{},
      terminals: %{},
      files_accessed: []
    }

    {:ok, state}
  end

  @doc """
  Handle streaming updates from the agent.

  These come in as the agent thinks, plans, and generates responses.
  Receives an UpdateNotification struct with the update content.
  """
  def handle_session_update(%UpdateNotification{update: update}, state) do
    # Render to the output frame
    UI.render_agent_update(state.ui.output_frame, update)

    {:noreply, state}
  end

  @doc """
  Handle agent requests to read files.

  The agent might need to read source code, config files, etc.
  Receives a FsReadTextFileRequest struct, returns a FsReadTextFileResponse struct.
  """
  def handle_fs_read_text_file(request, state) do
    # Access the path from the typed struct
    path = request.path

    UI.log_file_operation(state.ui.file_log_frame, :read, path)

    # Resolve relative paths from the current directory
    full_path = Path.expand(path)

    case File.read(full_path) do
      {:ok, content} ->
        UI.log_file_operation(state.ui.file_log_frame, :read, path, true)
        new_state = %{state | files_accessed: [path | state.files_accessed]}

        # Return a typed response struct
        response = %FsReadTextFileResponse{content: content}
        {:ok, response, new_state}

      {:error, reason} ->
        UI.log_file_operation(state.ui.file_log_frame, :read, path, false)

        error = %{
          code: -32001,
          message: "Failed to read file: #{reason}"
        }

        {:error, error, state}
    end
  end

  @doc """
  Handle agent requests to write files.

  The agent can create new files or modify existing ones.
  Receives a FsWriteTextFileRequest struct, returns a FsWriteTextFileResponse struct.
  """
  def handle_fs_write_text_file(request, state) do
    # Access fields from the typed struct
    path = request.path
    content = request.content

    UI.log_file_operation(state.ui.file_log_frame, :write, path)

    full_path = Path.expand(path)

    # Create parent directory if it doesn't exist
    full_path |> Path.dirname() |> File.mkdir_p()

    case File.write(full_path, content) do
      :ok ->
        UI.log_file_operation(state.ui.file_log_frame, :write, path, true)
        new_state = %{state | files_accessed: [path | state.files_accessed]}

        # Return a typed response struct (FsWriteTextFileResponse is empty)
        response = %FsWriteTextFileResponse{}
        {:ok, response, new_state}

      {:error, reason} ->
        UI.log_file_operation(state.ui.file_log_frame, :write, path, false)

        error = %{
          code: -32002,
          message: "Failed to write file: #{reason}"
        }

        {:error, error, state}
    end
  end

  @doc """
  Create a new terminal for the agent to use.
  Receives a Terminal.CreateRequest struct, returns a Terminal.CreateResponse struct.
  """
  def handle_terminal_create(request, state) do
    terminal_id = "term-" <> Base.encode16(:crypto.strong_rand_bytes(8), case: :lower)

    UI.log_terminal(state.ui.terminal_frame, request.command)

    # In a real implementation, you might use Port or System.cmd
    # For this demo, we'll simulate it
    terminal_info = %{
      id: terminal_id,
      command: request.command,
      cwd: request.cwd || File.cwd!(),
      output: ""
    }

    new_terminals = Map.put(state.terminals, terminal_id, terminal_info)

    # Return typed response struct
    response = %Terminal.CreateResponse{terminal_id: terminal_id}
    {:ok, response, %{state | terminals: new_terminals}}
  end

  @doc """
  Get output from a terminal.
  Receives a Terminal.OutputRequest struct, returns a Terminal.OutputResponse struct.
  """
  def handle_terminal_output(request, state) do
    case Map.get(state.terminals, request.terminal_id) do
      nil ->
        error = %{code: -32003, message: "Terminal not found"}
        {:error, error, state}

      terminal ->
        # Simulate running the command
        {output, exit_code} =
          try do
            System.cmd("sh", ["-c", terminal.command],
              cd: terminal.cwd,
              stderr_to_stdout: true
            )
          rescue
            _ -> {"Error executing command", 1}
          end

        # Update terminal with output
        updated_terminal = %{terminal | output: output}
        new_terminals = Map.put(state.terminals, request.terminal_id, updated_terminal)

        UI.log_terminal(state.ui.terminal_frame, terminal.command, output)

        # Return typed response struct
        response = %Terminal.OutputResponse{
          output: output,
          truncated: false,
          exit_status: %{exitCode: exit_code}
        }

        {:ok, response, %{state | terminals: new_terminals}}
    end
  end

  @doc """
  Wait for a terminal command to finish.
  Receives a Terminal.WaitForExitRequest struct, returns a Terminal.WaitForExitResponse struct.
  """
  def handle_terminal_wait_for_exit(request, state) do
    case Map.get(state.terminals, request.terminal_id) do
      nil ->
        error = %{code: -32003, message: "Terminal not found"}
        {:error, error, state}

      _terminal ->
        # In our simulation, commands finish immediately
        response = %Terminal.WaitForExitResponse{exit_code: 0}
        {:ok, response, state}
    end
  end

  @doc """
  Kill a running terminal command.
  Receives a Terminal.KillRequest struct, returns a Terminal.KillResponse struct.
  """
  def handle_terminal_kill(request, state) do
    case Map.get(state.terminals, request.terminal_id) do
      nil ->
        error = %{code: -32003, message: "Terminal not found"}
        {:error, error, state}

      _terminal ->
        new_terminals = Map.delete(state.terminals, request.terminal_id)
        response = %Terminal.KillResponse{}
        {:ok, response, %{state | terminals: new_terminals}}
    end
  end

  @doc """
  Release a terminal (cleanup).
  Receives a Terminal.ReleaseRequest struct, returns a Terminal.ReleaseResponse struct.
  """
  def handle_terminal_release(request, state) do
    new_terminals = Map.delete(state.terminals, request.terminal_id)
    response = %Terminal.ReleaseResponse{}
    {:ok, response, %{state | terminals: new_terminals}}
  end
end

:ok

🚀 Starting and Initializing the Agent

Now let’s connect to Claude Code ACP! This cell will:

1. Find the Claude Code ACP executable
2. Start the ACPex client connection
3. Spawn Claude Code ACP as a subprocess
4. Automatically send the initialize message

⚠️ Note: Make sure you’ve set your ANTHROPIC_API_KEY environment variable first

# Store the connection in the process dictionary so we can access it later
agent_connection = Process.get(:agent_connection)

if agent_connection && Process.alive?(agent_connection) do
  Kino.Markdown.new("""
  ⚠️ **Agent already running!**

  Click the 🛑 Stop Agent button below to stop it first.
  """)
else
  # Find Claude Code ACP
  claude_acp_path = System.find_executable(claude_code_acp_path)

  if claude_acp_path do
    UI.update_status(ui.status_frame, :connecting, :blue)

    # Start the agent
    result =
      ACPex.start_client(
        ClaudeClient,
        [ui: ui],
        agent_path: claude_acp_path
      )

    case result do
      {:ok, conn_pid} ->
        # Store connection for later use
        Process.put(:agent_connection, conn_pid)
        Process.put(:current_session_id, nil)

        # Immediately send initialize request (required by ACP protocol)
        init_response =
          ACPex.Protocol.Connection.send_request(
            conn_pid,
            "initialize",
            %{
              "protocolVersion" => 1,
              "capabilities" => %{
                "filesystem" => true,
                "terminal" => true
              },
              "clientInfo" => %{
                "name" => "Livebook",
                "version" => "0.1.0"
              }
            },
            # 30 second timeout
            30_000
          )

        case init_response do
          %{"result" => result} ->
            UI.update_status(ui.status_frame, :connected, :green)

            protocol_version = result["protocolVersion"] || result["protocol_version"]
            agent_info = result["agentInfo"] || result["agent_info"]
            capabilities = result["agentCapabilities"] || result["capabilities"]

            Kino.Markdown.new("""
            ✅ **Agent Connected and Initialized!**

            **Agent Path:** `#{claude_acp_path}`

            **Protocol Version:** #{protocol_version}

            **Agent Info:**
            ```json
            #{Jason.encode!(agent_info || %{}, pretty: true)}
            ```

            **Capabilities:**
            ```json
            #{Jason.encode!(capabilities || %{}, pretty: true)}
            ```

            You're now ready to create a session and send prompts!
            """)

          %{"error" => error} ->
            UI.update_status(ui.status_frame, :error, :red)

            Kino.Markdown.new("""
            ❌ **Initialization Failed!**

            Error: #{error["message"]}
            """)

          _ ->
            UI.update_status(ui.status_frame, :error, :red)

            Kino.Markdown.new("""
            ⚠️ **Unexpected Response**

            #{inspect(init_response)}
            """)
        end

      {:error, reason} ->
        UI.update_status(ui.status_frame, :error, :red)

        Kino.Markdown.new("""
        ❌ **Connection Failed!**

        Error: #{inspect(reason)}

        Make sure Claude Code ACP is properly installed and `ANTHROPIC_API_KEY` is set.
        """)
    end
  else
    UI.update_status(ui.status_frame, :error, :red)

    Kino.Markdown.new("""
    ❌ **Claude Code ACP not found!**

    Please install it and restart the runtime.
    """)
  end
end

💬 Create a Session

Let’s create a conversation session:

agent_connection = Process.get(:agent_connection)

if agent_connection && Process.alive?(agent_connection) do
  # Create a new session
  session_response =
    ACPex.Protocol.Connection.send_request(
      agent_connection,
      "session/new",
      %{
        "cwd" => File.cwd!(),
        "mcpServers" => []
      },
      30_000
    )

  case session_response do
    %{"result" => result} ->
      # Get session ID (accept both camelCase and snake_case)
      session_id = result["sessionId"] || result["session_id"]
      # Store session ID for later
      Process.put(:current_session_id, session_id)

      Kino.Markdown.new("""
      ✅ **Session Created!**

      Session ID: `#{session_id}`

      You're now ready to send prompts to the agent!
      """)

    %{"error" => error} ->
      Kino.Markdown.new("""
      ❌ **Session Creation Failed!**

      Error: #{error["message"]}
      """)

    _ ->
      Kino.Markdown.new("""
      ⚠️ **Unexpected Response**

      #{inspect(session_response)}
      """)
  end
else
  Kino.Markdown.new("""
  ⚠️ **Agent not connected!**

  Please run the "Starting the Agent Connection" cell above first.
  """)
end

🎯 Send Prompts to the Agent

Now for the fun part! Use the interactive form to send prompts:

# Get connection and session
agent_connection = Process.get(:agent_connection)
session_id = Process.get(:current_session_id)

# Listen to the form submission
form_stream = Kino.Control.stream(ui.prompt_form)

Kino.listen(form_stream, fn event ->
  if agent_connection && Process.alive?(agent_connection) && session_id do
    # Clear the output frame
    Kino.Frame.clear(ui.output_frame)

    # Get the prompt from the form event
    prompt_text = event.data.prompt

    if prompt_text && String.trim(prompt_text) != "" do
      # Show that we're sending
      UI.render_agent_update(ui.output_frame, %{
        "kind" => "message",
        "content" => "**You:** #{prompt_text}"
      })

      # Send the prompt
      Task.async(fn ->
        response =
          ACPex.Protocol.Connection.send_request(
            agent_connection,
            "session/prompt",
            %{
              "sessionId" => session_id,
              "prompt" => [
                %{
                  "type" => "text",
                  "text" => prompt_text
                }
              ]
            },
            # 2 minute timeout
            120_000
          )

        case response do
          %{"result" => result} ->
            stop_reason = result["stopReason"] || result["stop_reason"]
            UI.render_agent_update(ui.output_frame, %{
              "kind" => "message",
              "content" => "✅ **Complete!**\n\nStop Reason: #{stop_reason}"
            })

          %{"error" => error} ->
            UI.render_agent_update(ui.output_frame, %{
              "kind" => "message",
              "content" => "❌ **Error:** #{error["message"]}"
            })

          _ ->
            UI.render_agent_update(ui.output_frame, %{
              "kind" => "message",
              "content" => "⚠️ Unexpected response: #{inspect(response)}"
            })
        end
      end)
    end
  else
    Kino.Frame.render(
      ui.output_frame,
      Kino.Markdown.new("⚠️ **Not connected!** Please connect and initialize first.")
    )
  end
end)

Kino.Markdown.new("""
✨ **Ready to chat!**

Type your prompt in the form above and click "📤 Send Prompt" to submit.

The agent will:
- Stream its thinking process in real-time
- Read files if needed (you'll see them in the File Operations log)
- Execute commands if needed (you'll see them in Terminal Output)
- Generate intelligent responses

Try asking:
- "Analyze the architecture of this ACPex library"
- "What are the main modules and what do they do?"
- "Find all public functions and generate documentation stubs"
- "Add a test for the ACPex.Json.decode function"
""")

🛑 Stop the Agent

When you’re done, stop the agent:

# Listen to stop button
stop_stream = Kino.Control.stream(ui.stop_button)

Kino.listen(stop_stream, fn _event ->
  agent_connection = Process.get(:agent_connection)

  if agent_connection && Process.alive?(agent_connection) do
    # Stop the connection
    Process.exit(agent_connection, :normal)
    Process.put(:agent_connection, nil)
    Process.put(:current_session_id, nil)

    UI.update_status(ui.status_frame, :disconnected, :red)

    Kino.Frame.render(
      ui.output_frame,
      Kino.Markdown.new("👋 **Agent stopped!**")
    )
  end
end)

Kino.Markdown.new("✅ Stop button listener active")

📚 Example Prompts to Try

Here are some interesting prompts to explore ACPex’s capabilities:

1. Code Analysis

Analyze the ACPex.Protocol.Connection module. Explain its role in the architecture
and how it manages sessions.

2. Generate Tests

Look at lib/acpex/json.ex and generate comprehensive unit tests for the decode
functions. Place them in test/acpex/json_test.exs.

3. Documentation

Review all public functions in lib/acpex.ex and ensure they have proper @doc
and @spec annotations. Add any that are missing.

4. Refactoring

Examine the session routing logic in lib/acpex/protocol/session.ex and suggest
any improvements for clarity or performance.

5. Create New Feature

Add a new function to ACPex that lists all active connections and their session
counts. Include tests.

🎉 Summary

Congratulations! You’ve built a complete, working integration between:

• Livebook - Your interactive notebook environment
• Kino - Beautiful, reactive UI components
• ACPex - Your Elixir ACP protocol implementation
• Claude Code ACP - Anthropic’s powerful AI coding assistant

What You Learned

1. Protocol Implementation: How to implement a full ACPex client with all callbacks
2. File System Integration: How agents can read and write files through your client
3. Terminal Integration: How agents can execute commands safely
4. Real-time UI: How to build reactive interfaces with Kino
5. Agent Communication: How the Agent Client Protocol enables tool interoperability

Next Steps

• Experiment with different prompts and see what Claude can do
• Try implementing additional ACP features (like session persistence)
• Build your own custom agent using ACPex
• Create a web interface using Phoenix LiveView
• Integrate with other MCP-compatible tools

Resources

• ACPex Documentation: Check out the HexDocs for ACPex
• ACP Specification: https://agentclientprotocol.com
• Claude Code ACP: https://github.com/zed-industries/claude-code-acp
• Kino Documentation: https://hexdocs.pm/kino

Happy Coding! 🚀