codexcw — Python examples

The PyPI package is codexcw (in bindings/python), a native extension backed by the Rust core.

pip install codexcw

Runners drive Codex (the default), Claude Code, or Grok Build; the selected agent's executable must be on PATH and authenticated — codex new enough to support codex exec --json, claude new enough to support --output-format stream-json, and grok new enough to support streaming-messages-json. Codex defaults are automation-friendly: read-only sandbox, approval never, ephemeral sessions, color off, git-check skipped.

Sync and async APIs

The package ships both: a synchronous API in codexcw and an asyncio API in codexcw.aio with the same shapes. Every recipe below is shown in both.

# Synchronous
from codexcw import Runner, Request

runner = Runner()
result = runner.run(Request(prompt="diga oi"))
print(result.final_message)

# Asynchronous (codexcw.aio)
import asyncio
from codexcw import Request
from codexcw.aio import Runner

async def main():
    runner = Runner()
    result = await runner.run(Request(prompt="diga oi"))
    print(result.final_message)

asyncio.run(main())

Streaming events

# Sync
from codexcw import Runner, Request

runner = Runner()
session = runner.start(Request(prompt="resuma este repo"))
for event in session.events():
    if event.type == "item.completed" and event.item.type == "agent_message":
        print(event.item.text)
result = session.wait()
print("tokens:", result.usage.total_tokens)

# Async
from codexcw import Request
from codexcw.aio import Runner

runner = Runner()
session = await runner.start(Request(prompt="resuma este repo"))
async for event in session.events():
    if event.type == "item.completed" and event.item.type == "agent_message":
        print(event.item.text)
result = await session.wait()
print("tokens:", result.usage.total_tokens)

Per-event callback

on_event runs for every event. Raising cancels the run.

# Sync
def on_event(event):
    if event.type == "item.completed" and event.item.type == "command_execution":
        print("$", event.item.command)

runner.run(Request(prompt="trabalhe"), on_event)

# Async — the callback may be a plain function or a coroutine.
async def on_event(event):
    if event.type == "item.completed" and event.item.type == "command_execution":
        print("$", event.item.command)

await runner.run(Request(prompt="trabalhe"), on_event)

# A callback that aborts on the first command execution.
from codexcw import CodexcwError

def stop_on_command(event):
    if event.type == "item.started" and event.item.type == "command_execution":
        raise RuntimeError("stop")

try:
    runner.run(Request(prompt="..."), stop_on_command)
except RuntimeError as err:
    print("cancelled:", err)

Resume a session

# Sync
first = runner.run(Request(prompt="crie um arquivo TODO.md"))
thread_id = first.thread_id

second = runner.run(Request(prompt="agora adicione 3 itens ao TODO.md", resume_id=thread_id))
print(second.final_message)

# Async
first = await runner.run(Request(prompt="crie um arquivo TODO.md"))
second = await runner.run(Request(prompt="continue", resume_id=first.thread_id))

# Resume the most recent thread, or disable cwd filtering with resume_all:
runner.run(Request(prompt="continue", resume_last=True))
runner.run(Request(prompt="continue", resume_id=thread_id, resume_all=True))

Resume runs reject dir, add_dirs, and profile — raised as a CodexcwError with kind == "invalidRequest".

Sandbox modes

# Read-only is the default. Let Codex write inside the workspace:
runner.run(Request(prompt="refatore o pacote foo", sandbox="workspace-write"))
await runner.run(Request(prompt="refatore o pacote foo", sandbox="workspace-write"))

# Remove sandbox filesystem restrictions:
runner.run(Request(prompt="...", sandbox="danger-full-access"))

Approval policies

# Defaults to "never". The safer interactive middle ground:
req = Request(prompt="...", sandbox="workspace-write", approval="on-request")
runner.run(req)              # sync
await runner.run(req)        # async

⚠️ Bypass sandbox and approvals

Danger. dangerously_bypass_sandbox runs Codex with --dangerously-bypass-approvals-and-sandbox: no sandbox, no approval prompts. Only use this in a disposable, fully-trusted environment.

runner.run(Request(prompt="...", dangerously_bypass_sandbox=True))
await runner.run(Request(prompt="...", dangerously_bypass_sandbox=True))

# Run enabled hooks without persisted trust:
runner.run(Request(prompt="...", dangerously_bypass_hooks=True))

Run many with bounded concurrency

# Sync
group = runner.run_many(
    [Request(prompt="review A"), Request(prompt="review B"), Request(prompt="review C")],
    max_concurrent=2,
)
for run_event in group.events():
    print(f"[{run_event.index}] {run_event.event.type}")

for r in group.wait():
    if r.error is not None:
        print(f"[{r.index}] failed:", r.error.kind, r.error)
    else:
        print(f"[{r.index}] {r.result.final_message}")

