Transition from Reflex to the OpenAI Agents API

from openai.types.beta.environment_param import EnvironmentParamOpenAIHosted

from common import inline_file, run, workspace_session


async def main() -> None:
    environment: EnvironmentParamOpenAIHosted = {
        "type": "openai_hosted",
        "network": {"access": "disabled"},
        "files": [
            inline_file(
                "brief.txt", "Plan a synchronous Python service's move to async.\n"
            )
        ],
    }
    async with workspace_session(environment=environment) as work:
        await work.turn("Read brief.txt and write a five-step plan to outputs/plan.md.")
        await work.save("plan.md")


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

from common import run, workspace_session


async def main() -> None:
    async with workspace_session() as work:
        await work.turn(
            "Write a five-step async Python migration plan to outputs/plan.md."
        )
        await work.save("plan.md")
        await work.turn(
            "Read outputs/plan.md and write a checklist to outputs/checklist.md."
        )
        await work.save("checklist.md")


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

from common import default_agent, run, workspace_session

PERSONA = """You are a careful Python reviewer.
Prioritize correctness and explain the evidence for each finding.
Keep the summary concise. Mark assumptions explicitly.
"""


async def main() -> None:
    async with workspace_session(agent=default_agent(PERSONA)) as work:
        await work.turn(
            "Review this proposal: replace all requests calls with aiohttp without "
            "changing callers. Write risks and a safer plan to outputs/review.md."
        )
        await work.save("review.md")


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

import sys

from common import default_agent, inline_file, run, workspace_session

LOCAL_MCP = """from mcp.server.fastmcp import FastMCP
server = FastMCP("workspace_notes")
@server.tool()
def migration_note() -> str:
    return "Update async callers and retain timeout handling."
server.run(transport="stdio")
"""


def configuration(route: str):
    from openai.types.beta.agent_tool_param import AgentToolConfigParamMcp
    from openai.types.beta.environment_param import EnvironmentParamOpenAIHosted

    environment: EnvironmentParamOpenAIHosted = {
        "type": "openai_hosted",
        "network": {"access": "disabled"},
    }
    tool: AgentToolConfigParamMcp = {
        "type": "mcp",
        "server_label": "openai_docs",
        "transport": {
            "type": "http",
            "server_url": "https://developers.openai.com/mcp",
        },
        "connection_origin": "service",
        "required": True,
    }
    prompt = "Use openai_docs to explain session follow-ups in outputs/mcp-notes.md."
    if route == "environment":
        tool["connection_origin"] = "environment"
        environment["network"] = {
            "access": "restricted",
            "allowed_domains": ["developers.openai.com"],
        }
    elif route == "stdio":
        tool = {
            "type": "mcp",
            "server_label": "workspace_notes",
            "required": True,
            "transport": {
                "type": "stdio",
                "command": "/workspace/mcp-env/bin/python",
                "args": ["/workspace/notes_mcp.py"],
                "cwd": "/workspace",
            },
        }
        environment["network"] = {"access": "enabled"}
        environment["files"] = [inline_file("notes_mcp.py", LOCAL_MCP)]
        environment["setup_commands"] = [
            {
                "command": "python -m venv /workspace/mcp-env && "
                "/workspace/mcp-env/bin/pip install 'mcp>=1,<2'"
            }
        ]
        prompt = "Call migration_note and save its advice to outputs/mcp-notes.md."
    elif route != "service":
        raise ValueError("Choose service, environment, or stdio")
    agent = default_agent()
    agent["tools"] = [tool]
    return agent, environment, prompt


async def main() -> None:
    route = sys.argv[1] if len(sys.argv) > 1 else "service"
    agent, environment, prompt = configuration(route)
    async with workspace_session(agent=agent, environment=environment) as work:
        await work.turn(prompt)
        await work.save("mcp-notes.md")


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

import os
import sys

from openai import AsyncOpenAI

from common import (
    ExampleError,
    WorkspaceSession,
    default_agent,
    delete_session,
    run,
    wait_idle,
)


async def main() -> None:
    action = sys.argv[1] if len(sys.argv) > 1 else ""
    if action not in {"start", "continue", "cancel", "delete"}:
        raise ExampleError(
            "Choose start, continue SESSION_ID, cancel SESSION_ID, or delete SESSION_ID"
        )
    if action != "start" and len(sys.argv) != 3:
        raise ExampleError("Provide the saved session ID")
    if not os.environ.get("OPENAI_API_KEY"):
        raise ExampleError("Set OPENAI_API_KEY before running this example")
    async with AsyncOpenAI(timeout=60, max_retries=0) as client:
        if action == "start":
            session = await client.beta.agents.sessions.create(
                agent=default_agent(),
                environment={
                    "type": "openai_hosted",
                    "network": {"access": "disabled"},
                },
            )
            print(f"Keep this session ID: {session.id}", flush=True)
            work = WorkspaceSession(client, session)
            await work.turn("Write an async Python migration plan to outputs/plan.md.")
            await work.save("plan.md")
            return
        session_id = sys.argv[2]
        if action == "delete":
            await delete_session(client, session_id)
            return
        if action == "cancel":
            await client.beta.agents.sessions.events.create(
                session_id, events=[{"type": "agent.session.input.cancel"}]
            )
            await wait_idle(client, session_id)
            return
        session = await wait_idle(client, session_id)
        if session.environment.type != "openai_hosted":
            raise ExampleError("Expected a managed sandbox session")
        environment = await client.beta.agents.environments.retrieve(
            session.environment.id
        )
        if environment.status != "connected":
            raise ExampleError(
                "Workspace unavailable; download published artifacts and upload them into a new session"
            )
        work = WorkspaceSession(client, session)
        await work.turn(
            "Read outputs/plan.md; if missing, stop and explain. Otherwise write "
            "the next steps to outputs/next-steps.md."
        )
        await work.save("next-steps.md")


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

