3
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?

@nogataka

Python版 Codex SDK を作ったので、使い方を徹底解説する

3
Posted at

はじめに

2025年末、AIコーディングエージェント市場で大きな変化がありました。Codex CLIがGitHub Copilotのシェアを追い抜いたのです。

参考記事によると、現在のシェアは以下のようになっています：

ツール シェア
Claude Code 29.1%
Codex CLI 25.2%
Copilot Agent 25.1%
Cursor 11.0%
Gemini CLI 6.6%

この流れを受けて、Codex CLIへの注目度は日に日に高まっています。

しかし、公式のCodex SDKはTypeScript版しか提供されていません。Pythonユーザーとしてこれは不便...ということで、Python版のCodex SDKを自作しました！

この記事では、Python版Codex SDKの使い方を、基礎から応用まで徹底解説します。この記事を読めば、あなたもCodex SDKを使った自分だけのAIアプリを作れるようになります。

リポジトリ

Python版Codex SDKは以下のリポジトリで公開しています：

pip install codex-sdk-py

GitHub: https://github.com/nogataka/codex-sdk-py

Codex SDKとは？

Codex SDKは、OpenAIが提供するCodex CLIをプログラムから制御するためのライブラリです。

通常、Codex CLIはターミナルから手動で操作しますが、SDKを使うことで：

  • 自動化: スクリプトから自動でコード生成・修正を実行
  • 統合: 自分のアプリケーションにCodexの機能を組み込む
  • カスタマイズ: 独自のワークフローを構築

といったことが可能になります。

内部的には、SDKはCodex CLIのバイナリを起動し、JSONL形式でイベントをやり取りすることで通信しています。

TypeScript版との完全互換性

このPython版SDKは、公式TypeScript版SDKを忠実にポーティングしています。「Pythonで書き直した劣化版」ではなく、全ての機能・データ構造・動作が完全に同一です。

ファイル構成の対応

TypeScript Python 説明
src/codex.ts codex_sdk/codex.py Codex クラス
src/thread.ts codex_sdk/thread.py Thread クラス
src/exec.ts codex_sdk/exec.py CodexExec クラス
src/events.ts codex_sdk/events.py イベント型定義
src/items.ts codex_sdk/items.py アイテム型定義

型システムの完全対応

イベント型（全8種）

ThreadStartedEvent, TurnStartedEvent, TurnCompletedEvent, TurnFailedEvent,
ItemStartedEvent, ItemUpdatedEvent, ItemCompletedEvent, ThreadErrorEvent

アイテム型（全8種）

AgentMessageItem, ReasoningItem, CommandExecutionItem, FileChangeItem,
McpToolCallItem, WebSearchItem, TodoListItem, ErrorItem

Enum（全4種）

ApprovalMode, SandboxMode, ModelReasoningEffort, WebSearchMode

CLI引数・環境変数も完全一致

TypeScript版と同じCLI引数を生成し、同じ環境変数を使用します：

項目 対応状況
--model, --sandbox, --cd 等のCLI引数 ✅ 完全一致
OPENAI_BASE_URL, CODEX_API_KEY 等の環境変数 ✅ 完全一致
TOML設定のシリアライズロジック ✅ 完全一致
出力スキーマファイルの処理 ✅ 完全一致

言語慣例による差異（動作への影響なし）

項目 TypeScript Python
命名規則 camelCase snake_case
キャンセル機構 AbortSignal asyncio.Event
Enum実装 文字列リテラルUnion StrEnum

これらは各言語の標準的な慣例に従ったものであり、機能的な違いは一切ありません。TypeScript版のドキュメントやサンプルコードは、命名規則を変換するだけでそのままPython版でも使えます。

インストール

前提条件

  • Python 3.10以上
  • Codex CLIがインストールされていること
  • OpenAI APIキーが設定されていること

SDKのインストール

pip install codex-sdk-py

これだけです！

基本的な使い方

最小構成のコード

まずは最もシンプルな例から始めましょう。

import asyncio
from codex_sdk import Codex

async def main():
    # 1. Codexクライアントを作成
    codex = Codex()

    # 2. 新しいスレッド（会話）を開始
    thread = codex.start_thread()

    # 3. プロンプトを送信して結果を取得
    turn = await thread.run("Pythonでフィボナッチ数列を計算する関数を作成して")

    # 4. 結果を表示
    print(turn.final_response)

asyncio.run(main())