# Async
group = await runner.run_many(
    [Request(prompt="review A"), Request(prompt="review B"), Request(prompt="review C")],
    max_concurrent=2,
)
async for run_event in group.events():
    print(f"[{run_event.index}] {run_event.event.type}")
results = await group.wait()

Config overrides

from codexcw import ConfigOverride

req = Request(prompt="...", config=[
    ConfigOverride(key="model_reasoning_effort", value='"high"'),
    ConfigOverride(key="tools.web_search", value="true"),
])
runner.run(req)
await runner.run(req)

Fast mode (`/fast`)

Codex Fast mode uses the priority service tier.

req = Request(
    prompt="...",
    config=[ConfigOverride(key="service_tier", value='"priority"')],
)
runner.run(req)
await runner.run(req)

Structured output

import json

schema = json.dumps({
    "type": "object",
    "properties": {"summary": {"type": "string"}},
    "required": ["summary"],
})

req = Request(prompt="resuma o repo como JSON", output_schema=schema, output_last_message_path="out.json")
result = runner.run(req)             # sync
print(json.loads(result.final_message))

result = await runner.run(req)       # async

Working directory and extra dirs

req = Request(prompt="...", dir="/work/project", add_dirs=["/work/shared", "/work/vendor"])
runner.run(req)
await runner.run(req)

Model and profile

req = Request(prompt="...", model="o3", profile="work")
runner.run(req)
await runner.run(req)

Claude Code agent

The runner also wraps Claude Code's non-interactive mode (claude -p --output-format stream-json). Select it with the agent keyword; the claude executable must be on PATH and authenticated. Events are normalized into the same event model — thread.started carries the Claude session id, tool calls become item.started/item.completed pairs, and the final result maps to turn.completed — with raw always keeping the original Claude JSON line.

import codexcw

runner = codexcw.Runner(agent=codexcw.AGENT_CLAUDE)

result = runner.run(
    codexcw.Request(
        prompt="crie um arquivo TODO.md",
        model=codexcw.CLAUDE_MODEL_HAIKU,  # "haiku", "sonnet", or "opus"
        permission_mode=codexcw.PERMISSION_ACCEPT_EDITS,
    )
)

print("tokens:", result.usage.total_tokens)
print("cache writes:", result.usage.cache_creation_input_tokens)
print("cost (USD):", result.usage.total_cost_usd)
print("per-model usage:", result.usage.model_usage)

# Tool filters and resume work per request:
runner.run(
    codexcw.Request(
        prompt="rode os testes",
        model=codexcw.CLAUDE_MODEL_SONNET,
        allowed_tools=["Bash(pytest *)", "Read"],
        disallowed_tools=["WebSearch"],
    )
)

first = runner.run(codexcw.Request(prompt="lembre disto", persistent=True))
runner.run(
    codexcw.Request(
        prompt="continue",
        resume_id=first.thread_id,  # or resume_last=True
        persistent=True,
    )
)

Claude runs support dir (applied as the process working directory), add_dirs, output_schema/output_schema_path (passed as --json-schema), and dangerously_bypass_sandbox (passed as --dangerously-skip-permissions). permission_mode, allowed_tools, and disallowed_tools are claude-only; codex-only fields (sandbox, approval, profile, config, images, feature flags) raise a typed invalidRequest error on a claude runner. The exported permission constants cover acceptEdits, auto, bypassPermissions, manual, dontAsk, and plan.

Grok Build agent

Select Grok Build with AGENT_GROK. The wrapper invokes grok with --no-auto-update, --output-format streaming-messages-json, --verbatim, and a temporary --prompt-file that is removed after the process exits.

import codexcw

grok = codexcw.Runner(agent=codexcw.AGENT_GROK)
result = grok.run(
    codexcw.Request(
        prompt="revise este repositório",
        dir="/work/project",
        model="grok-code-fast-1",
        profile="reviewer",  # passed as --agent
        sandbox="workspace-write",
        permission_mode=codexcw.PERMISSION_ACCEPT_EDITS,
        allowed_tools=["Bash(pytest *)"],
        disallowed_tools=["WebSearch"],
        output_schema='{"type":"object"}',
    )
)
print(result.final_message)

# The asyncio facade accepts the same options.
async_grok = codexcw.aio.Runner(agent=codexcw.AGENT_GROK)

New Grok runs default to the read-only sandbox and dontAsk permission mode. workspace-write maps to Grok's workspace profile and danger-full-access maps to off. Approval never maps to dontAsk; the other approval policies map to Grok's default mode. An explicit permission_mode is mutually exclusive with approval. dangerously_bypass_sandbox selects --sandbox off with bypassPermissions.

