使用 OpenAI Agents SDK 构建 AI 编程 Agent

OPENAI_API_KEY=sk-your-openai-key
CLOUDFLARE_SANDBOX_API_KEY=your-bridge-token
CLOUDFLARE_SANDBOX_WORKER_URL=https://cloudflare-sandbox-bridge.your-subdomain.workers.dev

# /// script
# requires-python = ">=3.12"
# dependencies = ["openai-agents[cloudflare]"]
# ///
"""One-shot coding agent backed by a Cloudflare Sandbox."""

from __future__ import annotations

import asyncio
import os
import sys
from pathlib import Path

from agents import Runner
from agents.extensions.sandbox.cloudflare import (
    CloudflareSandboxClient,
    CloudflareSandboxClientOptions,
)
from agents.run import RunConfig
from agents.sandbox import SandboxAgent, SandboxRunConfig
from agents.sandbox.capabilities import Shell

MODEL = "gpt-5.4"

INSTRUCTIONS = """\
You are an expert developer working inside a sandbox.
The sandbox has bun, node, npm, and python available on the PATH.
Implement the user's task in /workspace, test it, then copy deliverable files to /workspace/output/.
""".strip()


async def copy_output(session, dest: Path) -> list[Path]:
    """Download files from /workspace/output/ in the sandbox to a local directory."""
    dest.mkdir(parents=True, exist_ok=True)
    ls = await session.exec("find", "/workspace/output", "-maxdepth", "1", "-type", "f", shell=False)
    if not ls.ok():
        return []
    copied: list[Path] = []
    for name in (l.strip() for l in ls.stdout.decode().splitlines() if l.strip()):
        handle = await session.read(Path(name))
        local = dest / Path(name).name
        payload = handle.read(); handle.close()
        local.write_bytes(payload if isinstance(payload, bytes) else payload.encode())
        copied.append(local)
    return copied


async def run(prompt: str, output_dir: Path) -> None:
    worker_url = os.environ.get("CLOUDFLARE_SANDBOX_WORKER_URL", "")
    if not worker_url:
        sys.exit("Error: CLOUDFLARE_SANDBOX_WORKER_URL is not set.")

    agent = SandboxAgent(
        name="Developer",
        model=MODEL,
        instructions=INSTRUCTIONS,
        capabilities=[Shell()],
    )

    client = CloudflareSandboxClient()
    options = CloudflareSandboxClientOptions(worker_url=worker_url)
    session = await client.create(manifest=agent.default_manifest, options=options)

    try:
        async with session:
            run_config = RunConfig(
                sandbox=SandboxRunConfig(session=session),
                tracing_disabled=True,
            )

            # Stream tool calls so the user can follow progress.
            result = Runner.run_streamed(agent, prompt, run_config=run_config)
            async for ev in result.stream_events():
                if ev.type == "run_item_stream_event" and ev.name == "tool_called":
                    print(f"  [tool] {getattr(ev.item.raw_item, 'name', '')}")
                elif ev.type == "run_item_stream_event" and ev.name == "tool_output":
                    print(f"  [output] {str(getattr(ev.item, 'output', ''))[:200]}")

            # Copy output files from the sandbox to the local machine.
            copied = await copy_output(session, output_dir)
            if copied:
                print(f"\nCopied {len(copied)} file(s) to {output_dir}:")
                for p in copied:
                    print(f"   {p}")
            else:
                print("\nAgent did not produce any output files.")
    finally:
        await client.delete(session)


if __name__ == "__main__":
    prompt = sys.argv[1] if len(sys.argv) > 1 else "Create a hello world HTTP server using Bun.serve"
    asyncio.run(run(prompt, Path("output")))