このコードのポイントを解説します：

概念 説明
Codex SDKのメインエントリーポイント。設定を渡してインスタンス化します
Thread 1つの会話セッションを表します。複数ターンの対話が可能です
Turn 1回のやり取り（ユーザーの入力→エージェントの応答）の結果です

会話を継続する

同じThreadインスタンスでrun()を繰り返し呼び出すことで、文脈を維持したまま会話を継続できます。

async def conversation_example():
    codex = Codex()
    thread = codex.start_thread()

    # 最初のターン：関数を作成
    turn1 = await thread.run("Pythonでフィボナッチ数列を計算する関数を作成して")
    print("=== Turn 1 ===")
    print(turn1.final_response)

    # 2回目のターン：テストを追加（前の文脈を覚えている）
    turn2 = await thread.run("その関数のユニットテストを書いて")
    print("\n=== Turn 2 ===")
    print(turn2.final_response)

    # 3回目のターン：最適化（さらに文脈を維持）
    turn3 = await thread.run("メモ化を使って最適化して")
    print("\n=== Turn 3 ===")
    print(turn3.final_response)

asyncio.run(conversation_example())

ストリーミングレスポンス

run()メソッドは、ターンが完了するまで待ってから結果を返します。しかし、リアルタイムで進捗を確認したい場合はrun_streamed()を使います。

async def streaming_example():
    codex = Codex()
    thread = codex.start_thread()

    # ストリーミングモードで実行
    streamed = await thread.run_streamed("大規模なリファクタリングを実行して")

    # イベントをリアルタイムで処理
    async for event in streamed.events:
        event_type = event.get("type")

        match event_type:
            case "item.started":
                print(f"[開始] アイテム処理を開始...")

            case "item.completed":
                item = event.get("item", {})
                item_type = item.get("type")
                print(f"[完了] {item_type}")

            case "turn.completed":
                usage = event.get("usage", {})
                print(f"\n[終了] トークン使用量: {usage}")

asyncio.run(streaming_example())

イベントの種類

ストリーミングで受け取れるイベントは以下の8種類です：

イベント 説明
thread.started スレッドが開始された
turn.started ターンが開始された
turn.completed ターンが完了した（使用量情報を含む）
turn.failed ターンが失敗した（エラー情報を含む）
item.started アイテムの処理が開始された
item.updated アイテムが更新された
item.completed アイテムの処理が完了した
thread.error 回復不能なエラーが発生した

構造化出力（Structured Output）

AIの応答をJSON形式で受け取ることができます。これにより、プログラムで処理しやすい形式でデータを取得できます。

基本的な使い方

async def structured_output_example():
    codex = Codex()
    thread = codex.start_thread()

    # JSONスキーマを定義
    schema = {
        "type": "object",
        "properties": {
            "summary": {
                "type": "string",
                "description": "変更内容の要約"
            },
            "files_changed": {
                "type": "array",
                "items": {"type": "string"},
                "description": "変更されたファイルのリスト"
            },
            "status": {
                "type": "string",
                "enum": ["success", "partial", "failed"],
                "description": "処理結果のステータス"
            }
        },
        "required": ["summary", "files_changed", "status"],
        "additionalProperties": False
    }

    # スキーマを指定して実行
    turn = await thread.run(
        "このリポジトリの状態を分析して",
        {"output_schema": schema}
    )

    # JSONとして解析可能な応答を取得
    import json
    result = json.loads(turn.final_response)

    print(f"ステータス: {result['status']}")
    print(f"要約: {result['summary']}")
    print(f"変更ファイル数: {len(result['files_changed'])}")

asyncio.run(structured_output_example())

Pydanticとの連携

Pydanticモデルを使うと、より型安全に構造化出力を扱えます。

from pydantic import BaseModel
from typing import Literal

class CodeReviewResult(BaseModel):
    """コードレビューの結果"""
    issues: list[str]
    suggestions: list[str]
    severity: Literal["low", "medium", "high", "critical"]
    approved: bool

async def pydantic_example():
    codex = Codex()
    thread = codex.start_thread()

    # Pydanticモデルからスキーマを生成
    turn = await thread.run(
        "このコードをレビューして",
        {"output_schema": CodeReviewResult.model_json_schema()}
    )

    # Pydanticモデルとしてパース
    result = CodeReviewResult.model_validate_json(turn.final_response)

    if result.approved:
        print("コードレビュー: 承認")
    else:
        print("コードレビュー: 要修正")
        for issue in result.issues:
            print(f"  - {issue}")

