OpenAI Agents API on Runloop

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "openai>=3.13.0",
#     "runloop-api-client>=1.31.0",
# ]
# ///

"""Run one OpenAI Agents API turn inside a Runloop devbox, then tear down both resources.

Requires RUNLOOP_API_KEY, OPENAI_API_KEY, and OPENAI_EXECUTOR_API_KEY. The two
OpenAI keys must share an owner, organization, and project. Only the executor
key is passed into the devbox.

Run with uv:
    uv run openai-agents-api-runloop.py
"""

import asyncio
import os
import shlex

from openai import AsyncOpenAI
from runloop_api_client import AsyncRunloopSDK

MODEL = "gpt-5.6-sol"

WORKSPACE = "/home/user/workspace"
CODEX_HOME = "/home/user/.codex-runtime"
CODEX = f"{CODEX_HOME}/node_modules/.bin/codex"

BRIEF = "Migrate a synchronous Python service to async without changing its API.\n"
PROMPT = "Read brief.txt and write a five-step migration plan to plan.md."

# Failure arrives as an event type, not an exception.
FAILURE_EVENTS = {
    "error",
    "agent.session.environment.failed",
    "agent.session.failed",
    "agent.session.turn.failed",
    "agent.session.turn.cancelled",
}


async def main() -> None:
    executor_key = os.environ["OPENAI_EXECUTOR_API_KEY"]

    async with AsyncOpenAI(timeout=360) as client, AsyncRunloopSDK() as runloop:
        session = await client.beta.agents.sessions.create(
            agent={"model": MODEL},
            environment={"type": "self_hosted", "workspace_directory": WORKSPACE},
        )
        devbox = None
        print(f"Session: {session.id}", flush=True)
        try:
            assert session.environment.type == "self_hosted"
            async with asyncio.timeout(300):
                created = await runloop.api.devboxes.create(
                    name=f"agents-api-{session.id[-12:]}",
                    environment_variables={"CODEX_API_KEY": executor_key},
                    launch_parameters={"keep_alive_time_seconds": 600},
                )
                devbox = runloop.devbox.from_id(created.id)
                print(f"Devbox: {devbox.id}", flush=True)
                await devbox.await_running()

                setup = await devbox.cmd.exec(
                    f"mkdir -p {WORKSPACE} && "
                    f"npm install --prefix {CODEX_HOME} @openai/codex@alpha"
                )
                if setup.exit_code != 0:
                    raise RuntimeError(f"Executor installation failed: {await setup.stderr()}")

                await devbox.file.write(file_path=f"{WORKSPACE}/brief.txt", contents=BRIEF)

                # Binds this devbox to the session; stays up for the life of the devbox.
                await devbox.cmd.exec_async(
                    f"cd {WORKSPACE} && exec "
                    + shlex.join(
                        [
                            CODEX,
                            "exec-server",
                            "--remote",
                            session.environment.remote_url,
                            "--environment-id",
                            session.environment.id,
                        ]
                    )
                )

                async with client.beta.agents.sessions.stream(
                    session.id, input=PROMPT
                ) as events:
                    async for event in events:
                        if event.type in FAILURE_EVENTS:
                            raise RuntimeError(f"Agent failed: {event.type}")
                        if event.type == "agent.session.turn.output_text.delta":
                            print(event.delta, end="", flush=True)
                        # Subagent turns complete first; only the main turn ends the run.
                        if (
                            event.type == "agent.session.turn.completed"
                            and event.turn.subagent_id is None
                        ):
                            break
                    else:
                        raise RuntimeError("Stream ended without a completed turn")

                # The real output is the file the agent wrote, not the streamed text.
                plan = await devbox.file.read(file_path=f"{WORKSPACE}/plan.md")
                if not plan.strip():
                    raise RuntimeError("The agent did not write a migration plan")
                print(f"\n\nplan.md:\n{plan}")
        finally:
            # Deleting the session does not stop the devbox; both need explicit teardown.
            try:
                if devbox is not None:
                    async with asyncio.timeout(30):
                        await devbox.shutdown()
            finally:
                async with asyncio.timeout(30):
                    await client.beta.agents.sessions.delete(session.id)


if __name__ == "__main__":
    asyncio.run(main())