import os
import re
import sys

from openai import AsyncOpenAI
from openai.types.beta.environment_param import EnvironmentParamOpenAIHosted

from common import ExampleError, default_agent, run, workspace_session

GITHUB_MCP = "https://api.githubcopilot.com/mcp/"


async def main() -> None:
    mode = sys.argv[1] if len(sys.argv) > 1 else "mcp"
    if mode not in {"mcp", "api"}:
        raise ExampleError("Choose mcp or api")
    for name in ("OPENAI_API_KEY", "GITHUB_TOKEN"):
        if not os.environ.get(name):
            raise ExampleError(f"Set {name} before running this example")
    repository = os.environ.get("GITHUB_REPOSITORY", "")
    if mode == "mcp" and not re.fullmatch(r"[\w.-]+/[\w.-]+", repository):
        raise ExampleError("Set GITHUB_REPOSITORY to owner/repo")
    async with AsyncOpenAI(timeout=60, max_retries=0) as client:
        vault = await client.beta.agents.vaults.create(name="Transition example GitHub")
        print(f"Vault: {vault.id}", flush=True)
        succeeded = False
        try:
            agent = default_agent()
            environment: EnvironmentParamOpenAIHosted = {
                "type": "openai_hosted",
                "network": {"access": "disabled"},
            }
            if mode == "mcp":
                credential = await client.beta.agents.vaults.credentials.create(
                    vault.id,
                    name="GitHub MCP token",
                    auth={
                        "type": "static_bearer",
                        "mcp_server_url": GITHUB_MCP,
                        "token": os.environ["GITHUB_TOKEN"],
                    },
                )
                agent["tools"] = [
                    {
                        "type": "mcp",
                        "server_label": "github",
                        "transport": {"type": "http", "server_url": GITHUB_MCP},
                        "connection_origin": "service",
                        "credential_id": credential.id,
                        "allowed_tools": ["search_issues", "issue_read"],
                        "required": True,
                    }
                ]
                prompt = (
                    f"Use GitHub MCP to find up to five open issues in {repository}. "
                    "Write their titles and URLs to outputs/github.md. Do not change anything."
                )
            else:
                await client.beta.agents.vaults.credentials.create(
                    vault.id,
                    name="GitHub API token",
                    auth={
                        "type": "environment_variable",
                        "secret_name": "GITHUB_TOKEN",
                        "secret_value": os.environ["GITHUB_TOKEN"],
                        "networking": {
                            "type": "limited",
                            "allowed_hosts": ["api.github.com"],
                        },
                    },
                )
                environment["network"] = {
                    "access": "restricted",
                    "allowed_domains": ["api.github.com"],
                }
                prompt = (
                    "Run curl --fail --silent --show-error https://api.github.com/user "
                    '-H "Authorization: Bearer $GITHUB_TOKEN" and write only the returned '
                    "login field to outputs/github.md. Do not print environment variables."
                )
            async with workspace_session(
                agent=agent, environment=environment, vault_ids=[vault.id]
            ) as work:
                await work.turn(prompt)
                await work.save("github.md")
            succeeded = True
        finally:
            try:
                await client.beta.agents.vaults.delete(vault.id)
            except Exception:  # noqa: BLE001
                message = f"Vault cleanup failed; delete vault {vault.id} later"
                if succeeded:
                    raise ExampleError(message) from None
                print(message, file=sys.stderr)


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

import asyncio
import base64
import os
import sys
from collections.abc import AsyncIterator, Coroutine
from contextlib import asynccontextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Any

from openai import AsyncOpenAI, ConflictError
from openai.types.beta.agent_session import AgentSession
from openai.types.beta.agents.session_create_params import Agent
from openai.types.beta.environment_param import EnvironmentParamOpenAIHosted
from openai.types.beta.hosted_environment_file_param import (
    HostedEnvironmentFileParamInline,
)


class ExampleError(RuntimeError):
    """Messages must not contain credentials, file contents, or connection URLs."""


def default_agent(
    instructions: str = "Write clear files and verify your work.",
) -> Agent:
    return {
        "model": os.environ.get("OPENAI_AGENT_MODEL", "gpt-6-astra"),
        "instructions": instructions,
    }