asyncio.run(pydantic_example())

画像の添付

スクリーンショットや図を添付して、視覚的な情報をCodexに渡すことができます。

async def image_example():
    codex = Codex()
    thread = codex.start_thread()

    # テキストと画像を組み合わせた入力
    turn = await thread.run([
        {"type": "text", "text": "このUIのスクリーンショットを分析して、改善点を提案して"},
        {"type": "local_image", "path": "./screenshots/ui.png"},
        {"type": "local_image", "path": "./screenshots/error_screen.jpg"},
    ])

    print(turn.final_response)

asyncio.run(image_example())

入力タイプは2種類あります：

タイプ 説明
text テキスト入力
local_image ローカルの画像ファイル

サンドボックスモード

Codexがファイルシステムにどのようにアクセスできるかを制御できます。

from codex_sdk import Codex, SandboxMode

async def sandbox_example():
    codex = Codex()

    # 読み取り専用モード（安全）
    read_only_thread = codex.start_thread({
        "sandbox_mode": SandboxMode.READ_ONLY,
    })

    # ワークスペース内のみ書き込み可能（推奨）
    workspace_thread = codex.start_thread({
        "sandbox_mode": SandboxMode.WORKSPACE_WRITE,
    })

    # フルアクセス（危険！）
    full_access_thread = codex.start_thread({
        "sandbox_mode": SandboxMode.DANGER_FULL_ACCESS,
    })

各モードの説明

モード 説明 用途
READ_ONLY 読み取りのみ コード分析、レビュー
WORKSPACE_WRITE ワークスペース内のみ書き込み可 通常の開発作業
DANGER_FULL_ACCESS システム全体へのアクセス システム管理タスク（慎重に！）

承認ポリシー

エージェントがアクションを実行する前に承認を求めるかどうかを制御します。

from codex_sdk import Codex, ApprovalMode

async def approval_example():
    codex = Codex()

    # 承認なしで自動実行（完全自動化向け）
    auto_thread = codex.start_thread({
        "approval_policy": ApprovalMode.NEVER,
    })

    # 失敗時のみ承認を求める（バランス型）
    on_failure_thread = codex.start_thread({
        "approval_policy": ApprovalMode.ON_FAILURE,
    })

    # 常に承認を求める（安全重視）
    always_approve_thread = codex.start_thread({
        "approval_policy": ApprovalMode.UNTRUSTED,
    })

各モードの説明

モード 説明 用途
NEVER 承認なし CI/CD、バッチ処理
ON_REQUEST 明示的なリクエスト時のみ 通常の対話
ON_FAILURE 失敗時のみ 自動化 + 安全性
UNTRUSTED 常に承認 本番環境、重要な操作

作業ディレクトリの設定

デフォルトでは現在のディレクトリで動作しますが、明示的に指定することもできます。

async def working_directory_example():
    codex = Codex()

    # 特定のプロジェクトディレクトリで作業
    thread = codex.start_thread({
        "working_directory": "/path/to/my-project",
    })

    # Gitリポジトリでない場合はチェックをスキップ
    thread_no_git = codex.start_thread({
        "working_directory": "/path/to/non-git-folder",
        "skip_git_repo_check": True,
    })

asyncio.run(working_directory_example())

Note: Codexは通常、作業ディレクトリがGitリポジトリであることを要求します。非Gitディレクトリで作業する場合はskip_git_repo_check: Trueを設定してください。

スレッドの再開

セッションは~/.codex/sessionsに保存されます。プログラムを再起動しても、スレッドIDがあれば会話を再開できます。

import os

async def resume_example():
    codex = Codex()

    # 最初のセッション
    thread = codex.start_thread()
    turn = await thread.run("プロジェクトを分析して")

    # スレッドIDを保存
    thread_id = thread.id
    print(f"スレッドID: {thread_id}")

    # ... 後で再開 ...

    # スレッドを再開
    resumed_thread = codex.resume_thread(thread_id)
    turn2 = await resumed_thread.run("さっきの分析結果を基に改善して")
    print(turn2.final_response)

asyncio.run(resume_example())

タスクのキャンセル

長時間実行されるタスクを途中でキャンセルできます。

import asyncio