resume_id and resume_last map to --resume and --continue. A resume request without an explicit sandbox omits the flag so Grok restores the session's saved sandbox. Grok always persists sessions, so persistent does not change its invocation. Grok does not expose an account-usage helper.

The prompt and stdin string are buffered into the temporary prompt file before launch. Grok's Messages stream reports reasoning text but no separate reasoning token count, so reasoning_output_tokens is zero. add_dirs, images, config, enable, disable, strict_config, ignore_user_config, ignore_rules, require_git_repo, output_last_message_path, dangerously_bypass_hooks, and resume_all raise a typed invalidRequest error because Grok has no equivalent option.

Stdin input

# Prompt via stdin only:
runner.run(Request(stdin="diga oi"))
await runner.run(Request(stdin="diga oi"))

# Prompt plus extra stdin context (wrapped in <stdin> markers):
runner.run(Request(prompt="resuma o diff abaixo", stdin=large_diff))

Custom executable and environment

runner = Runner(executable="/opt/codex/bin/codex", env={"CODEX_HOME": "/tmp/codex-home"})

Account usage and limits

get_account_usage reads account limits and credits through codex app-server. It accepts an executable override and child-process environment. CODEX_HOME defaults to ~/.codex when it is not set. timeout bounds each JSON-RPC request, in seconds; None uses the 10-second default.

# Sync
from codexcw import AccountUsageRequest, get_account_usage

usage = get_account_usage(AccountUsageRequest(env={"CODEX_HOME": "/tmp/codex-home"}, timeout=5.0))

if usage.account is not None:
    print("account:", usage.account.email)
if usage.rate_limits.primary is not None:
    print("primary used:", usage.rate_limits.primary.used_percent)
if usage.token_usage is not None:
    print("lifetime tokens:", usage.token_usage.summary.lifetime_tokens)

account and token_usage are None when codex answers those reads with a JSON-RPC error; transport errors and timeouts fail the whole call.

Claude account usage is available through the Claude Code /usage command:

from codexcw import ClaudeAccountUsageRequest, get_claude_account_usage

usage = get_claude_account_usage(ClaudeAccountUsageRequest(timeout=5.0))
print(usage.report)
for window in usage.windows:
    print(window.label, window.used_percent, window.resets_at)

raw keeps the original Claude JSON output, while windows contains the percentage-based limits parsed from the human-readable report.

# Async
from codexcw import AccountUsageRequest
from codexcw.aio import get_account_usage

usage = await get_account_usage(AccountUsageRequest(env={"CODEX_HOME": "/tmp/codex-home"}))

from codexcw import ClaudeAccountUsageRequest
from codexcw.aio import get_claude_account_usage

claude_usage = await get_claude_account_usage(ClaudeAccountUsageRequest(timeout=5.0))

Error handling

Failures raise a typed CodexcwError whose kind discriminates the cause.

from codexcw import CodexcwError

try:
    result = runner.run(Request(prompt="..."))   # or: await runner.run(...)
    print(result.final_message)
except CodexcwError as err:
    if err.kind == "exit":
        print(f"agent exited {err.code}: {err.stderr}")
    elif err.kind == "codex":
        print("codex reported:", err)
    elif err.kind == "claude":
        print("claude reported:", err)
    elif err.kind == "grok":
        print("grok reported:", err)
    elif err.kind == "decode":
        print(f"bad JSONL on line {err.line}")
    elif err.kind == "promptRequired":
        print("prompt or stdin is required")
    else:
        print(err.kind, err)

Cancellation

# Sync
session = runner.start(Request(prompt="..."))
session.cancel()           # stops the child process
session.wait()

# Async
session = await runner.start(Request(prompt="..."))
session.cancel()
await session.wait()

See the README for the cross-language overview and AGENTS.md for the project guide.

Navigation Menu

Uh oh!

codexcw — Python examples

Sync and async APIs

Streaming events

Per-event callback

Resume a session

Sandbox modes

Approval policies

⚠️ Bypass sandbox and approvals

Run many with bounded concurrency

Config overrides

Fast mode (`/fast`)

Structured output

Working directory and extra dirs

Model and profile

Claude Code agent

Grok Build agent

Stdin input

Custom executable and environment

Account usage and limits

Error handling

Cancellation

Uh oh!

FilesExpand file tree

python.md

Latest commit

History

python.md

File metadata and controls

codexcw — Python examples

Sync and async APIs

Streaming events

Per-event callback

Resume a session

Sandbox modes

Approval policies

⚠️ Bypass sandbox and approvals

Run many with bounded concurrency

Config overrides

Fast mode (/fast)

Structured output

Working directory and extra dirs

Model and profile

Claude Code agent

Grok Build agent

Stdin input

Custom executable and environment

Account usage and limits

Error handling

Cancellation

Fast mode (`/fast`)