diff --git a/sdk/python/README.md b/sdk/python/README.md index 1fa2e5a882..8ecdd029a3 100644 --- a/sdk/python/README.md +++ b/sdk/python/README.md @@ -1,38 +1,113 @@ # Codex App Server Python SDK (Experimental) -Experimental Python SDK for `codex app-server` JSON-RPC v2 over stdio, with a small default surface optimized for real scripts and apps. +Experimental Python SDK for `codex app-server` JSON-RPC v2. +The generated wire models come from the bundled v2 schema and use snake_case Python fields while preserving camelCase wire serialization. -The generated wire-model layer is currently sourced from the bundled v2 schema and exposed as Pydantic models with snake_case Python fields that serialize back to the app-server’s camelCase wire format. +It gives you a small typed API for: -## Install +- starting or resuming threads +- creating turns from Python +- streaming events or waiting for a final `TurnResult` +- using the same shape in sync and async code + +## Experimental + +This SDK is still experimental. + +- it is not published yet +- API details may still change before the first release +- packaging and release workflow are still evolving + +Use it for local development, dogfooding, and iteration inside this repo. Do not treat it as a stable public package yet. + +## What You Need + +- Python `>=3.10` +- local Codex auth/session already configured +- this repo checked out locally + +## Install From Source ```bash cd sdk/python python -m pip install -e . ``` +The package includes bundled Codex runtime binaries and automatically selects the binary for the current platform through `AppServerConfig().codex_bin`. + +## Core Model + +The public API is intentionally small: + +- `Codex` / `AsyncCodex`: session entrypoint +- `Thread` / `AsyncThread`: a conversation thread +- `Turn` / `AsyncTurn`: one user turn within a thread +- `TurnResult`: final status, text, items, and usage + +Typical flow: + +1. create a `Codex` client +2. start or resume a thread +3. create a turn from input +4. call `run()` or iterate `stream()` + ## Quickstart +### Sync + ```python from codex_app_server import Codex, TextInput with Codex() as codex: - thread = codex.thread_start(model="gpt-5") + thread = codex.thread_start( + model="gpt-5", + config={"model_reasoning_effort": "high"}, + ) result = thread.turn(TextInput("Say hello in one sentence.")).run() - print(result.text) + + print("status:", result.status) + print("text:", result.text) ``` -## Docs map +### Async -- Golden path tutorial: `docs/getting-started.md` -- API reference (signatures + behavior): `docs/api-reference.md` -- Common decisions and pitfalls: `docs/faq.md` -- Runnable examples index: `examples/README.md` -- Jupyter walkthrough notebook: `notebooks/sdk_walkthrough.ipynb` +```python +import asyncio -## Examples +from codex_app_server import AsyncCodex, TextInput -Start here: + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start( + model="gpt-5", + config={"model_reasoning_effort": "high"}, + ) + turn = await thread.turn(TextInput("Say hello in one sentence.")) + result = await turn.run() + + print("status:", result.status) + print("text:", result.text) + + +asyncio.run(main()) +``` + +## Current Limitations + +- Only one active `Turn.stream()` or `Turn.run()` consumer is supported per client instance. +- Starting a second active turn consumer on the same `Codex` or `AsyncCodex` raises `RuntimeError`. +- `Codex()` is eager and performs startup plus `initialize` in the constructor. + +## Behavior Notes + +- `AsyncCodex` is intended to be used with `async with AsyncCodex() as codex:`. +- `TurnResult.text` prefers streamed assistant deltas and falls back to completed raw response items when no deltas are emitted. +- For transient overload handling, use `retry_on_overload(...)`. + +## Learn By Example + +Runnable examples: ```bash cd sdk/python @@ -40,40 +115,32 @@ python examples/01_quickstart_constructor/sync.py python examples/01_quickstart_constructor/async.py ``` -## Bundled runtime binaries (out of the box) +More docs: -The SDK ships with platform-specific bundled binaries, so end users do not need updater scripts. +- Getting started: `docs/getting-started.md` +- API reference: `docs/api-reference.md` +- FAQ and pitfalls: `docs/faq.md` +- Examples index: `examples/README.md` +- Notebook walkthrough: `notebooks/sdk_walkthrough.ipynb` -Runtime binary source (single source, no fallback): +## Maintainer Workflow -- `src/codex_app_server/bin/darwin-arm64/codex` -- `src/codex_app_server/bin/darwin-x64/codex` -- `src/codex_app_server/bin/linux-arm64/codex` -- `src/codex_app_server/bin/linux-x64/codex` -- `src/codex_app_server/bin/windows-arm64/codex.exe` -- `src/codex_app_server/bin/windows-x64/codex.exe` - -## Maintainer workflow (refresh binaries/types) +Refresh bundled binaries and generated artifacts with: ```bash cd sdk/python python scripts/update_sdk_artifacts.py --channel stable --bundle-all-platforms -# or +``` + +or: + +```bash +cd sdk/python python scripts/update_sdk_artifacts.py --channel alpha --bundle-all-platforms ``` -This refreshes all bundled OS/arch binaries and regenerates protocol-derived Python types. +## Compatibility -## Compatibility and versioning - -- Package: `codex-app-server-sdk` -- Current SDK version in this repo: `0.2.0` -- Python: `>=3.10` +- Package name: `codex-app-server-sdk` +- SDK version in this repo: `0.2.0` - Target protocol: Codex `app-server` JSON-RPC v2 -- Recommendation: keep SDK and `codex` CLI reasonably up to date together - -## Notes - -- `Codex()` is eager and performs startup + `initialize` in the constructor. -- Use context managers (`with Codex() as codex:`) to ensure shutdown. -- For transient overload, use `codex_app_server.retry.retry_on_overload`. diff --git a/sdk/python/docs/api-reference.md b/sdk/python/docs/api-reference.md new file mode 100644 index 0000000000..597b069707 --- /dev/null +++ b/sdk/python/docs/api-reference.md @@ -0,0 +1,180 @@ +# Codex App Server SDK — API Reference + +Public surface of `codex_app_server` for app-server v2. + +This SDK surface is experimental. The current implementation intentionally allows only one active `Turn.stream()` or `Turn.run()` consumer per client instance at a time. + +## Package Entry + +```python +from codex_app_server import ( + Codex, + AsyncCodex, + Thread, + AsyncThread, + Turn, + AsyncTurn, + TurnResult, + InitializeResult, + Input, + InputItem, + TextInput, + ImageInput, + LocalImageInput, + SkillInput, + MentionInput, + ThreadItem, + TurnStatus, +) +``` + +- Version: `codex_app_server.__version__` +- Requires Python >= 3.10 + +## Codex (sync) + +```python +Codex(config: AppServerConfig | None = None) +``` + +Properties/methods: + +- `metadata -> InitializeResult` +- `close() -> None` +- `thread_start(*, approval_policy=None, base_instructions=None, config=None, cwd=None, developer_instructions=None, ephemeral=None, model=None, model_provider=None, personality=None, sandbox=None) -> Thread` +- `thread_list(*, archived=None, cursor=None, cwd=None, limit=None, model_providers=None, sort_key=None, source_kinds=None) -> ThreadListResponse` +- `thread_resume(thread_id: str, *, approval_policy=None, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, personality=None, sandbox=None) -> Thread` +- `thread_fork(thread_id: str, *, approval_policy=None, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, sandbox=None) -> Thread` +- `thread_archive(thread_id: str) -> ThreadArchiveResponse` +- `thread_unarchive(thread_id: str) -> Thread` +- `models(*, include_hidden: bool = False) -> ModelListResponse` + +Context manager: + +```python +with Codex() as codex: + ... +``` + +## AsyncCodex (async parity) + +```python +AsyncCodex(config: AppServerConfig | None = None) +``` + +Properties/methods: + +- `metadata -> InitializeResult` +- `close() -> Awaitable[None]` +- `thread_start(*, approval_policy=None, base_instructions=None, config=None, cwd=None, developer_instructions=None, ephemeral=None, model=None, model_provider=None, personality=None, sandbox=None) -> Awaitable[AsyncThread]` +- `thread_list(*, archived=None, cursor=None, cwd=None, limit=None, model_providers=None, sort_key=None, source_kinds=None) -> Awaitable[ThreadListResponse]` +- `thread_resume(thread_id: str, *, approval_policy=None, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, personality=None, sandbox=None) -> Awaitable[AsyncThread]` +- `thread_fork(thread_id: str, *, approval_policy=None, base_instructions=None, config=None, cwd=None, developer_instructions=None, model=None, model_provider=None, sandbox=None) -> Awaitable[AsyncThread]` +- `thread_archive(thread_id: str) -> Awaitable[ThreadArchiveResponse]` +- `thread_unarchive(thread_id: str) -> Awaitable[AsyncThread]` +- `models(*, include_hidden: bool = False) -> Awaitable[ModelListResponse]` + +Async context manager: + +```python +async with AsyncCodex() as codex: + ... +``` + +## Thread / AsyncThread + +`Thread` and `AsyncThread` share the same shape and intent. + +### Thread + +- `turn(input: Input, *, approval_policy=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox_policy=None, summary=None) -> Turn` +- `read(*, include_turns: bool = False) -> ThreadReadResponse` +- `set_name(name: str) -> ThreadSetNameResponse` +- `compact() -> ThreadCompactStartResponse` + +### AsyncThread + +- `turn(input: Input, *, approval_policy=None, cwd=None, effort=None, model=None, output_schema=None, personality=None, sandbox_policy=None, summary=None) -> Awaitable[AsyncTurn]` +- `read(*, include_turns: bool = False) -> Awaitable[ThreadReadResponse]` +- `set_name(name: str) -> Awaitable[ThreadSetNameResponse]` +- `compact() -> Awaitable[ThreadCompactStartResponse]` + +## Turn / AsyncTurn + +### Turn + +- `steer(input: Input) -> TurnSteerResponse` +- `interrupt() -> TurnInterruptResponse` +- `stream() -> Iterator[Notification]` +- `run() -> TurnResult` + +Behavior notes: + +- `stream()` and `run()` are exclusive per client instance in the current experimental build +- starting a second turn consumer on the same `Codex` instance raises `RuntimeError` + +### AsyncTurn + +- `steer(input: Input) -> Awaitable[TurnSteerResponse]` +- `interrupt() -> Awaitable[TurnInterruptResponse]` +- `stream() -> AsyncIterator[Notification]` +- `run() -> Awaitable[TurnResult]` + +Behavior notes: + +- `stream()` and `run()` are exclusive per client instance in the current experimental build +- starting a second turn consumer on the same `AsyncCodex` instance raises `RuntimeError` + +## TurnResult + +```python +@dataclass +class TurnResult: + thread_id: str + turn_id: str + status: TurnStatus + error: TurnError | None + text: str + items: list[ThreadItem] + usage: ThreadTokenUsageUpdatedNotification | None +``` + +## Inputs + +```python +@dataclass class TextInput: text: str +@dataclass class ImageInput: url: str +@dataclass class LocalImageInput: path: str +@dataclass class SkillInput: name: str; path: str +@dataclass class MentionInput: name: str; path: str + +InputItem = TextInput | ImageInput | LocalImageInput | SkillInput | MentionInput +Input = list[InputItem] | InputItem +``` + +## Retry + errors + +```python +from codex_app_server import ( + retry_on_overload, + JsonRpcError, + MethodNotFoundError, + InvalidParamsError, + ServerBusyError, + is_retryable_error, +) +``` + +- `retry_on_overload(...)` retries transient overload errors with exponential backoff + jitter. +- `is_retryable_error(exc)` checks if an exception is transient/overload-like. + +## Example + +```python +from codex_app_server import Codex, TextInput + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + result = thread.turn(TextInput("Say hello in one sentence.")).run() + print(result.text) +``` diff --git a/sdk/python/docs/faq.md b/sdk/python/docs/faq.md index 410c210b27..52b7d254f6 100644 --- a/sdk/python/docs/faq.md +++ b/sdk/python/docs/faq.md @@ -8,24 +8,42 @@ ## `run()` vs `stream()` -- `Turn.run()` is the easiest path. It consumes events until completion and returns `TurnResult`. -- `Turn.stream()` yields raw notifications (`Notification`) so you can react event-by-event. +- `Turn.run()` / `AsyncTurn.run()` is the easiest path. It consumes events until completion and returns `TurnResult`. +- `Turn.stream()` / `AsyncTurn.stream()` yields raw notifications (`Notification`) so you can react event-by-event. Choose `run()` for most apps. Choose `stream()` for progress UIs, custom timeout logic, or custom parsing. ## Sync vs async clients -- `Codex` is the minimal sync SDK and best default. -- `AsyncAppServerClient` wraps the sync transport with `asyncio.to_thread(...)` for async-friendly call sites. +- `Codex` is the sync public API. +- `AsyncCodex` is an async replica of the same public API shape. If your app is not already async, stay with `Codex`. -## `thread(...)` vs `thread_resume(...)` +## Public kwargs are snake_case -- `codex.thread(thread_id)` only binds a local helper to an existing thread ID. -- `codex.thread_resume(thread_id, ...)` performs a `thread/resume` RPC and can apply overrides (model, instructions, sandbox, etc.). +Public API keyword names are snake_case. The SDK still maps them to wire camelCase under the hood. -Use `thread(...)` for simple continuation. Use `thread_resume(...)` when you need explicit resume semantics or override fields. +If you are migrating older code, update these names: + +- `approvalPolicy` -> `approval_policy` +- `baseInstructions` -> `base_instructions` +- `developerInstructions` -> `developer_instructions` +- `modelProvider` -> `model_provider` +- `modelProviders` -> `model_providers` +- `sortKey` -> `sort_key` +- `sourceKinds` -> `source_kinds` +- `outputSchema` -> `output_schema` +- `sandboxPolicy` -> `sandbox_policy` + +## Why only `thread_start(...)` and `thread_resume(...)`? + +The public API keeps only explicit lifecycle calls: + +- `thread_start(...)` to create new threads +- `thread_resume(thread_id, ...)` to continue existing threads + +This avoids duplicate ways to do the same operation and keeps behavior explicit. ## Why does constructor fail? @@ -49,7 +67,7 @@ python scripts/update_sdk_artifacts.py --channel stable --bundle-all-platforms A turn is complete only when `turn/completed` arrives for that turn ID. - `run()` waits for this automatically. -- With `stream()`, make sure you keep consuming notifications until completion. +- With `stream()`, keep consuming notifications until completion. ## How do I retry safely? @@ -60,6 +78,6 @@ Do not blindly retry all errors. For `InvalidParamsError` or `MethodNotFoundErro ## Common pitfalls - Starting a new thread for every prompt when you wanted continuity. -- Forgetting to `close()` (or not using `with Codex() as codex:`). +- Forgetting to `close()` (or not using context managers). - Ignoring `TurnResult.status` and `TurnResult.error`. -- Mixing SDK input classes with raw dicts incorrectly in minimal API paths. +- Mixing SDK input classes with raw dicts incorrectly. diff --git a/sdk/python/docs/getting-started.md b/sdk/python/docs/getting-started.md index a3d52afdf8..63904cfe68 100644 --- a/sdk/python/docs/getting-started.md +++ b/sdk/python/docs/getting-started.md @@ -1,6 +1,8 @@ # Getting Started -This is the fastest path from install to a multi-turn thread using the minimal SDK surface. +This is the fastest path from install to a multi-turn thread using the public SDK surface. + +The SDK is experimental. Treat the API, bundled runtime strategy, and packaging details as unstable until the first public release. ## 1) Install @@ -15,9 +17,9 @@ Requirements: - Python `>=3.10` - bundled runtime binary for your platform (shipped in package) -- Local Codex auth/session configured +- local Codex auth/session configured -## 2) Run your first turn +## 2) Run your first turn (sync) ```python from codex_app_server import Codex, TextInput @@ -25,7 +27,7 @@ from codex_app_server import Codex, TextInput with Codex() as codex: print("Server:", codex.metadata.server_name, codex.metadata.server_version) - thread = codex.thread_start(model="gpt-5") + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) result = thread.turn(TextInput("Say hello in one sentence.")).run() print("Thread:", result.thread_id) @@ -39,6 +41,7 @@ What happened: - `Codex()` started and initialized `codex app-server`. - `thread_start(...)` created a thread. - `turn(...).run()` consumed events until `turn/completed` and returned a `TurnResult`. +- one client can have only one active `Turn.stream()` / `Turn.run()` consumer at a time in the current experimental build ## 3) Continue the same thread (multi-turn) @@ -46,7 +49,7 @@ What happened: from codex_app_server import Codex, TextInput with Codex() as codex: - thread = codex.thread_start(model="gpt-5") + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) first = thread.turn(TextInput("Summarize Rust ownership in 2 bullets.")).run() second = thread.turn(TextInput("Now explain it to a Python developer.")).run() @@ -55,7 +58,25 @@ with Codex() as codex: print("second:", second.text) ``` -## 4) Resume an existing thread +## 4) Async parity + +```python +import asyncio +from codex_app_server import AsyncCodex, TextInput + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = await thread.turn(TextInput("Continue where we left off.")) + result = await turn.run() + print(result.text) + + +asyncio.run(main()) +``` + +## 5) Resume an existing thread ```python from codex_app_server import Codex, TextInput @@ -63,12 +84,12 @@ from codex_app_server import Codex, TextInput THREAD_ID = "thr_123" # replace with a real id with Codex() as codex: - thread = codex.thread(THREAD_ID) + thread = codex.thread_resume(THREAD_ID) result = thread.turn(TextInput("Continue where we left off.")).run() print(result.text) ``` -## 5) Next stops +## 6) Next stops - API surface and signatures: `docs/api-reference.md` - Common decisions/pitfalls: `docs/faq.md` diff --git a/sdk/python/examples/01_quickstart_constructor/async.py b/sdk/python/examples/01_quickstart_constructor/async.py new file mode 100644 index 0000000000..e3aa5a5a5c --- /dev/null +++ b/sdk/python/examples/01_quickstart_constructor/async.py @@ -0,0 +1,30 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, TextInput + + +async def main() -> None: + async with AsyncCodex() as codex: + print("Server:", codex.metadata.server_name, codex.metadata.server_version) + + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = await thread.turn(TextInput("Say hello in one sentence.")) + result = await turn.run() + + print("Status:", result.status) + print("Text:", result.text) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/01_quickstart_constructor/sync.py b/sdk/python/examples/01_quickstart_constructor/sync.py new file mode 100644 index 0000000000..a1e6478411 --- /dev/null +++ b/sdk/python/examples/01_quickstart_constructor/sync.py @@ -0,0 +1,20 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, TextInput + +with Codex() as codex: + print("Server:", codex.metadata.server_name, codex.metadata.server_version) + + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + result = thread.turn(TextInput("Say hello in one sentence.")).run() + print("Status:", result.status) + print("Text:", result.text) diff --git a/sdk/python/examples/02_turn_run/async.py b/sdk/python/examples/02_turn_run/async.py new file mode 100644 index 0000000000..b93d893041 --- /dev/null +++ b/sdk/python/examples/02_turn_run/async.py @@ -0,0 +1,37 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, TextInput + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = await thread.turn(TextInput("Give 3 bullets about SIMD.")) + result = await turn.run() + + print("thread_id:", result.thread_id) + print("turn_id:", result.turn_id) + print("status:", result.status) + if result.error is not None: + print("error:", result.error) + print("text:", result.text) + print("items.count:", len(result.items)) + if result.usage is None: + raise RuntimeError("missing usage for completed turn") + print("usage.thread_id:", result.usage.thread_id) + print("usage.turn_id:", result.usage.turn_id) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/02_turn_run/sync.py b/sdk/python/examples/02_turn_run/sync.py new file mode 100644 index 0000000000..256c4ec7d7 --- /dev/null +++ b/sdk/python/examples/02_turn_run/sync.py @@ -0,0 +1,28 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, TextInput + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + result = thread.turn(TextInput("Give 3 bullets about SIMD.")).run() + + print("thread_id:", result.thread_id) + print("turn_id:", result.turn_id) + print("status:", result.status) + if result.error is not None: + print("error:", result.error) + print("text:", result.text) + print("items.count:", len(result.items)) + if result.usage is None: + raise RuntimeError("missing usage for completed turn") + print("usage.thread_id:", result.usage.thread_id) + print("usage.turn_id:", result.usage.turn_id) diff --git a/sdk/python/examples/03_turn_stream_events/async.py b/sdk/python/examples/03_turn_stream_events/async.py new file mode 100644 index 0000000000..109f6c9961 --- /dev/null +++ b/sdk/python/examples/03_turn_stream_events/async.py @@ -0,0 +1,44 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, TextInput + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = await thread.turn(TextInput("Count from 1 to 200 with commas, then one summary sentence.")) + + # Best effort controls: models can finish quickly, so races are expected. + try: + _ = await turn.steer(TextInput("Keep it brief and stop after 20 numbers.")) + print("steer: sent") + except Exception as exc: + print("steer: skipped", type(exc).__name__) + + try: + _ = await turn.interrupt() + print("interrupt: sent") + except Exception as exc: + print("interrupt: skipped", type(exc).__name__) + + event_count = 0 + async for event in turn.stream(): + event_count += 1 + print(event.method, event.payload) + + print("events.count:", event_count) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/03_turn_stream_events/sync.py b/sdk/python/examples/03_turn_stream_events/sync.py new file mode 100644 index 0000000000..14293cb42d --- /dev/null +++ b/sdk/python/examples/03_turn_stream_events/sync.py @@ -0,0 +1,36 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, TextInput + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = thread.turn(TextInput("Count from 1 to 200 with commas, then one summary sentence.")) + + # Best effort controls: models can finish quickly, so races are expected. + try: + _ = turn.steer(TextInput("Keep it brief and stop after 20 numbers.")) + print("steer: sent") + except Exception as exc: + print("steer: skipped", type(exc).__name__) + + try: + _ = turn.interrupt() + print("interrupt: sent") + except Exception as exc: + print("interrupt: skipped", type(exc).__name__) + + event_count = 0 + for event in turn.stream(): + event_count += 1 + print(event.method, event.payload) + + print("events.count:", event_count) diff --git a/sdk/python/examples/04_models_and_metadata/async.py b/sdk/python/examples/04_models_and_metadata/async.py new file mode 100644 index 0000000000..c8427c08c8 --- /dev/null +++ b/sdk/python/examples/04_models_and_metadata/async.py @@ -0,0 +1,28 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex + + +async def main() -> None: + async with AsyncCodex() as codex: + print("metadata:", codex.metadata) + + models = await codex.models(include_hidden=True) + print("models.count:", len(models.data)) + if models.data: + print("first model id:", models.data[0].id) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/04_models_and_metadata/sync.py b/sdk/python/examples/04_models_and_metadata/sync.py new file mode 100644 index 0000000000..c6c3720d75 --- /dev/null +++ b/sdk/python/examples/04_models_and_metadata/sync.py @@ -0,0 +1,20 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex + +with Codex() as codex: + print("metadata:", codex.metadata) + + models = codex.models() + print("models.count:", len(models.data)) + if models.data: + print("first model id:", models.data[0].id) diff --git a/sdk/python/examples/05_existing_thread/async.py b/sdk/python/examples/05_existing_thread/async.py new file mode 100644 index 0000000000..605f88a52d --- /dev/null +++ b/sdk/python/examples/05_existing_thread/async.py @@ -0,0 +1,32 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, TextInput + + +async def main() -> None: + async with AsyncCodex() as codex: + original = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + first_turn = await original.turn(TextInput("Tell me one fact about Saturn.")) + first = await first_turn.run() + print("Created thread:", first.thread_id) + + resumed = await codex.thread_resume(first.thread_id) + second_turn = await resumed.turn(TextInput("Continue with one more fact.")) + second = await second_turn.run() + print(second.text) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/05_existing_thread/sync.py b/sdk/python/examples/05_existing_thread/sync.py new file mode 100644 index 0000000000..e6dd2f508d --- /dev/null +++ b/sdk/python/examples/05_existing_thread/sync.py @@ -0,0 +1,23 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, TextInput + +with Codex() as codex: + # Create an initial thread and turn so we have a real thread to resume. + original = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + first = original.turn(TextInput("Tell me one fact about Saturn.")).run() + print("Created thread:", first.thread_id) + + # Resume the existing thread by ID. + resumed = codex.thread_resume(first.thread_id) + second = resumed.turn(TextInput("Continue with one more fact.")).run() + print(second.text) diff --git a/sdk/python/examples/06_thread_lifecycle_and_controls/async.py b/sdk/python/examples/06_thread_lifecycle_and_controls/async.py new file mode 100644 index 0000000000..b809a772cd --- /dev/null +++ b/sdk/python/examples/06_thread_lifecycle_and_controls/async.py @@ -0,0 +1,70 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, TextInput + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + first = await (await thread.turn(TextInput("One sentence about structured planning."))).run() + second = await (await thread.turn(TextInput("Now restate it for a junior engineer."))).run() + + reopened = await codex.thread_resume(thread.id) + listing_active = await codex.thread_list(limit=20, archived=False) + reading = await reopened.read(include_turns=True) + + _ = await reopened.set_name("sdk-lifecycle-demo") + _ = await codex.thread_archive(reopened.id) + listing_archived = await codex.thread_list(limit=20, archived=True) + unarchived = await codex.thread_unarchive(reopened.id) + + resumed_info = "n/a" + try: + resumed = await codex.thread_resume( + unarchived.id, + model="gpt-5", + config={"model_reasoning_effort": "high"}, + ) + resumed_result = await (await resumed.turn(TextInput("Continue in one short sentence."))).run() + resumed_info = f"{resumed_result.turn_id} {resumed_result.status}" + except Exception as exc: + resumed_info = f"skipped({type(exc).__name__})" + + forked_info = "n/a" + try: + forked = await codex.thread_fork(unarchived.id, model="gpt-5") + forked_result = await (await forked.turn(TextInput("Take a different angle in one short sentence."))).run() + forked_info = f"{forked_result.turn_id} {forked_result.status}" + except Exception as exc: + forked_info = f"skipped({type(exc).__name__})" + + compact_info = "sent" + try: + _ = await unarchived.compact() + except Exception as exc: + compact_info = f"skipped({type(exc).__name__})" + + print("Lifecycle OK:", thread.id) + print("first:", first.turn_id, first.status) + print("second:", second.turn_id, second.status) + print("read.turns:", len(reading.thread.turns or [])) + print("list.active:", len(listing_active.data)) + print("list.archived:", len(listing_archived.data)) + print("resumed:", resumed_info) + print("forked:", forked_info) + print("compact:", compact_info) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/06_thread_lifecycle_and_controls/sync.py b/sdk/python/examples/06_thread_lifecycle_and_controls/sync.py new file mode 100644 index 0000000000..863d71275e --- /dev/null +++ b/sdk/python/examples/06_thread_lifecycle_and_controls/sync.py @@ -0,0 +1,63 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, TextInput + + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + first = thread.turn(TextInput("One sentence about structured planning.")).run() + second = thread.turn(TextInput("Now restate it for a junior engineer.")).run() + + reopened = codex.thread_resume(thread.id) + listing_active = codex.thread_list(limit=20, archived=False) + reading = reopened.read(include_turns=True) + + _ = reopened.set_name("sdk-lifecycle-demo") + _ = codex.thread_archive(reopened.id) + listing_archived = codex.thread_list(limit=20, archived=True) + unarchived = codex.thread_unarchive(reopened.id) + + resumed_info = "n/a" + try: + resumed = codex.thread_resume( + unarchived.id, + model="gpt-5", + config={"model_reasoning_effort": "high"}, + ) + resumed_result = resumed.turn(TextInput("Continue in one short sentence.")).run() + resumed_info = f"{resumed_result.turn_id} {resumed_result.status}" + except Exception as exc: + resumed_info = f"skipped({type(exc).__name__})" + + forked_info = "n/a" + try: + forked = codex.thread_fork(unarchived.id, model="gpt-5") + forked_result = forked.turn(TextInput("Take a different angle in one short sentence.")).run() + forked_info = f"{forked_result.turn_id} {forked_result.status}" + except Exception as exc: + forked_info = f"skipped({type(exc).__name__})" + + compact_info = "sent" + try: + _ = unarchived.compact() + except Exception as exc: + compact_info = f"skipped({type(exc).__name__})" + + print("Lifecycle OK:", thread.id) + print("first:", first.turn_id, first.status) + print("second:", second.turn_id, second.status) + print("read.turns:", len(reading.thread.turns or [])) + print("list.active:", len(listing_active.data)) + print("list.archived:", len(listing_archived.data)) + print("resumed:", resumed_info) + print("forked:", forked_info) + print("compact:", compact_info) diff --git a/sdk/python/examples/07_image_and_text/async.py b/sdk/python/examples/07_image_and_text/async.py new file mode 100644 index 0000000000..4f0b446902 --- /dev/null +++ b/sdk/python/examples/07_image_and_text/async.py @@ -0,0 +1,35 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, ImageInput, TextInput + +REMOTE_IMAGE_URL = "https://raw.githubusercontent.com/github/explore/main/topics/python/python.png" + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = await thread.turn( + [ + TextInput("What is in this image? Give 3 bullets."), + ImageInput(REMOTE_IMAGE_URL), + ] + ) + result = await turn.run() + + print("Status:", result.status) + print(result.text) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/07_image_and_text/sync.py b/sdk/python/examples/07_image_and_text/sync.py new file mode 100644 index 0000000000..05a9dd16dd --- /dev/null +++ b/sdk/python/examples/07_image_and_text/sync.py @@ -0,0 +1,26 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, ImageInput, TextInput + +REMOTE_IMAGE_URL = "https://raw.githubusercontent.com/github/explore/main/topics/python/python.png" + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + result = thread.turn( + [ + TextInput("What is in this image? Give 3 bullets."), + ImageInput(REMOTE_IMAGE_URL), + ] + ).run() + + print("Status:", result.status) + print(result.text) diff --git a/sdk/python/examples/08_local_image_and_text/async.py b/sdk/python/examples/08_local_image_and_text/async.py new file mode 100644 index 0000000000..6930ac3b52 --- /dev/null +++ b/sdk/python/examples/08_local_image_and_text/async.py @@ -0,0 +1,38 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import AsyncCodex, LocalImageInput, TextInput + +IMAGE_PATH = Path(__file__).resolve().parents[1] / "assets" / "sample_scene.png" +if not IMAGE_PATH.exists(): + raise FileNotFoundError(f"Missing bundled image: {IMAGE_PATH}") + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + turn = await thread.turn( + [ + TextInput("Read this local image and summarize what you see in 2 bullets."), + LocalImageInput(str(IMAGE_PATH.resolve())), + ] + ) + result = await turn.run() + + print("Status:", result.status) + print(result.text) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/08_local_image_and_text/sync.py b/sdk/python/examples/08_local_image_and_text/sync.py new file mode 100644 index 0000000000..2cf0b80b1c --- /dev/null +++ b/sdk/python/examples/08_local_image_and_text/sync.py @@ -0,0 +1,29 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, LocalImageInput, TextInput + +IMAGE_PATH = Path(__file__).resolve().parents[1] / "assets" / "sample_scene.png" +if not IMAGE_PATH.exists(): + raise FileNotFoundError(f"Missing bundled image: {IMAGE_PATH}") + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + result = thread.turn( + [ + TextInput("Read this local image and summarize what you see in 2 bullets."), + LocalImageInput(str(IMAGE_PATH.resolve())), + ] + ).run() + + print("Status:", result.status) + print(result.text) diff --git a/sdk/python/examples/09_async_parity/sync.py b/sdk/python/examples/09_async_parity/sync.py new file mode 100644 index 0000000000..8af2694d8f --- /dev/null +++ b/sdk/python/examples/09_async_parity/sync.py @@ -0,0 +1,23 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import Codex, TextInput + +with Codex() as codex: + print("Server:", codex.metadata.server_name, codex.metadata.server_version) + + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = thread.turn(TextInput("Say hello in one sentence.")) + result = turn.run() + + print("Thread:", result.thread_id) + print("Turn:", result.turn_id) + print("Text:", result.text.strip()) diff --git a/sdk/python/examples/10_error_handling_and_retry/async.py b/sdk/python/examples/10_error_handling_and_retry/async.py new file mode 100644 index 0000000000..810ee7aca1 --- /dev/null +++ b/sdk/python/examples/10_error_handling_and_retry/async.py @@ -0,0 +1,91 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio +import random +from collections.abc import Awaitable, Callable +from typing import TypeVar + +from codex_app_server import ( + AsyncCodex, + JsonRpcError, + ServerBusyError, + TextInput, + TurnStatus, + is_retryable_error, +) + +ResultT = TypeVar("ResultT") + + +async def retry_on_overload_async( + op: Callable[[], Awaitable[ResultT]], + *, + max_attempts: int = 3, + initial_delay_s: float = 0.25, + max_delay_s: float = 2.0, + jitter_ratio: float = 0.2, +) -> ResultT: + if max_attempts < 1: + raise ValueError("max_attempts must be >= 1") + + delay = initial_delay_s + attempt = 0 + while True: + attempt += 1 + try: + return await op() + except Exception as exc: # noqa: BLE001 + if attempt >= max_attempts or not is_retryable_error(exc): + raise + jitter = delay * jitter_ratio + sleep_for = min(max_delay_s, delay) + random.uniform(-jitter, jitter) + if sleep_for > 0: + await asyncio.sleep(sleep_for) + delay = min(max_delay_s, delay * 2) + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + try: + result = await retry_on_overload_async( + _run_turn(thread, "Summarize retry best practices in 3 bullets."), + max_attempts=3, + initial_delay_s=0.25, + max_delay_s=2.0, + ) + except ServerBusyError as exc: + print("Server overloaded after retries:", exc.message) + print("Text:") + return + except JsonRpcError as exc: + print(f"JSON-RPC error {exc.code}: {exc.message}") + print("Text:") + return + + if result.status == TurnStatus.failed: + print("Turn failed:", result.error) + + print("Text:", result.text) + + +def _run_turn(thread, prompt: str): + async def _inner(): + turn = await thread.turn(TextInput(prompt)) + return await turn.run() + + return _inner + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/10_error_handling_and_retry/sync.py b/sdk/python/examples/10_error_handling_and_retry/sync.py new file mode 100644 index 0000000000..65bb3bd12e --- /dev/null +++ b/sdk/python/examples/10_error_handling_and_retry/sync.py @@ -0,0 +1,40 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import ( + Codex, + JsonRpcError, + ServerBusyError, + TextInput, + TurnStatus, + retry_on_overload, +) + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + try: + result = retry_on_overload( + lambda: thread.turn(TextInput("Summarize retry best practices in 3 bullets.")).run(), + max_attempts=3, + initial_delay_s=0.25, + max_delay_s=2.0, + ) + except ServerBusyError as exc: + print("Server overloaded after retries:", exc.message) + print("Text:") + except JsonRpcError as exc: + print(f"JSON-RPC error {exc.code}: {exc.message}") + print("Text:") + else: + if result.status == TurnStatus.failed: + print("Turn failed:", result.error) + print("Text:", result.text) diff --git a/sdk/python/examples/11_cli_mini_app/async.py b/sdk/python/examples/11_cli_mini_app/async.py new file mode 100644 index 0000000000..9e78001d79 --- /dev/null +++ b/sdk/python/examples/11_cli_mini_app/async.py @@ -0,0 +1,96 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import ( + AsyncCodex, + TextInput, + ThreadTokenUsageUpdatedNotification, + TurnCompletedNotificationPayload, +) + + +def _status_value(status: object | None) -> str: + return str(getattr(status, "value", status)) + + +def _format_usage(usage: object | None) -> str: + if usage is None: + return "usage> (none)" + + last = getattr(usage, "last", None) + total = getattr(usage, "total", None) + if last is None or total is None: + return f"usage> {usage}" + + return ( + "usage>\n" + f" last: input={last.inputTokens} output={last.outputTokens} reasoning={last.reasoningOutputTokens} total={last.totalTokens} cached={last.cachedInputTokens}\n" + f" total: input={total.inputTokens} output={total.outputTokens} reasoning={total.reasoningOutputTokens} total={total.totalTokens} cached={total.cachedInputTokens}" + ) + + +async def main() -> None: + print("Codex async mini CLI. Type /exit to quit.") + + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + print("Thread:", thread.id) + + while True: + try: + user_input = (await asyncio.to_thread(input, "you> ")).strip() + except EOFError: + break + + if not user_input: + continue + if user_input in {"/exit", "/quit"}: + break + + turn = await thread.turn(TextInput(user_input)) + usage = None + status = None + error = None + printed_delta = False + + print("assistant> ", end="", flush=True) + async for event in turn.stream(): + payload = event.payload + if event.method == "item/agentMessage/delta": + delta = getattr(payload, "delta", "") + if delta: + print(delta, end="", flush=True) + printed_delta = True + continue + if isinstance(payload, ThreadTokenUsageUpdatedNotification): + usage = payload.token_usage + continue + if isinstance(payload, TurnCompletedNotificationPayload): + status = payload.turn.status + error = payload.turn.error + + if printed_delta: + print() + else: + print("[no text]") + + status_text = _status_value(status) + print(f"assistant.status> {status_text}") + if status_text == "failed": + print("assistant.error>", error) + + print(_format_usage(usage)) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/11_cli_mini_app/sync.py b/sdk/python/examples/11_cli_mini_app/sync.py new file mode 100644 index 0000000000..4d597572aa --- /dev/null +++ b/sdk/python/examples/11_cli_mini_app/sync.py @@ -0,0 +1,89 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import ( + Codex, + TextInput, + ThreadTokenUsageUpdatedNotification, + TurnCompletedNotificationPayload, +) + +print("Codex mini CLI. Type /exit to quit.") + + +def _status_value(status: object | None) -> str: + return str(getattr(status, "value", status)) + + +def _format_usage(usage: object | None) -> str: + if usage is None: + return "usage> (none)" + + last = getattr(usage, "last", None) + total = getattr(usage, "total", None) + if last is None or total is None: + return f"usage> {usage}" + + return ( + "usage>\n" + f" last: input={last.inputTokens} output={last.outputTokens} reasoning={last.reasoningOutputTokens} total={last.totalTokens} cached={last.cachedInputTokens}\n" + f" total: input={total.inputTokens} output={total.outputTokens} reasoning={total.reasoningOutputTokens} total={total.totalTokens} cached={total.cachedInputTokens}" + ) + + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + print("Thread:", thread.id) + + while True: + try: + user_input = input("you> ").strip() + except EOFError: + break + + if not user_input: + continue + if user_input in {"/exit", "/quit"}: + break + + turn = thread.turn(TextInput(user_input)) + usage = None + status = None + error = None + printed_delta = False + + print("assistant> ", end="", flush=True) + for event in turn.stream(): + payload = event.payload + if event.method == "item/agentMessage/delta": + delta = getattr(payload, "delta", "") + if delta: + print(delta, end="", flush=True) + printed_delta = True + continue + if isinstance(payload, ThreadTokenUsageUpdatedNotification): + usage = payload.token_usage + continue + if isinstance(payload, TurnCompletedNotificationPayload): + status = payload.turn.status + error = payload.turn.error + + if printed_delta: + print() + else: + print("[no text]") + + status_text = _status_value(status) + print(f"assistant.status> {status_text}") + if status_text == "failed": + print("assistant.error>", error) + + print(_format_usage(usage)) diff --git a/sdk/python/examples/12_turn_params_kitchen_sink/async.py b/sdk/python/examples/12_turn_params_kitchen_sink/async.py new file mode 100644 index 0000000000..6e05c0cbe9 --- /dev/null +++ b/sdk/python/examples/12_turn_params_kitchen_sink/async.py @@ -0,0 +1,75 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import ( + AskForApproval, + AsyncCodex, + Personality, + ReasoningEffort, + ReasoningSummary, + SandboxPolicy, + TextInput, +) + +OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "actions": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["summary", "actions"], + "additionalProperties": False, +} + +SANDBOX_POLICY = SandboxPolicy.model_validate( + { + "type": "readOnly", + "access": {"type": "fullAccess"}, + } +) +SUMMARY = ReasoningSummary.model_validate("concise") + +PROMPT = ( + "Analyze a safe rollout plan for enabling a feature flag in production. " + "Return JSON matching the requested schema." +) +APPROVAL_POLICY = AskForApproval.model_validate("never") + + +async def main() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + turn = await thread.turn( + TextInput(PROMPT), + approval_policy=APPROVAL_POLICY, + cwd=str(Path.cwd()), + effort=ReasoningEffort.medium, + model="gpt-5", + output_schema=OUTPUT_SCHEMA, + personality=Personality.pragmatic, + sandbox_policy=SANDBOX_POLICY, + summary=SUMMARY, + ) + result = await turn.run() + + print("Status:", result.status) + print("Text:", result.text) + print("Usage:", result.usage) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/12_turn_params_kitchen_sink/sync.py b/sdk/python/examples/12_turn_params_kitchen_sink/sync.py new file mode 100644 index 0000000000..9e632a32b5 --- /dev/null +++ b/sdk/python/examples/12_turn_params_kitchen_sink/sync.py @@ -0,0 +1,67 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import ( + AskForApproval, + Codex, + Personality, + ReasoningEffort, + ReasoningSummary, + SandboxPolicy, + TextInput, +) + +OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "actions": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["summary", "actions"], + "additionalProperties": False, +} + +SANDBOX_POLICY = SandboxPolicy.model_validate( + { + "type": "readOnly", + "access": {"type": "fullAccess"}, + } +) +SUMMARY = ReasoningSummary.model_validate("concise") + +PROMPT = ( + "Analyze a safe rollout plan for enabling a feature flag in production. " + "Return JSON matching the requested schema." +) +APPROVAL_POLICY = AskForApproval.model_validate("never") + +with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + + turn = thread.turn( + TextInput(PROMPT), + approval_policy=APPROVAL_POLICY, + cwd=str(Path.cwd()), + effort=ReasoningEffort.medium, + model="gpt-5", + output_schema=OUTPUT_SCHEMA, + personality=Personality.pragmatic, + sandbox_policy=SANDBOX_POLICY, + summary=SUMMARY, + ) + result = turn.run() + + print("Status:", result.status) + print("Text:", result.text) + print("Usage:", result.usage) diff --git a/sdk/python/examples/13_model_select_and_turn_params/async.py b/sdk/python/examples/13_model_select_and_turn_params/async.py new file mode 100644 index 0000000000..125f5f6015 --- /dev/null +++ b/sdk/python/examples/13_model_select_and_turn_params/async.py @@ -0,0 +1,117 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +import asyncio + +from codex_app_server import ( + AskForApproval, + AsyncCodex, + Personality, + ReasoningEffort, + ReasoningSummary, + SandboxPolicy, + TextInput, +) + +REASONING_RANK = { + "none": 0, + "minimal": 1, + "low": 2, + "medium": 3, + "high": 4, + "xhigh": 5, +} + + +def _pick_highest_model(models): + visible = [m for m in models if not m.hidden] or models + known_names = {m.id for m in visible} | {m.model for m in visible} + top_candidates = [m for m in visible if not (m.upgrade and m.upgrade in known_names)] + pool = top_candidates or visible + return max(pool, key=lambda m: (m.model, m.id)) + + +def _pick_highest_turn_effort(model) -> ReasoningEffort: + if not model.supported_reasoning_efforts: + return ReasoningEffort.medium + + best = max( + model.supported_reasoning_efforts, + key=lambda option: REASONING_RANK.get(option.reasoning_effort.value, -1), + ) + return ReasoningEffort(best.reasoning_effort.value) + + +OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "actions": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["summary", "actions"], + "additionalProperties": False, +} + +SANDBOX_POLICY = SandboxPolicy.model_validate( + { + "type": "readOnly", + "access": {"type": "fullAccess"}, + } +) +APPROVAL_POLICY = AskForApproval.model_validate("never") + + +async def main() -> None: + async with AsyncCodex() as codex: + models = await codex.models(include_hidden=True) + selected_model = _pick_highest_model(models.data) + selected_effort = _pick_highest_turn_effort(selected_model) + + print("selected.model:", selected_model.model) + print("selected.effort:", selected_effort.value) + + thread = await codex.thread_start( + model=selected_model.model, + config={"model_reasoning_effort": selected_effort.value}, + ) + + first_turn = await thread.turn( + TextInput("Give one short sentence about reliable production releases."), + model=selected_model.model, + effort=selected_effort, + ) + first = await first_turn.run() + + print("agent.message:", first.text) + print("usage:", first.usage) + + second_turn = await thread.turn( + TextInput("Return JSON for a safe feature-flag rollout plan."), + approval_policy=APPROVAL_POLICY, + cwd=str(Path.cwd()), + effort=selected_effort, + model=selected_model.model, + output_schema=OUTPUT_SCHEMA, + personality=Personality.pragmatic, + sandbox_policy=SANDBOX_POLICY, + summary=ReasoningSummary.model_validate("concise"), + ) + second = await second_turn.run() + + print("agent.message.params:", second.text) + print("usage.params:", second.usage) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/sdk/python/examples/13_model_select_and_turn_params/sync.py b/sdk/python/examples/13_model_select_and_turn_params/sync.py new file mode 100644 index 0000000000..68ef15ddbd --- /dev/null +++ b/sdk/python/examples/13_model_select_and_turn_params/sync.py @@ -0,0 +1,108 @@ +import sys +from pathlib import Path + +_EXAMPLES_ROOT = Path(__file__).resolve().parents[1] +if str(_EXAMPLES_ROOT) not in sys.path: + sys.path.insert(0, str(_EXAMPLES_ROOT)) + +from _bootstrap import ensure_local_sdk_src + +ensure_local_sdk_src() + +from codex_app_server import ( + AskForApproval, + Codex, + Personality, + ReasoningEffort, + ReasoningSummary, + SandboxPolicy, + TextInput, +) + +REASONING_RANK = { + "none": 0, + "minimal": 1, + "low": 2, + "medium": 3, + "high": 4, + "xhigh": 5, +} + + +def _pick_highest_model(models): + visible = [m for m in models if not m.hidden] or models + known_names = {m.id for m in visible} | {m.model for m in visible} + top_candidates = [m for m in visible if not (m.upgrade and m.upgrade in known_names)] + pool = top_candidates or visible + return max(pool, key=lambda m: (m.model, m.id)) + + +def _pick_highest_turn_effort(model) -> ReasoningEffort: + if not model.supported_reasoning_efforts: + return ReasoningEffort.medium + + best = max( + model.supported_reasoning_efforts, + key=lambda option: REASONING_RANK.get(option.reasoning_effort.value, -1), + ) + return ReasoningEffort(best.reasoning_effort.value) + + +OUTPUT_SCHEMA = { + "type": "object", + "properties": { + "summary": {"type": "string"}, + "actions": { + "type": "array", + "items": {"type": "string"}, + }, + }, + "required": ["summary", "actions"], + "additionalProperties": False, +} + +SANDBOX_POLICY = SandboxPolicy.model_validate( + { + "type": "readOnly", + "access": {"type": "fullAccess"}, + } +) +APPROVAL_POLICY = AskForApproval.model_validate("never") + + +with Codex() as codex: + models = codex.models(include_hidden=True) + selected_model = _pick_highest_model(models.data) + selected_effort = _pick_highest_turn_effort(selected_model) + + print("selected.model:", selected_model.model) + print("selected.effort:", selected_effort.value) + + thread = codex.thread_start( + model=selected_model.model, + config={"model_reasoning_effort": selected_effort.value}, + ) + + first = thread.turn( + TextInput("Give one short sentence about reliable production releases."), + model=selected_model.model, + effort=selected_effort, + ).run() + + print("agent.message:", first.text) + print("usage:", first.usage) + + second = thread.turn( + TextInput("Return JSON for a safe feature-flag rollout plan."), + approval_policy=APPROVAL_POLICY, + cwd=str(Path.cwd()), + effort=selected_effort, + model=selected_model.model, + output_schema=OUTPUT_SCHEMA, + personality=Personality.pragmatic, + sandbox_policy=SANDBOX_POLICY, + summary=ReasoningSummary.model_validate("concise"), + ).run() + + print("agent.message.params:", second.text) + print("usage.params:", second.usage) diff --git a/sdk/python/examples/README.md b/sdk/python/examples/README.md new file mode 100644 index 0000000000..7d31faa6a6 --- /dev/null +++ b/sdk/python/examples/README.md @@ -0,0 +1,70 @@ +# Python SDK Examples + +Each example folder contains runnable versions: + +- `sync.py` (public sync surface: `Codex`) +- `async.py` (public async surface: `AsyncCodex`) + +All examples intentionally use only public SDK exports from `codex_app_server`. + +## Prerequisites + +- Python `>=3.10` +- Install SDK dependencies for the same Python interpreter you will use to run examples + +Recommended setup (from `sdk/python`): + +```bash +python -m venv .venv +source .venv/bin/activate +python -m pip install -U pip +python -m pip install -e . +``` + +## Run examples + +From `sdk/python`: + +```bash +python examples//sync.py +python examples//async.py +``` + +The examples bootstrap local imports from `sdk/python/src` automatically, so no `pip install -e .` step is required to run them from this repository checkout. +The only required install step is dependencies for your active interpreter. + +## Recommended first run + +```bash +python examples/01_quickstart_constructor/sync.py +python examples/01_quickstart_constructor/async.py +``` + +## Index + +- `01_quickstart_constructor/` + - first run / sanity check +- `02_turn_run/` + - inspect full turn output fields +- `03_turn_stream_events/` + - stream and print raw notifications +- `04_models_and_metadata/` + - read server metadata and model list +- `05_existing_thread/` + - resume a real existing thread (created in-script) +- `06_thread_lifecycle_and_controls/` + - thread lifecycle + control calls +- `07_image_and_text/` + - remote image URL + text multimodal turn +- `08_local_image_and_text/` + - local image + text multimodal turn using bundled sample image +- `09_async_parity/` + - parity-style sync flow (see async parity in other examples) +- `10_error_handling_and_retry/` + - overload retry pattern + typed error handling structure +- `11_cli_mini_app/` + - interactive chat loop +- `12_turn_params_kitchen_sink/` + - one turn using most optional `turn(...)` params (sync + async) +- `13_model_select_and_turn_params/` + - list models, pick highest model + highest supported reasoning effort, run turns, print message and usage diff --git a/sdk/python/examples/_bootstrap.py b/sdk/python/examples/_bootstrap.py new file mode 100644 index 0000000000..b4f9937f7d --- /dev/null +++ b/sdk/python/examples/_bootstrap.py @@ -0,0 +1,35 @@ +from __future__ import annotations + +import importlib.util +import sys +from pathlib import Path + + +def _ensure_runtime_dependencies(sdk_python_dir: Path) -> None: + if importlib.util.find_spec("pydantic") is not None: + return + + python = sys.executable + raise RuntimeError( + "Missing required dependency: pydantic.\n" + f"Interpreter: {python}\n" + "Install dependencies with the same interpreter used to run this example:\n" + f" {python} -m pip install -e {sdk_python_dir}\n" + "If you installed with `pip` from another Python, reinstall using the command above." + ) + + +def ensure_local_sdk_src() -> Path: + """Add sdk/python/src to sys.path so examples run without installing the package.""" + sdk_python_dir = Path(__file__).resolve().parents[1] + src_dir = sdk_python_dir / "src" + package_dir = src_dir / "codex_app_server" + if not package_dir.exists(): + raise RuntimeError(f"Could not locate local SDK package at {package_dir}") + + _ensure_runtime_dependencies(sdk_python_dir) + + src_str = str(src_dir) + if src_str not in sys.path: + sys.path.insert(0, src_str) + return src_dir diff --git a/sdk/python/examples/assets/sample_scene.png b/sdk/python/examples/assets/sample_scene.png new file mode 100644 index 0000000000..e4efdd429d Binary files /dev/null and b/sdk/python/examples/assets/sample_scene.png differ diff --git a/sdk/python/notebooks/sdk_walkthrough.ipynb b/sdk/python/notebooks/sdk_walkthrough.ipynb new file mode 100644 index 0000000000..64012a855f --- /dev/null +++ b/sdk/python/notebooks/sdk_walkthrough.ipynb @@ -0,0 +1,535 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# Codex Python SDK Walkthrough\n", + "\n", + "Public SDK surface only (`codex_app_server` root exports)." + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 1: bootstrap local SDK imports (no installation required)\n", + "import os\n", + "import sys\n", + "from pathlib import Path\n", + "\n", + "if sys.version_info < (3, 10):\n", + " raise RuntimeError(\n", + " f'Notebook requires Python 3.10+; current interpreter is {sys.version.split()[0]}.'\n", + " )\n", + "\n", + "try:\n", + " _ = os.getcwd()\n", + "except FileNotFoundError:\n", + " os.chdir(str(Path.home()))\n", + "\n", + "\n", + "def _is_sdk_python_dir(path: Path) -> bool:\n", + " return (path / 'pyproject.toml').exists() and (path / 'src' / 'codex_app_server').exists()\n", + "\n", + "\n", + "def _iter_home_fallback_candidates(home: Path):\n", + " # bounded depth scan under home to support launching notebooks from unrelated cwd values\n", + " patterns = ('sdk/python', '*/sdk/python', '*/*/sdk/python', '*/*/*/sdk/python')\n", + " for pattern in patterns:\n", + " yield from home.glob(pattern)\n", + "\n", + "\n", + "def _find_sdk_python_dir(start: Path) -> Path | None:\n", + " checked = set()\n", + "\n", + " def _consider(candidate: Path) -> Path | None:\n", + " resolved = candidate.resolve()\n", + " if resolved in checked:\n", + " return None\n", + " checked.add(resolved)\n", + " if _is_sdk_python_dir(resolved):\n", + " return resolved\n", + " return None\n", + "\n", + " for candidate in [start, *start.parents]:\n", + " found = _consider(candidate)\n", + " if found is not None:\n", + " return found\n", + "\n", + " for candidate in [start / 'sdk' / 'python', *(parent / 'sdk' / 'python' for parent in start.parents)]:\n", + " found = _consider(candidate)\n", + " if found is not None:\n", + " return found\n", + "\n", + " env_dir = os.environ.get('CODEX_PYTHON_SDK_DIR')\n", + " if env_dir:\n", + " found = _consider(Path(env_dir).expanduser())\n", + " if found is not None:\n", + " return found\n", + "\n", + " for entry in sys.path:\n", + " if not entry:\n", + " continue\n", + " entry_path = Path(entry).expanduser()\n", + " for candidate in (entry_path, entry_path / 'sdk' / 'python'):\n", + " found = _consider(candidate)\n", + " if found is not None:\n", + " return found\n", + "\n", + " home = Path.home()\n", + " for candidate in _iter_home_fallback_candidates(home):\n", + " found = _consider(candidate)\n", + " if found is not None:\n", + " return found\n", + "\n", + " return None\n", + "\n", + "\n", + "repo_python_dir = _find_sdk_python_dir(Path.cwd())\n", + "if repo_python_dir is None:\n", + " raise RuntimeError('Could not locate sdk/python. Set CODEX_PYTHON_SDK_DIR to your sdk/python path.')\n", + "\n", + "src_dir = repo_python_dir / 'src'\n", + "if str(src_dir) not in sys.path:\n", + " sys.path.insert(0, str(src_dir))\n", + "\n", + "# Force fresh imports after SDK upgrades in the same notebook kernel.\n", + "for module_name in list(sys.modules):\n", + " if module_name == 'codex_app_server' or module_name.startswith('codex_app_server.'):\n", + " sys.modules.pop(module_name, None)\n", + "\n", + "print('Kernel:', sys.executable)\n", + "print('SDK source:', src_dir)\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 2: imports (public only)\n", + "from codex_app_server import (\n", + " AsyncCodex,\n", + " Codex,\n", + " ImageInput,\n", + " LocalImageInput,\n", + " TextInput,\n", + " retry_on_overload,\n", + ")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 3: simple sync conversation\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " turn = thread.turn(TextInput('Explain gradient descent in 3 bullets.'))\n", + " result = turn.run()\n", + "\n", + " print('server:', codex.metadata)\n", + " print('status:', result.status)\n", + " print(result.text)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 4: multi-turn continuity in same thread\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + "\n", + " first = thread.turn(TextInput('Give a short summary of transformers.')).run()\n", + " second = thread.turn(TextInput('Now explain that to a high-school student.')).run()\n", + "\n", + " print('first status:', first.status)\n", + " print('second status:', second.status)\n", + " print('second text:', second.text)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 5: full thread lifecycle and branching (sync)\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " first = thread.turn(TextInput('One sentence about structured planning.')).run()\n", + " second = thread.turn(TextInput('Now restate it for a junior engineer.')).run()\n", + "\n", + " reopened = codex.thread_resume(thread.id)\n", + " listing_active = codex.thread_list(limit=20, archived=False)\n", + " reading = reopened.read(include_turns=True)\n", + "\n", + " _ = reopened.set_name('sdk-lifecycle-demo')\n", + " _ = codex.thread_archive(reopened.id)\n", + " listing_archived = codex.thread_list(limit=20, archived=True)\n", + " unarchived = codex.thread_unarchive(reopened.id)\n", + "\n", + " resumed_info = 'n/a'\n", + " try:\n", + " resumed = codex.thread_resume(\n", + " unarchived.id,\n", + " model='gpt-5',\n", + " config={'model_reasoning_effort': 'high'},\n", + " )\n", + " resumed_result = resumed.turn(TextInput('Continue in one short sentence.')).run()\n", + " resumed_info = f'{resumed_result.turn_id} {resumed_result.status}'\n", + " except Exception as e:\n", + " resumed_info = f'skipped({type(e).__name__})'\n", + "\n", + " forked_info = 'n/a'\n", + " try:\n", + " forked = codex.thread_fork(unarchived.id, model='gpt-5')\n", + " forked_result = forked.turn(TextInput('Take a different angle in one short sentence.')).run()\n", + " forked_info = f'{forked_result.turn_id} {forked_result.status}'\n", + " except Exception as e:\n", + " forked_info = f'skipped({type(e).__name__})'\n", + "\n", + " compact_info = 'sent'\n", + " try:\n", + " _ = unarchived.compact()\n", + " except Exception as e:\n", + " compact_info = f'skipped({type(e).__name__})'\n", + "\n", + " print('Lifecycle OK:', thread.id)\n", + " print('first:', first.turn_id, first.status)\n", + " print('second:', second.turn_id, second.status)\n", + " print('read.turns:', len(reading.thread.turns or []))\n", + " print('list.active:', len(listing_active.data))\n", + " print('list.archived:', len(listing_archived.data))\n", + " print('resumed:', resumed_info)\n", + " print('forked:', forked_info)\n", + " print('compact:', compact_info)\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 5b: one turn with most optional turn params\n", + "from pathlib import Path\n", + "from codex_app_server import (\n", + " AskForApproval,\n", + " Personality,\n", + " ReasoningEffort,\n", + " ReasoningSummary,\n", + " SandboxPolicy,\n", + ")\n", + "\n", + "output_schema = {\n", + " 'type': 'object',\n", + " 'properties': {\n", + " 'summary': {'type': 'string'},\n", + " 'actions': {'type': 'array', 'items': {'type': 'string'}},\n", + " },\n", + " 'required': ['summary', 'actions'],\n", + " 'additionalProperties': False,\n", + "}\n", + "\n", + "sandbox_policy = SandboxPolicy.model_validate({'type': 'readOnly', 'access': {'type': 'fullAccess'}})\n", + "summary = ReasoningSummary.model_validate('concise')\n", + "\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " turn = thread.turn(\n", + " TextInput('Propose a safe production feature-flag rollout. Return JSON matching the schema.'),\n", + " approval_policy=AskForApproval.never,\n", + " cwd=str(Path.cwd()),\n", + " effort=ReasoningEffort.medium,\n", + " model='gpt-5',\n", + " output_schema=output_schema,\n", + " personality=Personality.pragmatic,\n", + " sandbox_policy=sandbox_policy,\n", + " summary=summary,\n", + " )\n", + " result = turn.run()\n", + "\n", + " print('status:', result.status)\n", + " print(result.text)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 5c: choose highest model + highest supported reasoning, then run turns\n", + "from pathlib import Path\n", + "from codex_app_server import (\n", + " AskForApproval,\n", + " Personality,\n", + " ReasoningEffort,\n", + " ReasoningSummary,\n", + " SandboxPolicy,\n", + ")\n", + "\n", + "reasoning_rank = {\n", + " 'none': 0,\n", + " 'minimal': 1,\n", + " 'low': 2,\n", + " 'medium': 3,\n", + " 'high': 4,\n", + " 'xhigh': 5,\n", + "}\n", + "\n", + "\n", + "def pick_highest_model(models):\n", + " visible = [m for m in models if not m.hidden] or models\n", + " known_names = {m.id for m in visible} | {m.model for m in visible}\n", + " top_candidates = [m for m in visible if not (m.upgrade and m.upgrade in known_names)]\n", + " pool = top_candidates or visible\n", + " return max(pool, key=lambda m: (m.model, m.id))\n", + "\n", + "\n", + "def pick_highest_turn_effort(model) -> ReasoningEffort:\n", + " if not model.supported_reasoning_efforts:\n", + " return ReasoningEffort.medium\n", + " best = max(model.supported_reasoning_efforts, key=lambda opt: reasoning_rank.get(opt.reasoning_effort.value, -1))\n", + " return ReasoningEffort(best.reasoning_effort.value)\n", + "\n", + "\n", + "output_schema = {\n", + " 'type': 'object',\n", + " 'properties': {\n", + " 'summary': {'type': 'string'},\n", + " 'actions': {'type': 'array', 'items': {'type': 'string'}},\n", + " },\n", + " 'required': ['summary', 'actions'],\n", + " 'additionalProperties': False,\n", + "}\n", + "sandbox_policy = SandboxPolicy.model_validate({'type': 'readOnly', 'access': {'type': 'fullAccess'}})\n", + "\n", + "with Codex() as codex:\n", + " models = codex.models(include_hidden=True)\n", + " selected_model = pick_highest_model(models.data)\n", + " selected_effort = pick_highest_turn_effort(selected_model)\n", + "\n", + " print('selected.model:', selected_model.model)\n", + " print('selected.effort:', selected_effort.value)\n", + "\n", + " thread = codex.thread_start(model=selected_model.model, config={'model_reasoning_effort': selected_effort.value})\n", + "\n", + " first = thread.turn(\n", + " TextInput('Give one short sentence about reliable production releases.'),\n", + " model=selected_model.model,\n", + " effort=selected_effort,\n", + " ).run()\n", + " print('agent.message:', first.text)\n", + " print('usage:', first.usage)\n", + "\n", + " second = thread.turn(\n", + " TextInput('Return JSON for a safe feature-flag rollout plan.'),\n", + " approval_policy=AskForApproval.never,\n", + " cwd=str(Path.cwd()),\n", + " effort=selected_effort,\n", + " model=selected_model.model,\n", + " output_schema=output_schema,\n", + " personality=Personality.pragmatic,\n", + " sandbox_policy=sandbox_policy,\n", + " summary=ReasoningSummary.model_validate('concise'),\n", + " ).run()\n", + " print('agent.message.params:', second.text)\n", + " print('usage.params:', second.usage)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 6: multimodal with remote image\n", + "remote_image_url = 'https://raw.githubusercontent.com/github/explore/main/topics/python/python.png'\n", + "\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " result = thread.turn([\n", + " TextInput('What do you see in this image? 3 bullets.'),\n", + " ImageInput(remote_image_url),\n", + " ]).run()\n", + "\n", + " print('status:', result.status)\n", + " print(result.text)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 7: multimodal with local image (bundled asset)\n", + "local_image_path = repo_python_dir / 'examples' / 'assets' / 'sample_scene.png'\n", + "if not local_image_path.exists():\n", + " raise FileNotFoundError(f'Missing bundled image: {local_image_path}')\n", + "\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " result = thread.turn([\n", + " TextInput('Describe this local image in 2 bullets.'),\n", + " LocalImageInput(str(local_image_path.resolve())),\n", + " ]).run()\n", + "\n", + " print('status:', result.status)\n", + " print(result.text)\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 8: retry-on-overload pattern\n", + "with Codex() as codex:\n", + " thread = codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + "\n", + " result = retry_on_overload(\n", + " lambda: thread.turn(TextInput('List 5 failure modes in distributed systems.')).run(),\n", + " max_attempts=3,\n", + " initial_delay_s=0.25,\n", + " max_delay_s=2.0,\n", + " )\n", + "\n", + " print('status:', result.status)\n", + " print(result.text)" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 9: full thread lifecycle and branching (async)\n", + "import asyncio\n", + "\n", + "\n", + "async def async_lifecycle_demo():\n", + " async with AsyncCodex() as codex:\n", + " thread = await codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " first = await (await thread.turn(TextInput('One sentence about structured planning.'))).run()\n", + " second = await (await thread.turn(TextInput('Now restate it for a junior engineer.'))).run()\n", + "\n", + " reopened = await codex.thread_resume(thread.id)\n", + " listing_active = await codex.thread_list(limit=20, archived=False)\n", + " reading = await reopened.read(include_turns=True)\n", + "\n", + " _ = await reopened.set_name('sdk-lifecycle-demo')\n", + " _ = await codex.thread_archive(reopened.id)\n", + " listing_archived = await codex.thread_list(limit=20, archived=True)\n", + " unarchived = await codex.thread_unarchive(reopened.id)\n", + "\n", + " resumed_info = 'n/a'\n", + " try:\n", + " resumed = await codex.thread_resume(\n", + " unarchived.id,\n", + " model='gpt-5',\n", + " config={'model_reasoning_effort': 'high'},\n", + " )\n", + " resumed_result = await (await resumed.turn(TextInput('Continue in one short sentence.'))).run()\n", + " resumed_info = f'{resumed_result.turn_id} {resumed_result.status}'\n", + " except Exception as e:\n", + " resumed_info = f'skipped({type(e).__name__})'\n", + "\n", + " forked_info = 'n/a'\n", + " try:\n", + " forked = await codex.thread_fork(unarchived.id, model='gpt-5')\n", + " forked_result = await (await forked.turn(TextInput('Take a different angle in one short sentence.'))).run()\n", + " forked_info = f'{forked_result.turn_id} {forked_result.status}'\n", + " except Exception as e:\n", + " forked_info = f'skipped({type(e).__name__})'\n", + "\n", + " compact_info = 'sent'\n", + " try:\n", + " _ = await unarchived.compact()\n", + " except Exception as e:\n", + " compact_info = f'skipped({type(e).__name__})'\n", + "\n", + " print('Lifecycle OK:', thread.id)\n", + " print('first:', first.turn_id, first.status)\n", + " print('second:', second.turn_id, second.status)\n", + " print('read.turns:', len(reading.thread.turns or []))\n", + " print('list.active:', len(listing_active.data))\n", + " print('list.archived:', len(listing_archived.data))\n", + " print('resumed:', resumed_info)\n", + " print('forked:', forked_info)\n", + " print('compact:', compact_info)\n", + "\n", + "\n", + "await async_lifecycle_demo()\n", + "\n" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "# Cell 10: async stream + steer + interrupt (best effort)\n", + "import asyncio\n", + "\n", + "\n", + "async def async_stream_demo():\n", + " async with AsyncCodex() as codex:\n", + " thread = await codex.thread_start(model='gpt-5', config={'model_reasoning_effort': 'high'})\n", + " turn = await thread.turn(TextInput('Count from 1 to 200 with commas, then one summary sentence.'))\n", + "\n", + " try:\n", + " _ = await turn.steer(TextInput('Keep it brief and stop after 20 numbers.'))\n", + " print('steer: sent')\n", + " except Exception as e:\n", + " print('steer: skipped', type(e).__name__)\n", + "\n", + " try:\n", + " _ = await turn.interrupt()\n", + " print('interrupt: sent')\n", + " except Exception as e:\n", + " print('interrupt: skipped', type(e).__name__)\n", + "\n", + " event_count = 0\n", + " async for event in turn.stream():\n", + " event_count += 1\n", + " print(event.method, event.payload)\n", + "\n", + " print('events.count:', event_count)\n", + "\n", + "\n", + "await async_stream_demo()\n", + "\n" + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "name": "python", + "version": "3.10+" + } + }, + "nbformat": 4, + "nbformat_minor": 5 +} diff --git a/sdk/python/src/codex_app_server/__init__.py b/sdk/python/src/codex_app_server/__init__.py index aff63176b9..7686ffa794 100644 --- a/sdk/python/src/codex_app_server/__init__.py +++ b/sdk/python/src/codex_app_server/__init__.py @@ -1,10 +1,110 @@ -from .client import AppServerClient, AppServerConfig -from .errors import AppServerError, JsonRpcError, TransportClosedError +from .client import AppServerConfig +from .errors import ( + AppServerError, + AppServerRpcError, + InvalidParamsError, + InvalidRequestError, + InternalRpcError, + JsonRpcError, + MethodNotFoundError, + ParseError, + RetryLimitExceededError, + ServerBusyError, + TransportClosedError, + is_retryable_error, +) +from .generated.v2_types import ( + ThreadItem, + ThreadTokenUsageUpdatedNotification, + TurnCompletedNotificationPayload, +) +from .public_api import ( + AsyncCodex, + AsyncThread, + AsyncTurn, + Codex, + ImageInput, + InitializeResult, + Input, + InputItem, + LocalImageInput, + MentionInput, + SkillInput, + TextInput, + Thread, + Turn, + TurnResult, +) +from .public_types import ( + AskForApproval, + Personality, + PlanType, + ReasoningEffort, + ReasoningSummary, + SandboxMode, + SandboxPolicy, + ThreadForkParams, + ThreadListParams, + ThreadResumeParams, + ThreadSortKey, + ThreadSourceKind, + ThreadStartParams, + TurnStartParams, + TurnStatus, + TurnSteerParams, +) +from .retry import retry_on_overload + +__version__ = "0.2.0" __all__ = [ - "AppServerClient", + "__version__", "AppServerConfig", + "Codex", + "AsyncCodex", + "Thread", + "AsyncThread", + "Turn", + "AsyncTurn", + "TurnResult", + "InitializeResult", + "Input", + "InputItem", + "TextInput", + "ImageInput", + "LocalImageInput", + "SkillInput", + "MentionInput", + "ThreadItem", + "ThreadTokenUsageUpdatedNotification", + "TurnCompletedNotificationPayload", + "AskForApproval", + "Personality", + "PlanType", + "ReasoningEffort", + "ReasoningSummary", + "SandboxMode", + "SandboxPolicy", + "ThreadStartParams", + "ThreadResumeParams", + "ThreadListParams", + "ThreadSortKey", + "ThreadSourceKind", + "ThreadForkParams", + "TurnStatus", + "TurnStartParams", + "TurnSteerParams", + "retry_on_overload", "AppServerError", - "JsonRpcError", "TransportClosedError", + "JsonRpcError", + "AppServerRpcError", + "ParseError", + "InvalidRequestError", + "MethodNotFoundError", + "InvalidParamsError", + "InternalRpcError", + "ServerBusyError", + "RetryLimitExceededError", + "is_retryable_error", ] diff --git a/sdk/python/src/codex_app_server/async_client.py b/sdk/python/src/codex_app_server/async_client.py new file mode 100644 index 0000000000..6ca0c42a78 --- /dev/null +++ b/sdk/python/src/codex_app_server/async_client.py @@ -0,0 +1,208 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Iterator +from typing import AsyncIterator, Callable, Iterable, ParamSpec, TypeVar + +from pydantic import BaseModel + +from .client import AppServerClient, AppServerConfig +from .generated.v2_all import ( + AgentMessageDeltaNotification, + ModelListResponse, + ThreadArchiveResponse, + ThreadCompactStartResponse, + ThreadForkParams as V2ThreadForkParams, + ThreadForkResponse, + ThreadListParams as V2ThreadListParams, + ThreadListResponse, + ThreadReadResponse, + ThreadResumeParams as V2ThreadResumeParams, + ThreadResumeResponse, + ThreadSetNameResponse, + ThreadStartParams as V2ThreadStartParams, + ThreadStartResponse, + ThreadUnarchiveResponse, + TurnCompletedNotification, + TurnInterruptResponse, + TurnStartParams as V2TurnStartParams, + TurnStartResponse, + TurnSteerResponse, +) +from .models import InitializeResponse, JsonObject, Notification + +ModelT = TypeVar("ModelT", bound=BaseModel) +ParamsT = ParamSpec("ParamsT") +ReturnT = TypeVar("ReturnT") + + +class AsyncAppServerClient: + """Async wrapper around AppServerClient using thread offloading.""" + + def __init__(self, config: AppServerConfig | None = None) -> None: + self._sync = AppServerClient(config=config) + # Single stdio transport cannot be read safely from multiple threads. + self._transport_lock = asyncio.Lock() + + async def __aenter__(self) -> "AsyncAppServerClient": + await self.start() + return self + + async def __aexit__(self, _exc_type, _exc, _tb) -> None: + await self.close() + + async def _call_sync( + self, + fn: Callable[ParamsT, ReturnT], + /, + *args: ParamsT.args, + **kwargs: ParamsT.kwargs, + ) -> ReturnT: + async with self._transport_lock: + return await asyncio.to_thread(fn, *args, **kwargs) + + @staticmethod + def _next_from_iterator( + iterator: Iterator[AgentMessageDeltaNotification], + ) -> tuple[bool, AgentMessageDeltaNotification | None]: + try: + return True, next(iterator) + except StopIteration: + return False, None + + async def start(self) -> None: + await self._call_sync(self._sync.start) + + async def close(self) -> None: + await self._call_sync(self._sync.close) + + async def initialize(self) -> InitializeResponse: + return await self._call_sync(self._sync.initialize) + + def acquire_turn_consumer(self, turn_id: str) -> None: + self._sync.acquire_turn_consumer(turn_id) + + def release_turn_consumer(self, turn_id: str) -> None: + self._sync.release_turn_consumer(turn_id) + + async def request( + self, + method: str, + params: JsonObject | None, + *, + response_model: type[ModelT], + ) -> ModelT: + return await self._call_sync( + self._sync.request, + method, + params, + response_model=response_model, + ) + + async def thread_start(self, params: V2ThreadStartParams | JsonObject | None = None) -> ThreadStartResponse: + return await self._call_sync(self._sync.thread_start, params) + + async def thread_resume( + self, + thread_id: str, + params: V2ThreadResumeParams | JsonObject | None = None, + ) -> ThreadResumeResponse: + return await self._call_sync(self._sync.thread_resume, thread_id, params) + + async def thread_list(self, params: V2ThreadListParams | JsonObject | None = None) -> ThreadListResponse: + return await self._call_sync(self._sync.thread_list, params) + + async def thread_read(self, thread_id: str, include_turns: bool = False) -> ThreadReadResponse: + return await self._call_sync(self._sync.thread_read, thread_id, include_turns) + + async def thread_fork( + self, + thread_id: str, + params: V2ThreadForkParams | JsonObject | None = None, + ) -> ThreadForkResponse: + return await self._call_sync(self._sync.thread_fork, thread_id, params) + + async def thread_archive(self, thread_id: str) -> ThreadArchiveResponse: + return await self._call_sync(self._sync.thread_archive, thread_id) + + async def thread_unarchive(self, thread_id: str) -> ThreadUnarchiveResponse: + return await self._call_sync(self._sync.thread_unarchive, thread_id) + + async def thread_set_name(self, thread_id: str, name: str) -> ThreadSetNameResponse: + return await self._call_sync(self._sync.thread_set_name, thread_id, name) + + async def thread_compact(self, thread_id: str) -> ThreadCompactStartResponse: + return await self._call_sync(self._sync.thread_compact, thread_id) + + async def turn_start( + self, + thread_id: str, + input_items: list[JsonObject] | JsonObject | str, + params: V2TurnStartParams | JsonObject | None = None, + ) -> TurnStartResponse: + return await self._call_sync(self._sync.turn_start, thread_id, input_items, params) + + async def turn_interrupt(self, thread_id: str, turn_id: str) -> TurnInterruptResponse: + return await self._call_sync(self._sync.turn_interrupt, thread_id, turn_id) + + async def turn_steer( + self, + thread_id: str, + expected_turn_id: str, + input_items: list[JsonObject] | JsonObject | str, + ) -> TurnSteerResponse: + return await self._call_sync( + self._sync.turn_steer, + thread_id, + expected_turn_id, + input_items, + ) + + async def model_list(self, include_hidden: bool = False) -> ModelListResponse: + return await self._call_sync(self._sync.model_list, include_hidden) + + async def request_with_retry_on_overload( + self, + method: str, + params: JsonObject | None, + *, + response_model: type[ModelT], + max_attempts: int = 3, + initial_delay_s: float = 0.25, + max_delay_s: float = 2.0, + ) -> ModelT: + return await self._call_sync( + self._sync.request_with_retry_on_overload, + method, + params, + response_model=response_model, + max_attempts=max_attempts, + initial_delay_s=initial_delay_s, + max_delay_s=max_delay_s, + ) + + async def next_notification(self) -> Notification: + return await self._call_sync(self._sync.next_notification) + + async def wait_for_turn_completed(self, turn_id: str) -> TurnCompletedNotification: + return await self._call_sync(self._sync.wait_for_turn_completed, turn_id) + + async def stream_until_methods(self, methods: Iterable[str] | str) -> list[Notification]: + return await self._call_sync(self._sync.stream_until_methods, methods) + + async def stream_text( + self, + thread_id: str, + text: str, + params: V2TurnStartParams | JsonObject | None = None, + ) -> AsyncIterator[AgentMessageDeltaNotification]: + async with self._transport_lock: + iterator = self._sync.stream_text(thread_id, text, params) + while True: + has_value, chunk = await asyncio.to_thread( + self._next_from_iterator, + iterator, + ) + if not has_value: + break + yield chunk diff --git a/sdk/python/src/codex_app_server/bin/darwin-arm64/codex b/sdk/python/src/codex_app_server/bin/darwin-arm64/codex index 70c50814c6..2d2894b475 100755 Binary files a/sdk/python/src/codex_app_server/bin/darwin-arm64/codex and b/sdk/python/src/codex_app_server/bin/darwin-arm64/codex differ diff --git a/sdk/python/src/codex_app_server/generated/v2_types.py b/sdk/python/src/codex_app_server/generated/v2_types.py index 932ab438db..ef28f982e5 100644 --- a/sdk/python/src/codex_app_server/generated/v2_types.py +++ b/sdk/python/src/codex_app_server/generated/v2_types.py @@ -1,25 +1,23 @@ -"""Stable aliases over full v2 autogenerated models (datamodel-code-generator).""" +"""Stable aliases over the canonical generated v2 models.""" -from .v2_all.ModelListResponse import ModelListResponse -from .v2_all.ThreadCompactStartResponse import ThreadCompactStartResponse -from .v2_all.ThreadListResponse import ThreadListResponse -from .v2_all.ThreadReadResponse import ThreadReadResponse -from .v2_all.ThreadTokenUsageUpdatedNotification import ( +from .v2_all import ( + ModelListResponse, + ThreadCompactStartResponse, + ThreadItem, + ThreadListResponse, + ThreadReadResponse, ThreadTokenUsageUpdatedNotification, -) -from .v2_all.TurnCompletedNotification import ThreadItem153 as ThreadItem -from .v2_all.TurnCompletedNotification import ( TurnCompletedNotification as TurnCompletedNotificationPayload, + TurnSteerResponse, ) -from .v2_all.TurnSteerResponse import TurnSteerResponse __all__ = [ "ModelListResponse", "ThreadCompactStartResponse", + "ThreadItem", "ThreadListResponse", "ThreadReadResponse", "ThreadTokenUsageUpdatedNotification", "TurnCompletedNotificationPayload", "TurnSteerResponse", - "ThreadItem", ] diff --git a/sdk/python/src/codex_app_server/public_api.py b/sdk/python/src/codex_app_server/public_api.py new file mode 100644 index 0000000000..f337a0e0de --- /dev/null +++ b/sdk/python/src/codex_app_server/public_api.py @@ -0,0 +1,790 @@ +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from typing import AsyncIterator, Iterator + +from .async_client import AsyncAppServerClient +from .client import AppServerClient, AppServerConfig +from .generated.v2_all import ( + AgentMessageDeltaNotification, + RawResponseItemCompletedNotification, + ThreadArchiveResponse, + ThreadSetNameResponse, + TurnError, + TurnInterruptResponse, +) +from .generated.v2_types import ( + ModelListResponse, + ThreadCompactStartResponse, + ThreadItem, + ThreadListResponse, + ThreadReadResponse, + ThreadTokenUsageUpdatedNotification, + TurnCompletedNotificationPayload, + TurnSteerResponse, +) +from .models import InitializeResponse, JsonObject, Notification +from .public_types import ( + AskForApproval, + Personality, + ReasoningEffort, + ReasoningSummary, + SandboxMode, + SandboxPolicy, + ThreadForkParams, + ThreadListParams, + ThreadResumeParams, + ThreadSortKey, + ThreadSourceKind, + ThreadStartParams, + TurnStartParams, + TurnStatus, +) + + +@dataclass(slots=True) +class TurnResult: + thread_id: str + turn_id: str + status: TurnStatus + error: TurnError | None + text: str + items: list[ThreadItem] + usage: ThreadTokenUsageUpdatedNotification | None = None + + +@dataclass(slots=True) +class TextInput: + text: str + + +@dataclass(slots=True) +class ImageInput: + url: str + + +@dataclass(slots=True) +class LocalImageInput: + path: str + + +@dataclass(slots=True) +class SkillInput: + name: str + path: str + + +@dataclass(slots=True) +class MentionInput: + name: str + path: str + + +InputItem = TextInput | ImageInput | LocalImageInput | SkillInput | MentionInput +Input = list[InputItem] | InputItem + + +@dataclass(slots=True) +class InitializeResult: + server_name: str + server_version: str + user_agent: str + + +def _to_wire_item(item: InputItem) -> JsonObject: + if isinstance(item, TextInput): + return {"type": "text", "text": item.text} + if isinstance(item, ImageInput): + return {"type": "image", "url": item.url} + if isinstance(item, LocalImageInput): + return {"type": "localImage", "path": item.path} + if isinstance(item, SkillInput): + return {"type": "skill", "name": item.name, "path": item.path} + if isinstance(item, MentionInput): + return {"type": "mention", "name": item.name, "path": item.path} + raise TypeError(f"unsupported input item: {type(item)!r}") + + +def _to_wire_input(input: Input) -> list[JsonObject]: + if isinstance(input, list): + return [_to_wire_item(i) for i in input] + return [_to_wire_item(input)] + + +def _split_user_agent(user_agent: str) -> tuple[str | None, str | None]: + raw = user_agent.strip() + if not raw: + return None, None + if "/" in raw: + name, version = raw.split("/", 1) + return (name or None), (version or None) + parts = raw.split(maxsplit=1) + if len(parts) == 2: + return parts[0], parts[1] + return raw, None + + +def _enum_value(value: object) -> object: + return getattr(value, "value", value) + + +def _assistant_output_text_chunks( + notification: RawResponseItemCompletedNotification, +) -> list[str]: + item = notification.item.root + if _enum_value(getattr(item, "type", None)) != "message": + return [] + if getattr(item, "role", None) != "assistant": + return [] + + chunks: list[str] = [] + for content in getattr(item, "content", []) or []: + content_item = getattr(content, "root", content) + if _enum_value(getattr(content_item, "type", None)) != "output_text": + continue + text = getattr(content_item, "text", None) + if isinstance(text, str) and text: + chunks.append(text) + return chunks + + +def _build_turn_result( + completed: TurnCompletedNotificationPayload | None, + usage: ThreadTokenUsageUpdatedNotification | None, + delta_chunks: list[str], + raw_text_chunks: list[str], +) -> TurnResult: + if completed is None: + raise RuntimeError("turn completed event not received") + if completed.turn.status == TurnStatus.completed and usage is None: + raise RuntimeError( + "thread/tokenUsage/updated notification not received for completed turn" + ) + + text = "".join(delta_chunks) if delta_chunks else "".join(raw_text_chunks) + return TurnResult( + thread_id=completed.thread_id, + turn_id=completed.turn.id, + status=completed.turn.status, + error=completed.turn.error, + text=text, + items=list(completed.turn.items or []), + usage=usage, + ) + + +class Codex: + """Minimal typed SDK surface for app-server v2.""" + + def __init__(self, config: AppServerConfig | None = None) -> None: + self._client = AppServerClient(config=config) + try: + self._client.start() + self._init = self._parse_initialize(self._client.initialize()) + except Exception: + self._client.close() + raise + + def __enter__(self) -> "Codex": + return self + + def __exit__(self, _exc_type, _exc, _tb) -> None: + self.close() + + @staticmethod + def _parse_initialize(payload: InitializeResponse) -> InitializeResult: + user_agent = (payload.userAgent or "").strip() + server = payload.serverInfo + + server_name: str | None = None + server_version: str | None = None + + if server is not None: + server_name = (server.name or "").strip() or None + server_version = (server.version or "").strip() or None + + if (server_name is None or server_version is None) and user_agent: + parsed_name, parsed_version = _split_user_agent(user_agent) + if server_name is None: + server_name = parsed_name + if server_version is None: + server_version = parsed_version + + normalized_server_name = (server_name or "").strip() + normalized_server_version = (server_version or "").strip() + if not user_agent or not normalized_server_name or not normalized_server_version: + raise RuntimeError( + "initialize response missing required metadata " + f"(user_agent={user_agent!r}, server_name={normalized_server_name!r}, server_version={normalized_server_version!r})" + ) + + return InitializeResult( + server_name=normalized_server_name, + server_version=normalized_server_version, + user_agent=user_agent, + ) + + @property + def metadata(self) -> InitializeResult: + return self._init + + def close(self) -> None: + self._client.close() + + # BEGIN GENERATED: Codex.flat_methods + def thread_start( + self, + *, + approval_policy: AskForApproval | None = None, + base_instructions: str | None = None, + config: JsonObject | None = None, + cwd: str | None = None, + developer_instructions: str | None = None, + ephemeral: bool | None = None, + model: str | None = None, + model_provider: str | None = None, + personality: Personality | None = None, + sandbox: SandboxMode | None = None, + service_name: str | None = None, + service_tier: ServiceTier | None = None, + ) -> Thread: + params = ThreadStartParams( + approval_policy=approval_policy, + base_instructions=base_instructions, + config=config, + cwd=cwd, + developer_instructions=developer_instructions, + ephemeral=ephemeral, + model=model, + model_provider=model_provider, + personality=personality, + sandbox=sandbox, + service_name=service_name, + service_tier=service_tier, + ) + started = self._client.thread_start(params) + return Thread(self._client, started.thread.id) + + def thread_list( + self, + *, + archived: bool | None = None, + cursor: str | None = None, + cwd: str | None = None, + limit: int | None = None, + model_providers: list[str] | None = None, + search_term: str | None = None, + sort_key: ThreadSortKey | None = None, + source_kinds: list[ThreadSourceKind] | None = None, + ) -> ThreadListResponse: + params = ThreadListParams( + archived=archived, + cursor=cursor, + cwd=cwd, + limit=limit, + model_providers=model_providers, + search_term=search_term, + sort_key=sort_key, + source_kinds=source_kinds, + ) + return self._client.thread_list(params) + + def thread_resume( + self, + thread_id: str, + *, + approval_policy: AskForApproval | None = None, + base_instructions: str | None = None, + config: JsonObject | None = None, + cwd: str | None = None, + developer_instructions: str | None = None, + model: str | None = None, + model_provider: str | None = None, + personality: Personality | None = None, + sandbox: SandboxMode | None = None, + service_tier: ServiceTier | None = None, + ) -> Thread: + params = ThreadResumeParams( + thread_id=thread_id, + approval_policy=approval_policy, + base_instructions=base_instructions, + config=config, + cwd=cwd, + developer_instructions=developer_instructions, + model=model, + model_provider=model_provider, + personality=personality, + sandbox=sandbox, + service_tier=service_tier, + ) + resumed = self._client.thread_resume(thread_id, params) + return Thread(self._client, resumed.thread.id) + + def thread_fork( + self, + thread_id: str, + *, + approval_policy: AskForApproval | None = None, + base_instructions: str | None = None, + config: JsonObject | None = None, + cwd: str | None = None, + developer_instructions: str | None = None, + model: str | None = None, + model_provider: str | None = None, + sandbox: SandboxMode | None = None, + service_tier: ServiceTier | None = None, + ) -> Thread: + params = ThreadForkParams( + thread_id=thread_id, + approval_policy=approval_policy, + base_instructions=base_instructions, + config=config, + cwd=cwd, + developer_instructions=developer_instructions, + model=model, + model_provider=model_provider, + sandbox=sandbox, + service_tier=service_tier, + ) + forked = self._client.thread_fork(thread_id, params) + return Thread(self._client, forked.thread.id) + + def thread_archive(self, thread_id: str) -> ThreadArchiveResponse: + return self._client.thread_archive(thread_id) + + def thread_unarchive(self, thread_id: str) -> Thread: + unarchived = self._client.thread_unarchive(thread_id) + return Thread(self._client, unarchived.thread.id) + # END GENERATED: Codex.flat_methods + + def models(self, *, include_hidden: bool = False) -> ModelListResponse: + return self._client.model_list(include_hidden=include_hidden) + + +class AsyncCodex: + """Async mirror of :class:`Codex` with matching method shapes.""" + + def __init__(self, config: AppServerConfig | None = None) -> None: + self._client = AsyncAppServerClient(config=config) + self._init: InitializeResult | None = None + self._initialized = False + self._init_lock = asyncio.Lock() + + async def __aenter__(self) -> "AsyncCodex": + await self._ensure_initialized() + return self + + async def __aexit__(self, _exc_type, _exc, _tb) -> None: + await self.close() + + async def _ensure_initialized(self) -> None: + if self._initialized: + return + async with self._init_lock: + if self._initialized: + return + try: + await self._client.start() + payload = await self._client.initialize() + self._init = Codex._parse_initialize(payload) + self._initialized = True + except Exception: + await self._client.close() + self._init = None + self._initialized = False + raise + + @property + def metadata(self) -> InitializeResult: + if self._init is None: + raise RuntimeError( + "AsyncCodex is not initialized yet. Use `async with AsyncCodex()` or call an async API first." + ) + return self._init + + async def close(self) -> None: + await self._client.close() + self._init = None + self._initialized = False + + # BEGIN GENERATED: AsyncCodex.flat_methods + async def thread_start( + self, + *, + approval_policy: AskForApproval | None = None, + base_instructions: str | None = None, + config: JsonObject | None = None, + cwd: str | None = None, + developer_instructions: str | None = None, + ephemeral: bool | None = None, + model: str | None = None, + model_provider: str | None = None, + personality: Personality | None = None, + sandbox: SandboxMode | None = None, + service_name: str | None = None, + service_tier: ServiceTier | None = None, + ) -> AsyncThread: + await self._ensure_initialized() + params = ThreadStartParams( + approval_policy=approval_policy, + base_instructions=base_instructions, + config=config, + cwd=cwd, + developer_instructions=developer_instructions, + ephemeral=ephemeral, + model=model, + model_provider=model_provider, + personality=personality, + sandbox=sandbox, + service_name=service_name, + service_tier=service_tier, + ) + started = await self._client.thread_start(params) + return AsyncThread(self, started.thread.id) + + async def thread_list( + self, + *, + archived: bool | None = None, + cursor: str | None = None, + cwd: str | None = None, + limit: int | None = None, + model_providers: list[str] | None = None, + search_term: str | None = None, + sort_key: ThreadSortKey | None = None, + source_kinds: list[ThreadSourceKind] | None = None, + ) -> ThreadListResponse: + await self._ensure_initialized() + params = ThreadListParams( + archived=archived, + cursor=cursor, + cwd=cwd, + limit=limit, + model_providers=model_providers, + search_term=search_term, + sort_key=sort_key, + source_kinds=source_kinds, + ) + return await self._client.thread_list(params) + + async def thread_resume( + self, + thread_id: str, + *, + approval_policy: AskForApproval | None = None, + base_instructions: str | None = None, + config: JsonObject | None = None, + cwd: str | None = None, + developer_instructions: str | None = None, + model: str | None = None, + model_provider: str | None = None, + personality: Personality | None = None, + sandbox: SandboxMode | None = None, + service_tier: ServiceTier | None = None, + ) -> AsyncThread: + await self._ensure_initialized() + params = ThreadResumeParams( + thread_id=thread_id, + approval_policy=approval_policy, + base_instructions=base_instructions, + config=config, + cwd=cwd, + developer_instructions=developer_instructions, + model=model, + model_provider=model_provider, + personality=personality, + sandbox=sandbox, + service_tier=service_tier, + ) + resumed = await self._client.thread_resume(thread_id, params) + return AsyncThread(self, resumed.thread.id) + + async def thread_fork( + self, + thread_id: str, + *, + approval_policy: AskForApproval | None = None, + base_instructions: str | None = None, + config: JsonObject | None = None, + cwd: str | None = None, + developer_instructions: str | None = None, + model: str | None = None, + model_provider: str | None = None, + sandbox: SandboxMode | None = None, + service_tier: ServiceTier | None = None, + ) -> AsyncThread: + await self._ensure_initialized() + params = ThreadForkParams( + thread_id=thread_id, + approval_policy=approval_policy, + base_instructions=base_instructions, + config=config, + cwd=cwd, + developer_instructions=developer_instructions, + model=model, + model_provider=model_provider, + sandbox=sandbox, + service_tier=service_tier, + ) + forked = await self._client.thread_fork(thread_id, params) + return AsyncThread(self, forked.thread.id) + + async def thread_archive(self, thread_id: str) -> ThreadArchiveResponse: + await self._ensure_initialized() + return await self._client.thread_archive(thread_id) + + async def thread_unarchive(self, thread_id: str) -> AsyncThread: + await self._ensure_initialized() + unarchived = await self._client.thread_unarchive(thread_id) + return AsyncThread(self, unarchived.thread.id) + # END GENERATED: AsyncCodex.flat_methods + + async def models(self, *, include_hidden: bool = False) -> ModelListResponse: + await self._ensure_initialized() + return await self._client.model_list(include_hidden=include_hidden) + + +@dataclass(slots=True) +class Thread: + _client: AppServerClient + id: str + + # BEGIN GENERATED: Thread.flat_methods + def turn( + self, + input: Input, + *, + approval_policy: AskForApproval | None = None, + cwd: str | None = None, + effort: ReasoningEffort | None = None, + model: str | None = None, + output_schema: JsonObject | None = None, + personality: Personality | None = None, + sandbox_policy: SandboxPolicy | None = None, + service_tier: ServiceTier | None = None, + summary: ReasoningSummary | None = None, + ) -> Turn: + wire_input = _to_wire_input(input) + params = TurnStartParams( + thread_id=self.id, + input=wire_input, + approval_policy=approval_policy, + cwd=cwd, + effort=effort, + model=model, + output_schema=output_schema, + personality=personality, + sandbox_policy=sandbox_policy, + service_tier=service_tier, + summary=summary, + ) + turn = self._client.turn_start(self.id, wire_input, params=params) + return Turn(self._client, self.id, turn.turn.id) + # END GENERATED: Thread.flat_methods + + def read(self, *, include_turns: bool = False) -> ThreadReadResponse: + return self._client.thread_read(self.id, include_turns=include_turns) + + def set_name(self, name: str) -> ThreadSetNameResponse: + return self._client.thread_set_name(self.id, name) + + def compact(self) -> ThreadCompactStartResponse: + return self._client.thread_compact(self.id) + + +@dataclass(slots=True) +class AsyncThread: + _codex: AsyncCodex + id: str + + # BEGIN GENERATED: AsyncThread.flat_methods + async def turn( + self, + input: Input, + *, + approval_policy: AskForApproval | None = None, + cwd: str | None = None, + effort: ReasoningEffort | None = None, + model: str | None = None, + output_schema: JsonObject | None = None, + personality: Personality | None = None, + sandbox_policy: SandboxPolicy | None = None, + service_tier: ServiceTier | None = None, + summary: ReasoningSummary | None = None, + ) -> AsyncTurn: + await self._codex._ensure_initialized() + wire_input = _to_wire_input(input) + params = TurnStartParams( + thread_id=self.id, + input=wire_input, + approval_policy=approval_policy, + cwd=cwd, + effort=effort, + model=model, + output_schema=output_schema, + personality=personality, + sandbox_policy=sandbox_policy, + service_tier=service_tier, + summary=summary, + ) + turn = await self._codex._client.turn_start( + self.id, + wire_input, + params=params, + ) + return AsyncTurn(self._codex, self.id, turn.turn.id) + # END GENERATED: AsyncThread.flat_methods + + async def read(self, *, include_turns: bool = False) -> ThreadReadResponse: + await self._codex._ensure_initialized() + return await self._codex._client.thread_read(self.id, include_turns=include_turns) + + async def set_name(self, name: str) -> ThreadSetNameResponse: + await self._codex._ensure_initialized() + return await self._codex._client.thread_set_name(self.id, name) + + async def compact(self) -> ThreadCompactStartResponse: + await self._codex._ensure_initialized() + return await self._codex._client.thread_compact(self.id) + + +@dataclass(slots=True) +class Turn: + _client: AppServerClient + thread_id: str + id: str + + def steer(self, input: Input) -> TurnSteerResponse: + return self._client.turn_steer(self.thread_id, self.id, _to_wire_input(input)) + + def interrupt(self) -> TurnInterruptResponse: + return self._client.turn_interrupt(self.thread_id, self.id) + + def stream(self) -> Iterator[Notification]: + # TODO: replace this client-wide experimental guard with per-turn event demux. + self._client.acquire_turn_consumer(self.id) + try: + while True: + event = self._client.next_notification() + yield event + if ( + event.method == "turn/completed" + and isinstance(event.payload, TurnCompletedNotificationPayload) + and event.payload.turn.id == self.id + ): + break + finally: + self._client.release_turn_consumer(self.id) + + def run(self) -> TurnResult: + completed: TurnCompletedNotificationPayload | None = None + usage: ThreadTokenUsageUpdatedNotification | None = None + delta_chunks: list[str] = [] + raw_text_chunks: list[str] = [] + + stream = self.stream() + try: + for event in stream: + payload = event.payload + if ( + isinstance(payload, AgentMessageDeltaNotification) + and payload.turn_id == self.id + ): + delta_chunks.append(payload.delta) + continue + if ( + isinstance(payload, RawResponseItemCompletedNotification) + and payload.turn_id == self.id + ): + raw_text_chunks.extend(_assistant_output_text_chunks(payload)) + continue + if ( + isinstance(payload, ThreadTokenUsageUpdatedNotification) + and payload.turn_id == self.id + ): + usage = payload + continue + if ( + isinstance(payload, TurnCompletedNotificationPayload) + and payload.turn.id == self.id + ): + completed = payload + finally: + stream.close() + + return _build_turn_result(completed, usage, delta_chunks, raw_text_chunks) + + +@dataclass(slots=True) +class AsyncTurn: + _codex: AsyncCodex + thread_id: str + id: str + + async def steer(self, input: Input) -> TurnSteerResponse: + await self._codex._ensure_initialized() + return await self._codex._client.turn_steer( + self.thread_id, + self.id, + _to_wire_input(input), + ) + + async def interrupt(self) -> TurnInterruptResponse: + await self._codex._ensure_initialized() + return await self._codex._client.turn_interrupt(self.thread_id, self.id) + + async def stream(self) -> AsyncIterator[Notification]: + await self._codex._ensure_initialized() + # TODO: replace this client-wide experimental guard with per-turn event demux. + self._codex._client.acquire_turn_consumer(self.id) + try: + while True: + event = await self._codex._client.next_notification() + yield event + if ( + event.method == "turn/completed" + and isinstance(event.payload, TurnCompletedNotificationPayload) + and event.payload.turn.id == self.id + ): + break + finally: + self._codex._client.release_turn_consumer(self.id) + + async def run(self) -> TurnResult: + completed: TurnCompletedNotificationPayload | None = None + usage: ThreadTokenUsageUpdatedNotification | None = None + delta_chunks: list[str] = [] + raw_text_chunks: list[str] = [] + + stream = self.stream() + try: + async for event in stream: + payload = event.payload + if ( + isinstance(payload, AgentMessageDeltaNotification) + and payload.turn_id == self.id + ): + delta_chunks.append(payload.delta) + continue + if ( + isinstance(payload, RawResponseItemCompletedNotification) + and payload.turn_id == self.id + ): + raw_text_chunks.extend(_assistant_output_text_chunks(payload)) + continue + if ( + isinstance(payload, ThreadTokenUsageUpdatedNotification) + and payload.turn_id == self.id + ): + usage = payload + continue + if ( + isinstance(payload, TurnCompletedNotificationPayload) + and payload.turn.id == self.id + ): + completed = payload + finally: + await stream.aclose() + + return _build_turn_result(completed, usage, delta_chunks, raw_text_chunks) diff --git a/sdk/python/src/codex_app_server/public_types.py b/sdk/python/src/codex_app_server/public_types.py new file mode 100644 index 0000000000..d844a314cb --- /dev/null +++ b/sdk/python/src/codex_app_server/public_types.py @@ -0,0 +1,39 @@ +"""Shallow public aliases over the generated v2 wire models.""" + +from .generated.v2_all import ( + AskForApproval, + Personality, + PlanType, + ReasoningEffort, + ReasoningSummary, + SandboxMode, + SandboxPolicy, + ThreadForkParams, + ThreadListParams, + ThreadResumeParams, + ThreadSortKey, + ThreadSourceKind, + ThreadStartParams, + TurnStartParams, + TurnStatus, + TurnSteerParams, +) + +__all__ = [ + "AskForApproval", + "Personality", + "PlanType", + "ReasoningEffort", + "ReasoningSummary", + "SandboxMode", + "SandboxPolicy", + "ThreadForkParams", + "ThreadListParams", + "ThreadResumeParams", + "ThreadSortKey", + "ThreadSourceKind", + "ThreadStartParams", + "TurnStartParams", + "TurnStatus", + "TurnSteerParams", +] diff --git a/sdk/python/tests/test_async_client_behavior.py b/sdk/python/tests/test_async_client_behavior.py new file mode 100644 index 0000000000..580ff2a93b --- /dev/null +++ b/sdk/python/tests/test_async_client_behavior.py @@ -0,0 +1,64 @@ +from __future__ import annotations + +import asyncio +import time + +from codex_app_server.async_client import AsyncAppServerClient + + +def test_async_client_serializes_transport_calls() -> None: + async def scenario() -> int: + client = AsyncAppServerClient() + active = 0 + max_active = 0 + + def fake_model_list(include_hidden: bool = False) -> bool: + nonlocal active, max_active + active += 1 + max_active = max(max_active, active) + time.sleep(0.05) + active -= 1 + return include_hidden + + client._sync.model_list = fake_model_list # type: ignore[method-assign] + await asyncio.gather(client.model_list(), client.model_list()) + return max_active + + assert asyncio.run(scenario()) == 1 + + +def test_async_stream_text_is_incremental_and_blocks_parallel_calls() -> None: + async def scenario() -> tuple[str, list[str], bool]: + client = AsyncAppServerClient() + + def fake_stream_text(thread_id: str, text: str, params=None): # type: ignore[no-untyped-def] + yield "first" + time.sleep(0.03) + yield "second" + yield "third" + + def fake_model_list(include_hidden: bool = False) -> str: + return "done" + + client._sync.stream_text = fake_stream_text # type: ignore[method-assign] + client._sync.model_list = fake_model_list # type: ignore[method-assign] + + stream = client.stream_text("thread-1", "hello") + first = await anext(stream) + + blocked_before_stream_done = False + competing_call = asyncio.create_task(client.model_list()) + await asyncio.sleep(0.01) + blocked_before_stream_done = not competing_call.done() + + remaining: list[str] = [] + async for item in stream: + remaining.append(item) + + await competing_call + return first, remaining, blocked_before_stream_done + + first, remaining, blocked = asyncio.run(scenario()) + assert first == "first" + assert remaining == ["second", "third"] + assert blocked diff --git a/sdk/python/tests/test_public_api_runtime_behavior.py b/sdk/python/tests/test_public_api_runtime_behavior.py new file mode 100644 index 0000000000..414460905d --- /dev/null +++ b/sdk/python/tests/test_public_api_runtime_behavior.py @@ -0,0 +1,286 @@ +from __future__ import annotations + +import asyncio +from collections import deque +from pathlib import Path + +import pytest + +import codex_app_server.public_api as public_api_module +from codex_app_server.client import AppServerClient +from codex_app_server.generated.v2_all import ( + AgentMessageDeltaNotification, + RawResponseItemCompletedNotification, + ThreadTokenUsageUpdatedNotification, +) +from codex_app_server.models import InitializeResponse, Notification +from codex_app_server.public_api import AsyncCodex, AsyncTurn, Codex, Turn +from codex_app_server.public_types import TurnStatus + +ROOT = Path(__file__).resolve().parents[1] + + +def _delta_notification( + *, + thread_id: str = "thread-1", + turn_id: str = "turn-1", + text: str = "delta-text", +) -> Notification: + return Notification( + method="item/agentMessage/delta", + payload=AgentMessageDeltaNotification.model_validate( + { + "delta": text, + "itemId": "item-1", + "threadId": thread_id, + "turnId": turn_id, + } + ), + ) + + +def _raw_response_notification( + *, + thread_id: str = "thread-1", + turn_id: str = "turn-1", + text: str = "raw-text", +) -> Notification: + return Notification( + method="rawResponseItem/completed", + payload=RawResponseItemCompletedNotification.model_validate( + { + "item": { + "type": "message", + "role": "assistant", + "content": [{"type": "output_text", "text": text}], + }, + "threadId": thread_id, + "turnId": turn_id, + } + ), + ) + + +def _usage_notification( + *, + thread_id: str = "thread-1", + turn_id: str = "turn-1", +) -> Notification: + return Notification( + method="thread/tokenUsage/updated", + payload=ThreadTokenUsageUpdatedNotification.model_validate( + { + "threadId": thread_id, + "turnId": turn_id, + "tokenUsage": { + "last": { + "cachedInputTokens": 0, + "inputTokens": 1, + "outputTokens": 2, + "reasoningOutputTokens": 0, + "totalTokens": 3, + }, + "total": { + "cachedInputTokens": 0, + "inputTokens": 1, + "outputTokens": 2, + "reasoningOutputTokens": 0, + "totalTokens": 3, + }, + }, + } + ), + ) + + +def _completed_notification( + *, + thread_id: str = "thread-1", + turn_id: str = "turn-1", + status: str = "completed", +) -> Notification: + return Notification( + method="turn/completed", + payload=public_api_module.TurnCompletedNotificationPayload.model_validate( + { + "threadId": thread_id, + "turn": { + "id": turn_id, + "items": [], + "status": status, + }, + } + ), + ) + + +def test_codex_init_failure_closes_client(monkeypatch: pytest.MonkeyPatch) -> None: + closed: list[bool] = [] + + class FakeClient: + def __init__(self, config=None) -> None: # noqa: ANN001,ARG002 + self._closed = False + + def start(self) -> None: + return None + + def initialize(self) -> InitializeResponse: + return InitializeResponse.model_validate({}) + + def close(self) -> None: + self._closed = True + closed.append(True) + + monkeypatch.setattr(public_api_module, "AppServerClient", FakeClient) + + with pytest.raises(RuntimeError, match="missing required metadata"): + Codex() + + assert closed == [True] + + +def test_async_codex_init_failure_closes_client() -> None: + async def scenario() -> None: + codex = AsyncCodex() + close_calls = 0 + + async def fake_start() -> None: + return None + + async def fake_initialize() -> InitializeResponse: + return InitializeResponse.model_validate({}) + + async def fake_close() -> None: + nonlocal close_calls + close_calls += 1 + + codex._client.start = fake_start # type: ignore[method-assign] + codex._client.initialize = fake_initialize # type: ignore[method-assign] + codex._client.close = fake_close # type: ignore[method-assign] + + with pytest.raises(RuntimeError, match="missing required metadata"): + await codex.models() + + assert close_calls == 1 + assert codex._initialized is False + assert codex._init is None + + asyncio.run(scenario()) + + +def test_async_codex_initializes_only_once_under_concurrency() -> None: + async def scenario() -> None: + codex = AsyncCodex() + start_calls = 0 + initialize_calls = 0 + ready = asyncio.Event() + + async def fake_start() -> None: + nonlocal start_calls + start_calls += 1 + + async def fake_initialize() -> InitializeResponse: + nonlocal initialize_calls + initialize_calls += 1 + ready.set() + await asyncio.sleep(0.02) + return InitializeResponse.model_validate( + { + "userAgent": "codex-cli/1.2.3", + "serverInfo": {"name": "codex-cli", "version": "1.2.3"}, + } + ) + + async def fake_model_list(include_hidden: bool = False): # noqa: ANN202,ARG001 + await ready.wait() + return object() + + codex._client.start = fake_start # type: ignore[method-assign] + codex._client.initialize = fake_initialize # type: ignore[method-assign] + codex._client.model_list = fake_model_list # type: ignore[method-assign] + + await asyncio.gather(codex.models(), codex.models()) + + assert start_calls == 1 + assert initialize_calls == 1 + + asyncio.run(scenario()) + + +def test_turn_stream_rejects_second_active_consumer() -> None: + client = AppServerClient() + notifications: deque[Notification] = deque( + [ + _delta_notification(turn_id="turn-1"), + _completed_notification(turn_id="turn-1"), + ] + ) + client.next_notification = notifications.popleft # type: ignore[method-assign] + + first_stream = Turn(client, "thread-1", "turn-1").stream() + assert next(first_stream).method == "item/agentMessage/delta" + + second_stream = Turn(client, "thread-1", "turn-2").stream() + with pytest.raises(RuntimeError, match="Concurrent turn consumers are not yet supported"): + next(second_stream) + + first_stream.close() + + +def test_async_turn_stream_rejects_second_active_consumer() -> None: + async def scenario() -> None: + codex = AsyncCodex() + + async def fake_ensure_initialized() -> None: + return None + + notifications: deque[Notification] = deque( + [ + _delta_notification(turn_id="turn-1"), + _completed_notification(turn_id="turn-1"), + ] + ) + + async def fake_next_notification() -> Notification: + return notifications.popleft() + + codex._ensure_initialized = fake_ensure_initialized # type: ignore[method-assign] + codex._client.next_notification = fake_next_notification # type: ignore[method-assign] + + first_stream = AsyncTurn(codex, "thread-1", "turn-1").stream() + assert (await anext(first_stream)).method == "item/agentMessage/delta" + + second_stream = AsyncTurn(codex, "thread-1", "turn-2").stream() + with pytest.raises(RuntimeError, match="Concurrent turn consumers are not yet supported"): + await anext(second_stream) + + await first_stream.aclose() + + asyncio.run(scenario()) + + +def test_turn_run_falls_back_to_completed_raw_response_text() -> None: + client = AppServerClient() + notifications: deque[Notification] = deque( + [ + _raw_response_notification(text="hello from raw response"), + _usage_notification(), + _completed_notification(), + ] + ) + client.next_notification = notifications.popleft # type: ignore[method-assign] + + result = Turn(client, "thread-1", "turn-1").run() + + assert result.status == TurnStatus.completed + assert result.text == "hello from raw response" + + +def test_retry_examples_compare_status_with_enum() -> None: + for path in ( + ROOT / "examples" / "10_error_handling_and_retry" / "sync.py", + ROOT / "examples" / "10_error_handling_and_retry" / "async.py", + ): + source = path.read_text() + assert '== "failed"' not in source + assert "TurnStatus.failed" in source diff --git a/sdk/python/tests/test_public_api_signatures.py b/sdk/python/tests/test_public_api_signatures.py new file mode 100644 index 0000000000..9d16a32233 --- /dev/null +++ b/sdk/python/tests/test_public_api_signatures.py @@ -0,0 +1,209 @@ +from __future__ import annotations + +import importlib.resources as resources +import inspect +from typing import Any + +from codex_app_server import AppServerConfig +from codex_app_server.models import InitializeResponse +from codex_app_server.public_api import AsyncCodex, AsyncThread, Codex, Thread + + +def _keyword_only_names(fn: object) -> list[str]: + signature = inspect.signature(fn) + return [ + param.name + for param in signature.parameters.values() + if param.kind == inspect.Parameter.KEYWORD_ONLY + ] + + +def _assert_no_any_annotations(fn: object) -> None: + signature = inspect.signature(fn) + for param in signature.parameters.values(): + if param.annotation is Any: + raise AssertionError(f"{fn} has public parameter typed as Any: {param.name}") + if signature.return_annotation is Any: + raise AssertionError(f"{fn} has public return annotation typed as Any") + + +def test_root_exports_app_server_config() -> None: + assert AppServerConfig.__name__ == "AppServerConfig" + + +def test_package_includes_py_typed_marker() -> None: + marker = resources.files("codex_app_server").joinpath("py.typed") + assert marker.is_file() + + +def test_generated_public_signatures_are_snake_case_and_typed() -> None: + expected = { + Codex.thread_start: [ + "approval_policy", + "base_instructions", + "config", + "cwd", + "developer_instructions", + "ephemeral", + "model", + "model_provider", + "personality", + "sandbox", + "service_name", + "service_tier", + ], + Codex.thread_list: [ + "archived", + "cursor", + "cwd", + "limit", + "model_providers", + "search_term", + "sort_key", + "source_kinds", + ], + Codex.thread_resume: [ + "approval_policy", + "base_instructions", + "config", + "cwd", + "developer_instructions", + "model", + "model_provider", + "personality", + "sandbox", + "service_tier", + ], + Codex.thread_fork: [ + "approval_policy", + "base_instructions", + "config", + "cwd", + "developer_instructions", + "model", + "model_provider", + "sandbox", + "service_tier", + ], + Thread.turn: [ + "approval_policy", + "cwd", + "effort", + "model", + "output_schema", + "personality", + "sandbox_policy", + "service_tier", + "summary", + ], + AsyncCodex.thread_start: [ + "approval_policy", + "base_instructions", + "config", + "cwd", + "developer_instructions", + "ephemeral", + "model", + "model_provider", + "personality", + "sandbox", + "service_name", + "service_tier", + ], + AsyncCodex.thread_list: [ + "archived", + "cursor", + "cwd", + "limit", + "model_providers", + "search_term", + "sort_key", + "source_kinds", + ], + AsyncCodex.thread_resume: [ + "approval_policy", + "base_instructions", + "config", + "cwd", + "developer_instructions", + "model", + "model_provider", + "personality", + "sandbox", + "service_tier", + ], + AsyncCodex.thread_fork: [ + "approval_policy", + "base_instructions", + "config", + "cwd", + "developer_instructions", + "model", + "model_provider", + "sandbox", + "service_tier", + ], + AsyncThread.turn: [ + "approval_policy", + "cwd", + "effort", + "model", + "output_schema", + "personality", + "sandbox_policy", + "service_tier", + "summary", + ], + } + + for fn, expected_kwargs in expected.items(): + actual = _keyword_only_names(fn) + assert actual == expected_kwargs, f"unexpected kwargs for {fn}: {actual}" + assert all(name == name.lower() for name in actual), f"non snake_case kwargs in {fn}: {actual}" + _assert_no_any_annotations(fn) + + +def test_lifecycle_methods_are_codex_scoped() -> None: + assert hasattr(Codex, "thread_resume") + assert hasattr(Codex, "thread_fork") + assert hasattr(Codex, "thread_archive") + assert hasattr(Codex, "thread_unarchive") + assert hasattr(AsyncCodex, "thread_resume") + assert hasattr(AsyncCodex, "thread_fork") + assert hasattr(AsyncCodex, "thread_archive") + assert hasattr(AsyncCodex, "thread_unarchive") + assert not hasattr(Codex, "thread") + assert not hasattr(AsyncCodex, "thread") + + assert not hasattr(Thread, "resume") + assert not hasattr(Thread, "fork") + assert not hasattr(Thread, "archive") + assert not hasattr(Thread, "unarchive") + assert not hasattr(AsyncThread, "resume") + assert not hasattr(AsyncThread, "fork") + assert not hasattr(AsyncThread, "archive") + assert not hasattr(AsyncThread, "unarchive") + + for fn in ( + Codex.thread_archive, + Codex.thread_unarchive, + AsyncCodex.thread_archive, + AsyncCodex.thread_unarchive, + ): + _assert_no_any_annotations(fn) + + +def test_initialize_metadata_parses_user_agent_shape() -> None: + parsed = Codex._parse_initialize(InitializeResponse.model_validate({"userAgent": "codex-cli/1.2.3"})) + assert parsed.user_agent == "codex-cli/1.2.3" + assert parsed.server_name == "codex-cli" + assert parsed.server_version == "1.2.3" + + +def test_initialize_metadata_requires_non_empty_information() -> None: + try: + Codex._parse_initialize(InitializeResponse.model_validate({})) + except RuntimeError as exc: + assert "missing required metadata" in str(exc) + else: + raise AssertionError("expected RuntimeError when initialize metadata is missing") diff --git a/sdk/python/tests/test_real_app_server_integration.py b/sdk/python/tests/test_real_app_server_integration.py new file mode 100644 index 0000000000..f324cf4bf7 --- /dev/null +++ b/sdk/python/tests/test_real_app_server_integration.py @@ -0,0 +1,215 @@ +from __future__ import annotations + +import asyncio +import json +import os +import subprocess +import sys +import tempfile +from pathlib import Path + +import pytest + +from codex_app_server import AsyncCodex, Codex, TextInput + +ROOT = Path(__file__).resolve().parents[1] +EXAMPLES_DIR = ROOT / "examples" +NOTEBOOK_PATH = ROOT / "notebooks" / "sdk_walkthrough.ipynb" + +# 11_cli_mini_app is interactive; we still run it by feeding '/exit'. +EXAMPLE_CASES: list[tuple[str, str]] = [ + ("01_quickstart_constructor", "sync.py"), + ("01_quickstart_constructor", "async.py"), + ("02_turn_run", "sync.py"), + ("02_turn_run", "async.py"), + ("03_turn_stream_events", "sync.py"), + ("03_turn_stream_events", "async.py"), + ("04_models_and_metadata", "sync.py"), + ("04_models_and_metadata", "async.py"), + ("05_existing_thread", "sync.py"), + ("05_existing_thread", "async.py"), + ("06_thread_lifecycle_and_controls", "sync.py"), + ("06_thread_lifecycle_and_controls", "async.py"), + ("07_image_and_text", "sync.py"), + ("07_image_and_text", "async.py"), + ("08_local_image_and_text", "sync.py"), + ("08_local_image_and_text", "async.py"), + ("09_async_parity", "sync.py"), + # 09_async_parity async path is represented by 01 async + dedicated async-based cases above. + ("10_error_handling_and_retry", "sync.py"), + ("10_error_handling_and_retry", "async.py"), + ("11_cli_mini_app", "sync.py"), + ("11_cli_mini_app", "async.py"), + ("12_turn_params_kitchen_sink", "sync.py"), + ("12_turn_params_kitchen_sink", "async.py"), + ("13_model_select_and_turn_params", "sync.py"), + ("13_model_select_and_turn_params", "async.py"), +] + + +def _run_example( + folder: str, script: str, *, timeout_s: int = 150 +) -> subprocess.CompletedProcess[str]: + path = EXAMPLES_DIR / folder / script + assert path.exists(), f"Missing example script: {path}" + + env = os.environ.copy() + + # Feed '/exit' only to interactive mini-cli examples. + stdin = "/exit\n" if folder == "11_cli_mini_app" else None + + return subprocess.run( + [sys.executable, str(path)], + cwd=str(ROOT), + env=env, + input=stdin, + text=True, + capture_output=True, + timeout=timeout_s, + check=False, + ) + + +def _notebook_cell_source(cell_index: int) -> str: + notebook = json.loads(NOTEBOOK_PATH.read_text()) + return "".join(notebook["cells"][cell_index]["source"]) + + +def test_real_initialize_and_model_list(): + with Codex() as codex: + metadata = codex.metadata + assert isinstance(metadata.user_agent, str) and metadata.user_agent.strip() + assert isinstance(metadata.server_name, str) and metadata.server_name.strip() + assert isinstance(metadata.server_version, str) and metadata.server_version.strip() + + models = codex.models(include_hidden=True) + assert isinstance(models.data, list) + + +def test_real_thread_and_turn_start_smoke(): + with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + result = thread.turn(TextInput("hello")).run() + + assert isinstance(result.thread_id, str) and result.thread_id.strip() + assert isinstance(result.turn_id, str) and result.turn_id.strip() + assert isinstance(result.items, list) + assert result.usage is not None + assert result.usage.thread_id == result.thread_id + assert result.usage.turn_id == result.turn_id + + +def test_real_async_thread_turn_usage_and_ids_smoke() -> None: + async def _run() -> None: + async with AsyncCodex() as codex: + thread = await codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + result = await (await thread.turn(TextInput("say ok"))).run() + + assert isinstance(result.thread_id, str) and result.thread_id.strip() + assert isinstance(result.turn_id, str) and result.turn_id.strip() + assert isinstance(result.items, list) + assert result.usage is not None + assert result.usage.thread_id == result.thread_id + assert result.usage.turn_id == result.turn_id + + asyncio.run(_run()) + + +def test_notebook_bootstrap_resolves_sdk_from_unrelated_cwd() -> None: + cell_1_source = _notebook_cell_source(1) + env = os.environ.copy() + env["CODEX_PYTHON_SDK_DIR"] = str(ROOT) + + with tempfile.TemporaryDirectory() as temp_cwd: + result = subprocess.run( + [sys.executable, "-c", cell_1_source], + cwd=temp_cwd, + env=env, + text=True, + capture_output=True, + timeout=60, + check=False, + ) + + assert result.returncode == 0, ( + f"Notebook bootstrap failed from unrelated cwd.\n" + f"STDOUT:\n{result.stdout}\n" + f"STDERR:\n{result.stderr}" + ) + assert "SDK source:" in result.stdout + assert "codex_app_server" in result.stdout or "sdk/python/src" in result.stdout + + +def test_real_streaming_smoke_turn_completed(): + with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = thread.turn(TextInput("Reply with one short sentence.")) + + saw_delta = False + saw_completed = False + for evt in turn.stream(): + if evt.method == "item/agentMessage/delta": + saw_delta = True + if evt.method == "turn/completed": + saw_completed = True + + assert saw_completed + # Some environments can produce zero deltas for very short output; + # this assert keeps the smoke test informative but non-flaky. + assert isinstance(saw_delta, bool) + + +def test_real_turn_interrupt_smoke(): + with Codex() as codex: + thread = codex.thread_start(model="gpt-5", config={"model_reasoning_effort": "high"}) + turn = thread.turn(TextInput("Count from 1 to 200 with commas.")) + + # Best effort: interrupting quickly may race with completion on fast models. + _ = turn.interrupt() + + # Confirm the session is still usable after interrupt race. + follow_up = thread.turn(TextInput("Say 'ok' only.")).run() + assert follow_up.status.value in {"completed", "failed"} + +@pytest.mark.parametrize(("folder", "script"), EXAMPLE_CASES) +def test_real_examples_run_and_assert(folder: str, script: str): + result = _run_example(folder, script) + + assert result.returncode == 0, ( + f"Example failed: {folder}/{script}\n" + f"STDOUT:\n{result.stdout}\n" + f"STDERR:\n{result.stderr}" + ) + + out = result.stdout + + # Minimal content assertions so we validate behavior, not just exit code. + if folder == "01_quickstart_constructor": + assert "Status:" in out and "Text:" in out + assert "Server: None None" not in out + elif folder == "02_turn_run": + assert "thread_id:" in out and "turn_id:" in out and "status:" in out + assert "usage: None" not in out + elif folder == "03_turn_stream_events": + assert "turn/completed" in out + elif folder == "04_models_and_metadata": + assert "models.count:" in out + assert "server_name=None" not in out + assert "server_version=None" not in out + elif folder == "05_existing_thread": + assert "Created thread:" in out + elif folder == "06_thread_lifecycle_and_controls": + assert "Lifecycle OK:" in out + elif folder in {"07_image_and_text", "08_local_image_and_text"}: + assert "completed" in out.lower() or "Status:" in out + elif folder == "09_async_parity": + assert "Thread:" in out and "Turn:" in out + elif folder == "10_error_handling_and_retry": + assert "Text:" in out + elif folder == "11_cli_mini_app": + assert "Thread:" in out + elif folder == "12_turn_params_kitchen_sink": + assert "Status:" in out and "Usage:" in out + elif folder == "13_model_select_and_turn_params": + assert "selected.model:" in out and "agent.message.params:" in out and "usage.params:" in out + assert "usage.params: None" not in out