> ## Documentation Index
> Fetch the complete documentation index at: https://docs.runloop.ai/llms.txt
> Use this file to discover all available pages before exploring further.

# Axon + ACP with OpenCode

> End-to-end example: create an Axon, attach Broker with the ACP protocol, and stream OpenCode output.

<Tip>
  The canonical, fastest-moving copies of these scripts live in
  [runloop-examples/axon-broker-agents](https://github.com/runloopai/runloop-examples/tree/main/axon-broker-agents).
  Clone or browse that directory for the latest fixes and dependency pins; this page mirrors the scripts here for in-context reading and may lag GitHub.
</Tip>

## What you need

* A Runloop API key (`RUNLOOP_API_KEY`)
* **Python**: 3.11+ and [`uv`](https://github.com/astral-sh/uv)
* **TypeScript**: [Bun](https://bun.sh) and the repo’s `package.json` dependencies (see the example folder)

## What this example does

This walkthrough ties together [Axons](/docs/axons/overview), [Broker](/docs/axons/broker), and the [ACP protocol adapter](/docs/axons/broker/acp). The scripts create an Axon event stream, start a devbox with a `broker_mount` that runs OpenCode in ACP mode, publish ACP-shaped events onto the stream, and read agent output from the same subscription until the turn completes.

## Environment variables

```bash theme={null}
export RUNLOOP_API_KEY="your-runloop-api-key"
```

## Run from the examples repo

Clone the repo and run the script for your language:

<CodeGroup>
  ```bash Python theme={null}
  git clone https://github.com/runloopai/runloop-examples \
    && cd runloop-examples/axon-broker-agents \
    && uv run axon_acp_docs.py
  ```

  ```bash TypeScript theme={null}
  git clone https://github.com/runloopai/runloop-examples \
    && cd runloop-examples/axon-broker-agents \
    && bun install \
    && bun run axon_acp_docs.ts
  ```
</CodeGroup>

## Full scripts

The following matches [axon\_acp\_docs.py](https://github.com/runloopai/runloop-examples/blob/main/axon-broker-agents/axon_acp_docs.py) and [axon\_acp\_docs.ts](https://github.com/runloopai/runloop-examples/blob/main/axon-broker-agents/axon_acp_docs.ts) on `main`.

<CodeGroup>
  ```python Python theme={null}
  # /// script
  # requires-python = ">=3.11"
  # dependencies = [
  #     "runloop-api-client",
  #     "agent-client-protocol",
  # ]
  # ///
  # Run this script with: uv run axon_acp_docs.py

  from __future__ import annotations

  import asyncio
  import json
  import os
  import warnings
  import acp

  from acp import (
      InitializeRequest,
      NewSessionRequest,
      PROTOCOL_VERSION,
      PromptRequest,
  )
  from acp.schema import (
      Implementation,
      TextContentBlock,
  )

  from runloop_api_client import AsyncRunloopSDK
  from runloop_api_client.types.axon_publish_params import AxonPublishParams
  from typing import Literal

  warnings.filterwarnings("ignore", message="Pydantic serializer warnings")


  def make_axon_event(
      event_type: str,
      payload: InitializeRequest | NewSessionRequest | PromptRequest | str,
      *,
      origin: Literal["EXTERNAL_EVENT", "AGENT_EVENT", "USER_EVENT"] = "USER_EVENT",
      source: str = "axon_acp",
  ) -> AxonPublishParams:
      """Build a publish-ready event with sensible defaults."""
      wire_payload = (
          payload
          if isinstance(payload, str)
          else json.dumps(
              payload.model_dump(mode="json", by_alias=True, exclude_none=True)
          )
      )
      return {
          "event_type": event_type,
          "origin": origin,
          "payload": wire_payload,
          "source": source,
      }


  async def main(sdk: AsyncRunloopSDK) -> None:
      # Create an Axon for session communication
      axon = await sdk.axon.create(name="acp-tutorial-axon")

      print("creating a devbox and installing opencode")

      # Create a Devbox with an ACP-compliant agent, Opencode
      async with await sdk.devbox.create(
          name="acp-tutorial-opencode-devbox",
          mounts=[
              {
                  "type": "broker_mount",
                  "axon_id": axon.id,
                  "protocol": "acp",
                  "agent_binary": "opencode",
                  "launch_args": ["acp"],
              }
          ],
          launch_parameters={
              "launch_commands": ["npm i -g opencode-ai"],
          },
      ) as devbox:
          print(f"created devbox, id={devbox.id}")

          async with await axon.subscribe_sse() as stream:
              await axon.publish(
                  **make_axon_event(
                      "initialize",
                      acp.InitializeRequest(
                          protocol_version=PROTOCOL_VERSION,
                          client_info=Implementation(
                              name="runloop-axon", version="1.0.0"
                          ),
                      ),
                  )
              )
              await axon.publish(
                  **make_axon_event(
                      "session/new",
                      NewSessionRequest(cwd="/home/user", mcp_servers=[]),
                  )
              )

              session_id: str = ""
              prompt_sent = False
              user_prompt = "Who are you?"

              async for ev in stream:
                  # Phase 1: Wait for session/new response from the agent
                  if (
                      not session_id
                      and ev.event_type == "session/new"
                      and ev.origin == "AGENT_EVENT"
                  ):
                      session_id = json.loads(ev.payload)["sessionId"]
                      print(f"> {user_prompt}")
                      print("< ", end="", flush=True)
                      prompt = PromptRequest(
                          session_id=session_id,
                          prompt=[TextContentBlock(type="text", text=user_prompt)],
                      )
                      await axon.publish(**make_axon_event("session/prompt", prompt))
                      prompt_sent = True
                      continue

                  # Phase 2: Stream agent response
                  if prompt_sent:
                      # Check for session/update events with agent_message_chunk
                      if ev.event_type == "session/update" and ev.origin == "AGENT_EVENT":
                          parsed = json.loads(ev.payload)
                          if parsed.get("update", {}).get("sessionUpdate") == "agent_message_chunk":
                              text_part = parsed.get("update", {}).get("content", {}).get("text")
                              if text_part:
                                  print(text_part, end="", flush=True)
                      if ev.event_type == "turn.completed":
                          break
              print()

              print(
                  f"\nView full Axon event stream at https://platform.runloop.ai/axons/{axon.id}"
              )


  async def run() -> None:
      async with AsyncRunloopSDK() as sdk:
          await main(sdk)


  if __name__ == "__main__":
      if not os.getenv("RUNLOOP_API_KEY"):
          print("RUNLOOP_API_KEY is not set")
          exit(1)
      asyncio.run(run())
  ```

  ```typescript TypeScript theme={null}
  // Run this script with: bun install && bun run axon_acp_docs.ts
  // Requires package.json with dependencies

  import { RunloopSDK } from "@runloop/api-client";
  import type { AxonPublishParams } from "@runloop/api-client/resources";
  import {
    InitializeRequest,
    NewSessionRequest,
    PromptRequest,
    PROTOCOL_VERSION,
  } from "@agentclientprotocol/sdk";

  function makeAxonEvent(
    eventType: string,
    payload: InitializeRequest | NewSessionRequest | PromptRequest | string,
    {
      origin = "USER_EVENT",
      source = "axon_acp",
    }: { origin?: AxonPublishParams["origin"]; source?: string } = {}
  ): AxonPublishParams {
    const wirePayload = typeof payload === "string" ? payload : JSON.stringify(payload);
    return {
      event_type: eventType,
      origin,
      payload: wirePayload,
      source,
    };
  }

  async function main(sdk: RunloopSDK): Promise<void> {
    // Create an Axon for session communication
    const axon = await sdk.axon.create({ name: "acp-tutorial-axon" });

    console.log("creating a devbox and installing opencode");
    // Create a Devbox with an ACP-compliant agent, Opencode
    const devbox = await sdk.devbox.create({
      name: "acp-tutorial-opencode-devbox",
      mounts: [
        {
          type: "broker_mount",
          axon_id: axon.id,
          protocol: "acp",
          agent_binary: "opencode",
          launch_args: ["acp"],
        },
      ],
      launch_parameters: {
        launch_commands: ["npm i -g opencode-ai"],
      },
    });

    console.log(`created devbox, id=${devbox.id}`);

    try {
      const stream = await axon.subscribeSse();

      await axon.publish(
        makeAxonEvent("initialize", {
          protocolVersion: PROTOCOL_VERSION,
          clientInfo: { name: "runloop-axon", version: "1.0.0" },
        } as InitializeRequest)
      );

      await axon.publish(
        makeAxonEvent("session/new", {
          cwd: "/home/user",
          mcpServers: [],
        } as NewSessionRequest)
      );

      let sessionId = "";
      let promptSent = false;
      const userPrompt = "Who are you?";

      for await (const ev of stream) {
        // Phase 1: Wait for session/new response from the agent
        if (!sessionId && ev.event_type === "session/new" && ev.origin === "AGENT_EVENT") {
          sessionId = JSON.parse(ev.payload).sessionId;
          console.log(`> ${userPrompt}`);
          process.stdout.write("< ");

          const prompt: PromptRequest = {
            sessionId,
            prompt: [{ type: "text", text: userPrompt }],
          };
          await axon.publish(makeAxonEvent("session/prompt", prompt));
          promptSent = true;
          continue;
        }

        // Phase 2: Stream agent response
        if (promptSent) {
          // Check for session/update events with agent_message_chunk
          if (ev.event_type === "session/update" && ev.origin === "AGENT_EVENT") {
            const parsed = JSON.parse(ev.payload);
            if (parsed.update?.sessionUpdate === "agent_message_chunk") {
              const textPart = parsed.update?.content?.text;
              if (textPart) {
                process.stdout.write(textPart);
              }
            }
          }
          if (ev.event_type === "turn.completed") {
            break;
          }
        }
      }
      console.log();

      console.log(
        `\nView full Axon event stream at https://platform.runloop.ai/axons/${axon.id}`
      );
    } finally {
      await devbox.shutdown();
    }
  }

  async function run(): Promise<void> {
    const sdk = new RunloopSDK();
    await main(sdk);
  }

  if (!process.env.RUNLOOP_API_KEY) {
    console.log("RUNLOOP_API_KEY is not set");
    process.exit(1);
  }

  run();
  ```
</CodeGroup>

## Learn more

* [Axons overview](/docs/axons/overview)
* [Broker](/docs/axons/broker)
* [ACP protocol adapter](/docs/axons/broker/acp)