Cookbook

# Examples and Integration Recipes

Ready-to-run snippets from `examples/` and integration patterns for sync/async, retries, and persistent sessions.

## Examples in This Repository

File | Main Method | Purpose
--- | --- | ---
`examples/run_simple.py` | `client.run()` | Basic prompt execution and final message output.
`examples/run_json_events.py` | `json_output=True` | Collect events, thread id, turn status, usage.
`examples/run_live_stream.py` | `client.run_live()` | Stream JSON events from a running sync process.
`examples/run_thread_session.py` | `start_thread()` | Create session and continue with follow-up prompts.
`examples/run_with_schema.py` | `run_with_schema()` | Constrain output to JSON schema and parse final JSON.
`examples/run_async.py` | `run_async()` + `run_live_async()` | Async run plus async event streaming.
`examples/run_persistent_session_store.py` | `JsonFileSessionStore` | Persist named sessions and resume by name.

## Quick Recipes

### 1. Plain Prompt (Minimal)

    from codex_local_sdk import CodexLocalClient

    client = CodexLocalClient()
    result = client.run_prompt("Summarize this project in 5 bullets.")
    print(result.final_message)

### 2. Explicit Request with Sandbox

    from codex_local_sdk import CodexExecRequest, CodexLocalClient, SandboxMode

    client = CodexLocalClient()
    result = client.run(
        CodexExecRequest(
            prompt="Find architecture risks.",
            sandbox=SandboxMode.READ_ONLY,
            skip_git_repo_check=True,
        ),
        timeout_seconds=120,
    )

## Async Pattern

    import asyncio
    from codex_local_sdk import CodexExecRequest, CodexLocalClient

    async def main() -> None:
        client = CodexLocalClient()

        result = await client.run_async(CodexExecRequest(prompt="Analyze repo"), timeout_seconds=120)
        print(result.final_message)

        live = await client.run_live_async(CodexExecRequest(prompt="Stream repo analysis", json_output=True))
        async for event in live.iter_events():
            print(event.type)
        final = await live.wait()
        print(final.turn_status)

    asyncio.run(main())

## Retry/Backoff Pattern

    from codex_local_sdk import CodexExecRequest, CodexLocalClient, RetryPolicy

    client = CodexLocalClient(
        retry_policy=RetryPolicy(
            max_attempts=3,
            initial_backoff_seconds=0.5,
            backoff_multiplier=2.0,
            max_backoff_seconds=4.0,
            retry_on_exit_codes=(7, 9),
            jitter_ratio=0.2,
            max_total_retry_seconds=15.0,
            retry_on_timeouts=True,
        )
    )

    result = client.run(CodexExecRequest(prompt="Run resiliently"), timeout_seconds=30)
    print(result.return_code, result.ok)

## Persistent Session Pattern

    from codex_local_sdk import CodexLocalClient, JsonFileSessionStore, SandboxMode

    client = CodexLocalClient(session_store=JsonFileSessionStore(".codex_sessions.json"))

    session, first = client.start_thread(
        prompt="Create migration plan",
        sandbox=SandboxMode.READ_ONLY,
        session_name="migration-plan",
    )
    print(session.session_id, first.turn_status)

    second = client.resume(
        prompt="Continue with phase 1 tasks",
        session_name="migration-plan",
        last=False,
        json_output=True,
        timeout_seconds=120,
    )
    print(second.final_message)

    same_session = client.open_session("migration-plan")
    third = same_session.continue_prompt("Continue with test strategy", json_output=True)
    print(third.turn_status)

## Telemetry Hook Pattern

    from codex_local_sdk import CodexClientEvent, CodexExecRequest, CodexLocalClient

    def on_event(event: CodexClientEvent) -> None:
        print(event.type, event.operation, event.attempt, event.return_code)

    client = CodexLocalClient(event_hook=on_event)
    client.run(CodexExecRequest(prompt="Analyze repo"), timeout_seconds=60)

## Service Integration Skeleton

    class CodexService:
        def __init__(self) -> None:
            self.client = CodexLocalClient(default_cwd="/workspace/project")

        def analyze_repo(self) -> str:
            result = self.client.run_prompt(
                "Provide architecture summary and top 5 technical risks.",
                json_output=True,
            )
            return result.final_message or ""

        async def analyze_repo_async(self) -> str:
            result = await self.client.run_async(
                CodexExecRequest(prompt="Provide async architecture summary", json_output=True)
            )
            return result.final_message or ""