async def cancel_example():
    codex = Codex()
    thread = codex.start_thread()

    # キャンセル用のイベントを作成
    cancel_event = asyncio.Event()

    async def cancel_after_timeout():
        """30秒後にキャンセル"""
        await asyncio.sleep(30)
        print("タイムアウト！キャンセルします...")
        cancel_event.set()

    # キャンセルタイマーを開始
    cancel_task = asyncio.create_task(cancel_after_timeout())

    try:
        turn = await thread.run(
            "大規模なコードベースを全て分析して",
            {"cancel_event": cancel_event}
        )
        print(turn.final_response)
    except asyncio.CancelledError:
        print("タスクがキャンセルされました")
    finally:
        cancel_task.cancel()

asyncio.run(cancel_example())

環境変数と設定のカスタマイズ

環境変数の制御

codex = Codex({
    "env": {
        "PATH": "/usr/local/bin:/usr/bin",
        "MY_API_KEY": "secret_key",
    },
})

設定のオーバーライド

codex = Codex({
    "config": {
        "show_raw_agent_reasoning": True,  # 推論過程を表示
        "sandbox_workspace_write": {
            "network_access": True,  # ネットワークアクセスを許可
        },
    },
})

実践的なサンプル：コードレビューボット

これまでの知識を組み合わせて、実践的なコードレビューボットを作ってみましょう。

import asyncio
import json
from pydantic import BaseModel
from typing import Literal
from codex_sdk import Codex, SandboxMode, ApprovalMode

class ReviewResult(BaseModel):
    """コードレビュー結果"""
    overall_score: int  # 1-10
    issues: list[str]
    suggestions: list[str]
    security_concerns: list[str]
    verdict: Literal["approve", "request_changes", "needs_discussion"]

async def code_review_bot(file_path: str):
    """
    指定されたファイルのコードレビューを実行する
    """
    # 安全な設定でCodexを初期化
    codex = Codex()
    thread = codex.start_thread({
        "sandbox_mode": SandboxMode.READ_ONLY,  # 読み取り専用
        "approval_policy": ApprovalMode.NEVER,  # 自動実行
    })

    # レビュープロンプト
    prompt = f"""
    以下のファイルをレビューしてください: {file_path}

    以下の観点でチェックしてください：
    1. コードの品質と可読性
    2. 潜在的なバグ
    3. セキュリティ上の懸念
    4. パフォーマンスの問題
    5. ベストプラクティスへの準拠
    """

    # 構造化出力でレビュー結果を取得
    turn = await thread.run(
        prompt,
        {"output_schema": ReviewResult.model_json_schema()}
    )

    # 結果をパース
    result = ReviewResult.model_validate_json(turn.final_response)

    # レポートを出力
    print("=" * 50)
    print(f"コードレビュー結果: {file_path}")
    print("=" * 50)
    print(f"\nスコア: {result.overall_score}/10")
    print(f"判定: {result.verdict}")

    if result.issues:
        print("\n問題点:")
        for issue in result.issues:
            print(f"  - {issue}")

    if result.security_concerns:
        print("\nセキュリティ上の懸念:")
        for concern in result.security_concerns:
            print(f"  - {concern}")

    if result.suggestions:
        print("\n改善提案:")
        for suggestion in result.suggestions:
            print(f"  - {suggestion}")

    return result

# 実行
if __name__ == "__main__":
    asyncio.run(code_review_bot("./src/main.py"))

まとめ

この記事では、Python版Codex SDKの使い方を解説しました。

学んだこと

  1. 基本的な使い方: Codex → Thread → Turn の流れ
  2. ストリーミング: run_streamed()でリアルタイム処理
  3. 構造化出力: JSONスキーマ/Pydanticで型安全な出力
  4. 画像添付: 視覚情報をCodexに渡す
  5. サンドボックス: ファイルアクセスの制御
  6. 承認ポリシー: アクション実行の制御
  7. スレッド再開: セッションの永続化

リポジトリ

Python版Codex SDKは以下のリポジトリで公開しています：

pip install codex-sdk-py

GitHub: https://github.com/nogataka/codex-sdk-py

参考リンク

3
4
0

Register as a new user and use Qiita more conveniently

  1. You get articles that match your needs
  2. You can efficiently read back useful information
  3. You can use dark theme
What you can do with signing up
Sign upLogin
3
4

Delete article

Deleted articles cannot be recovered.

Draft of this article would be also deleted.

Are you sure you want to delete this article?