def inline_file(filename: str, contents: str) -> HostedEnvironmentFileParamInline:
    return {
        "type": "inline",
        "path": f"/workspace/{filename}",
        "data": base64.b64encode(contents.encode()).decode(),
    }


async def wait_idle(client: AsyncOpenAI, session_id: str) -> AgentSession:
    async with asyncio.timeout(90):
        while True:
            session = await client.beta.agents.sessions.retrieve(session_id)
            if session.status == "idle":
                return session
            if session.status in {"failed", "requires_action"}:
                raise ExampleError(
                    "Session needs attention; inspect its saved state before continuing"
                )
            await asyncio.sleep(1)


async def delete_session(client: AsyncOpenAI, session_id: str) -> None:
    async with asyncio.timeout(90):
        session = await client.beta.agents.sessions.retrieve(session_id)
        if session.status == "in_progress":
            await client.beta.agents.sessions.events.create(
                session_id, events=[{"type": "agent.session.input.cancel"}]
            )
        for attempt in range(6):
            try:
                await client.beta.agents.sessions.delete(session_id)
                return
            except ConflictError:
                if attempt == 5:
                    raise ExampleError(
                        "Cleanup did not finish; delete the printed session ID later"
                    ) from None
                await asyncio.sleep(2**attempt)


@dataclass
class WorkspaceSession:
    client: AsyncOpenAI
    session: AgentSession
    turn_id: str | None = None

    async def turn(self, prompt: str) -> None:
        self.turn_id = None
        async with asyncio.timeout(300):
            async with self.client.beta.agents.sessions.stream(
                self.session.id, input=prompt
            ) as events:
                completed = None
                async for event in events:
                    if (
                        event.type == "agent.session.turn.failed"
                        or event.type == "agent.session.turn.cancelled"
                    ):
                        if event.turn.subagent_id is not None:
                            continue
                        raise ExampleError(f"Turn failed: {event.type}")
                    if event.type in {
                        "error",
                        "agent.session.failed",
                        "agent.session.environment.failed",
                    }:
                        raise ExampleError(f"Session failed: {event.type}")
                    if (
                        event.type == "agent.session.turn.completed"
                        and event.turn.subagent_id is None
                    ):
                        completed = event.turn.id
                    if event.type == "agent.session.idle" and completed is not None:
                        self.turn_id = completed
                        return
                raise ExampleError(
                    "Stream ended before main-turn completion and session idle"
                )

    async def save(self, filename: str) -> Path:
        if self.turn_id is None:
            raise ExampleError("Complete a turn before downloading its artifact")
        directory = Path(__file__).parent / "artifacts" / self.session.id
        directory.mkdir(parents=True, exist_ok=True)
        target = directory / filename
        async for artifact in self.client.beta.agents.sessions.artifacts.list(
            self.session.id
        ):
            if (
                artifact.turn_id != self.turn_id
                or artifact.path != f"/workspace/outputs/{filename}"
            ):
                continue
            async with (
                self.client.beta.agents.sessions.artifacts.with_streaming_response.content(
                    artifact.id, session_id=self.session.id
                ) as response
            ):
                await response.stream_to_file(target)
            if not target.read_bytes().strip():
                raise ExampleError("Expected a non-empty artifact")
            print(f"Saved {target}")
            return target
        raise ExampleError("Expected artifact missing; inspect the saved session items")


@asynccontextmanager
async def workspace_session(
    agent: Agent | None = None,
    environment: EnvironmentParamOpenAIHosted | None = None,
    vault_ids: list[str] | None = None,
) -> AsyncIterator[WorkspaceSession]:
    if not os.environ.get("OPENAI_API_KEY"):
        raise ExampleError("Set OPENAI_API_KEY before running this example")
    async with AsyncOpenAI(timeout=60, max_retries=0) as client:
        session = await client.beta.agents.sessions.create(
            agent=agent if agent is not None else default_agent(),
            environment=environment
            if environment is not None
            else {"type": "openai_hosted", "network": {"access": "disabled"}},
            vault_ids=vault_ids or [],
        )
        print(f"Session: {session.id}", flush=True)
        succeeded = False
        try:
            yield WorkspaceSession(client, session)
            succeeded = True
        finally:
            try:
                await delete_session(client, session.id)
            except Exception:
                if succeeded:
                    raise
                print(
                    f"Session cleanup failed; delete session {session.id} later",
                    file=sys.stderr,
                )


def run(example: Coroutine[Any, Any, None]) -> None:
    try:
        asyncio.run(example)
    except KeyboardInterrupt:
        print(
            "Interrupted; use the printed session ID to check or clean up work.",
            file=sys.stderr,
        )
        raise SystemExit(130) from None
    except ExampleError as error:
        print(f"Example failed: {error}", file=sys.stderr)
        raise SystemExit(1) from None
    except Exception as error:  # noqa: BLE001
        # SDK response bodies can contain credentials or connection details.
        print(
            f"Example failed ({type(error).__name__}); inspect session state.",
            file=sys.stderr,
        )
        raise SystemExit(1) from None