Axon + ACP with OpenCode

git clone https://github.com/runloopai/runloop-examples \
  && cd runloop-examples/axon-broker-agents \
  && uv run axon_acp_docs.py

git clone https://github.com/runloopai/runloop-examples \
  && cd runloop-examples/axon-broker-agents \
  && bun install \
  && bun run axon_acp_docs.ts

# /// 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())

// 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();