wehub-resource-sync
242 lines
8.5 KiB
Python
242 lines
8.5 KiB
Python
from __future__ import annotations
|
|
|
|
import argparse
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Any, Literal, cast
|
|
|
|
from openai.types.responses import ResponseTextDeltaEvent
|
|
|
|
from agents import ModelSettings, Runner
|
|
from agents.run import RunConfig
|
|
from agents.sandbox import Manifest, SandboxAgent, SandboxRunConfig
|
|
from agents.sandbox.config import DEFAULT_PYTHON_SANDBOX_IMAGE
|
|
from agents.sandbox.entries import File
|
|
|
|
if __package__ is None or __package__ == "":
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[2]))
|
|
|
|
from examples.sandbox.misc.workspace_shell import WorkspaceShellCapability
|
|
|
|
Backend = Literal["docker", "modal"]
|
|
WorkspacePersistenceMode = Literal["tar", "snapshot_filesystem", "snapshot_directory"]
|
|
|
|
DEFAULT_QUESTION = "Summarize this sandbox project in 2 sentences."
|
|
DEFAULT_BACKEND: Backend = "docker"
|
|
DEFAULT_MODAL_APP_NAME = "openai-agents-python-sandbox-example"
|
|
DEFAULT_MODAL_WORKSPACE_PERSISTENCE: WorkspacePersistenceMode = "tar"
|
|
|
|
|
|
def _stream_event_banner(event_name: str) -> str | None:
|
|
if event_name == "tool_called":
|
|
return "[tool call] shell"
|
|
if event_name == "tool_output":
|
|
return "[tool output] shell"
|
|
return None
|
|
|
|
|
|
def _build_manifest(backend: Backend) -> Manifest:
|
|
backend_label = "Docker" if backend == "docker" else "Modal"
|
|
return Manifest(
|
|
entries={
|
|
"README.md": File(
|
|
content=(
|
|
b"# Demo Project\n\n"
|
|
+ (
|
|
f"This sandbox contains a tiny demo project for the {backend_label} "
|
|
"sandbox runner.\n"
|
|
).encode()
|
|
+ b"The goal is to show how Runner can prepare a sandbox workspace.\n"
|
|
)
|
|
),
|
|
"src/app.py": File(
|
|
content=b'def greet(name: str) -> str:\n return f"Hello, {name}!"\n'
|
|
),
|
|
"docs/notes.md": File(
|
|
content=(
|
|
b"# Notes\n\n"
|
|
b"- The example is intentionally minimal.\n"
|
|
b"- The model should inspect files through the shell tool.\n"
|
|
)
|
|
),
|
|
}
|
|
)
|
|
|
|
|
|
def _build_agent(*, model: str, manifest: Manifest, backend: Backend) -> SandboxAgent:
|
|
backend_label = "Docker" if backend == "docker" else "Modal"
|
|
return SandboxAgent(
|
|
name=f"{backend_label} Sandbox Assistant",
|
|
model=model,
|
|
instructions=(
|
|
"Answer questions about the sandbox workspace. Inspect the project before answering, "
|
|
"and keep the response concise. "
|
|
"Do not guess file names like package.json or pyproject.toml. "
|
|
"This demo intentionally contains a tiny workspace."
|
|
),
|
|
# `default_manifest` tells the sandbox agent which workspace it should expect.
|
|
default_manifest=manifest,
|
|
# `WorkspaceShellCapability()` exposes one shell tool so the model can inspect files.
|
|
capabilities=[WorkspaceShellCapability()],
|
|
# `tool_choice="required"` makes the demo more deterministic by forcing the model
|
|
# to look at the workspace instead of answering from prior assumptions.
|
|
model_settings=ModelSettings(tool_choice="required"),
|
|
)
|
|
|
|
|
|
def _require_modal_dependency() -> tuple[Any, Any]:
|
|
try:
|
|
from agents.extensions.sandbox import ModalSandboxClient, ModalSandboxClientOptions
|
|
except Exception as exc: # pragma: no cover - import path depends on optional extras
|
|
raise SystemExit(
|
|
"Modal-backed runs require the optional repo extra.\n"
|
|
"Install it with: uv sync --extra modal"
|
|
) from exc
|
|
|
|
return ModalSandboxClient, ModalSandboxClientOptions
|
|
|
|
|
|
def _path_resolves_to(path: str, target: Path) -> bool:
|
|
try:
|
|
return Path(path or ".").resolve() == target
|
|
except OSError:
|
|
return False
|
|
|
|
|
|
def _import_docker_from_env() -> Any:
|
|
script_dir = Path(__file__).resolve().parent
|
|
original_sys_path = sys.path[:]
|
|
try:
|
|
sys.path = [entry for entry in sys.path if not _path_resolves_to(entry, script_dir)]
|
|
from docker import from_env as docker_from_env # type: ignore[import-untyped]
|
|
except Exception as exc: # pragma: no cover - import path depends on local Docker setup
|
|
raise SystemExit(
|
|
f"Docker-backed runs failed to import the Docker SDK: {exc}\n"
|
|
"Install the repo dependencies with: make sync\n"
|
|
"If you are running this file directly, try:\n"
|
|
"uv run python -m examples.sandbox.basic --backend docker"
|
|
) from exc
|
|
finally:
|
|
sys.path = original_sys_path
|
|
|
|
return docker_from_env
|
|
|
|
|
|
def _require_docker_dependency() -> tuple[Any, Any, Any]:
|
|
docker_from_env = _import_docker_from_env()
|
|
from agents.sandbox.sandboxes.docker import DockerSandboxClient, DockerSandboxClientOptions
|
|
|
|
return docker_from_env, DockerSandboxClient, DockerSandboxClientOptions
|
|
|
|
|
|
async def _create_session(
|
|
*,
|
|
backend: Backend,
|
|
manifest: Manifest,
|
|
agent: SandboxAgent,
|
|
):
|
|
if backend == "docker":
|
|
docker_from_env, DockerSandboxClient, DockerSandboxClientOptions = (
|
|
_require_docker_dependency()
|
|
)
|
|
client = DockerSandboxClient(docker_from_env())
|
|
sandbox = await client.create(
|
|
manifest=manifest,
|
|
options=DockerSandboxClientOptions(image=DEFAULT_PYTHON_SANDBOX_IMAGE),
|
|
)
|
|
return client, sandbox
|
|
|
|
ModalSandboxClient, ModalSandboxClientOptions = _require_modal_dependency()
|
|
client = ModalSandboxClient()
|
|
sandbox = await client.create(
|
|
manifest=manifest,
|
|
options=ModalSandboxClientOptions(
|
|
app_name=DEFAULT_MODAL_APP_NAME,
|
|
workspace_persistence=DEFAULT_MODAL_WORKSPACE_PERSISTENCE,
|
|
),
|
|
)
|
|
return client, sandbox
|
|
|
|
|
|
async def main(
|
|
model: str,
|
|
question: str,
|
|
backend: Backend,
|
|
) -> None:
|
|
manifest = _build_manifest(backend)
|
|
agent = _build_agent(model=model, manifest=manifest, backend=backend)
|
|
client, sandbox = await _create_session(
|
|
backend=backend,
|
|
manifest=manifest,
|
|
agent=agent,
|
|
)
|
|
|
|
await sandbox.start()
|
|
print(await sandbox.ls("."))
|
|
|
|
try:
|
|
# `async with sandbox` keeps the example on the public session lifecycle API.
|
|
# `Runner` reuses the already-running session without starting it a second time.
|
|
async with sandbox:
|
|
# `Runner.run_streamed()` drives the model and yields text and tool events in real time.
|
|
result = Runner.run_streamed(
|
|
agent,
|
|
question,
|
|
run_config=RunConfig(
|
|
sandbox=SandboxRunConfig(session=sandbox),
|
|
workflow_name=f"{backend.title()} sandbox example",
|
|
),
|
|
)
|
|
saw_text_delta = False
|
|
saw_any_text = False
|
|
|
|
# The stream contains raw text deltas from the assistant plus structured tool events.
|
|
async for event in result.stream_events():
|
|
if event.type == "raw_response_event" and isinstance(
|
|
event.data, ResponseTextDeltaEvent
|
|
):
|
|
if not saw_text_delta:
|
|
print("assistant> ", end="", flush=True)
|
|
saw_text_delta = True
|
|
print(event.data.delta, end="", flush=True)
|
|
saw_any_text = True
|
|
continue
|
|
|
|
if event.type != "run_item_stream_event":
|
|
continue
|
|
|
|
banner = _stream_event_banner(event.name)
|
|
if banner is not None:
|
|
if saw_text_delta:
|
|
print()
|
|
saw_text_delta = False
|
|
print(banner)
|
|
|
|
if saw_text_delta:
|
|
print()
|
|
if not saw_any_text:
|
|
print(result.final_output)
|
|
finally:
|
|
await client.delete(sandbox)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--model", default="gpt-5.6-sol", help="Model name to use.")
|
|
parser.add_argument("--question", default=DEFAULT_QUESTION, help="Prompt to send to the agent.")
|
|
parser.add_argument(
|
|
"--backend",
|
|
default=DEFAULT_BACKEND,
|
|
choices=["docker", "modal"],
|
|
help="Sandbox backend to use for this example.",
|
|
)
|
|
args = parser.parse_args()
|
|
asyncio.run(
|
|
main(
|
|
args.model,
|
|
args.question,
|
|
cast(Backend, args.backend),
|
|
)
|
|
)
|