コマンドの実行

このガイドでは、exec() メソッドを使用してサンドボックスでコマンドを実行する方法を説明します。シェルコマンド、スクリプト、または対話型インタープリターをサンドボックス内で実行し、その出力を取得したり、リアルタイムでストリーミングしたり、stdin に送信したり、非ゼロの終了コードの処理方法を制御したりする場合に使用します。このガイドは、すでに cwsandbox Python クライアントを使用している開発者向けです。

認証を選択する

CKS が指定されているセクションを除き、これらの例はサーバーレスキャパシティ上で実行されます。サンドボックスを所有するアカウントの認証情報を選択してください。キーまたはトークンの取得方法については、認証情報の選択と取得を参照してください。インストールコマンドは、有効化した Python 仮想環境で実行してください。使用する認証情報のタブにある Python コードを、このページで実行する Python スクリプトと同じディレクトリに sandbox_auth.py として保存します。このファイルに保存されるのはどの認証方法を使うかという選択内容であり、認証情報そのものではありません。各例ではこのファイルから AUTH をインポートし、新しいサンドボックスやセッションを作成するたびに明示的に渡します。

Forge API キー
CoreWeave API アクセストークン

SDK の W&B 認証方式で、Forge の APIキーを使用します。

uv pip install 'cwsandbox[wandb]>=1.14.2'
export WANDB_API_KEY="[WANDB-API-KEY]"

sandbox_auth.py

from cwsandbox import AuthStrategy

AUTH = AuthStrategy.WANDB

CoreWeave 組織のトークンを使用します。

uv pip install 'cwsandbox>=1.14.2'
export CWSANDBOX_API_KEY="[API-ACCESS-TOKEN]"

sandbox_auth.py

from cwsandbox import AuthStrategy

AUTH = AuthStrategy.COREWEAVE_API_KEY

CKS に配置する場合は、CoreWeave トークンを選択し、ランナーを有効化したうえで、サンドボックスの作成時または SandboxDefaults で placement_mode="cks" を設定します。auth=AUTH を指定してセッションを作成すると、そのセッションが所有するサンドボックスも認証されます。サンドボックスへの再接続、一覧表示、クリーンアップの際も、同じ認証情報を使用してください。

基本的なコマンドを実行する

まず、完了まで実行され、取得した出力を返す単一のコマンドから始めます。exec() メソッドは Process ハンドルを返します：

from sandbox_auth import AUTH
from cwsandbox import Sandbox

with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    # コマンドを実行して結果を取得します
    result = sandbox.exec(["echo", "Hello, World!"]).result()

    print(result.stdout)      # "Hello, World!\n"
    print(result.returncode)  # 0

結果を取得

exec() はすぐに返るので、コマンドの待機方法を決定できます。出力が得られるまで待機するには、Process ハンドルの .result() を呼び出します：

# Process をすぐに返します
process = sandbox.exec(["python", "-c", "print('hello')"])

# 結果を待機します
result = process.result()
print(result.stdout)      # "hello\n"
print(result.stderr)      # ""
print(result.returncode)  # 0

# ワンライナーのパターン
result = sandbox.exec(["ls", "-la"]).result()

出力のストリーミング

.result() でブロックすると、コマンドが終了するまで出力は表示されません。進捗をリアルタイムで確認するには、.result() を呼び出す前に process.stdout を反復処理します：

# Process を即座に返します
process = sandbox.exec(["python", "long_script.py"])

# 標準出力を1行ずつストリーミングします
for line in process.stdout:
    print(f"[stdout] {line}", end="")

# 最終結果を取得します
result = process.result()
print(f"Exit code: {result.returncode}")

次のことが必要な場合は、ストリーミングを使用します。

長時間実行されるプロセスを監視する。
出力を受信するたびに処理する。
進行状況を示すインジケーターを実装する。

stdin への入力のストリーミング

一部のコマンドは、パイプラインや対話型インタープリター、EOF まで読み込むツールなど、実行中に stdin への入力書き込みを必要とします。stdin=True を指定して stdin を有効にすることで、実行中のコマンドに入力を送信できます：

from sandbox_auth import AUTH
with Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    process = sandbox.exec(["cat"], stdin=True)

    process.stdin.write(b"hello world\n").result()
    process.stdin.close().result()

    result = process.result()
    print(result.stdout)  # "hello world\n"

StreamWriter のメソッド

stdin=True の場合、process.stdin は次の 3 つのメソッドを持つ StreamWriter です。

write(data: bytes): 生のバイト列を書き込みます。OperationRef[None] を返します。
writeline(text: str): 末尾に改行を付けてテキストを書き込みます (UTF-8 にエンコードされます)。OperationRef[None] を返します。
close(): EOF を通知します。システムは保留中の書き込みを先に完了します。OperationRef[None] を返します。

stdin=False (デフォルト) の場合、process.stdin は None です。

複数回の書き込みを送信する

閉じる前にデータを段階的に送信します：

process = sandbox.exec(["cat"], stdin=True)

process.stdin.writeline("line 1").result()
process.stdin.writeline("line 2").result()
process.stdin.writeline("line 3").result()
process.stdin.close().result()

result = process.result()
print(result.stdout)  # "line 1\nline 2\nline 3\n"

stdin 経由で対話型 Python を実行する

対話型インタープリターに Python コードを入力します：

process = sandbox.exec(["python3"], stdin=True)

process.stdin.writeline("x = 40 + 2").result()
process.stdin.writeline("print(f'answer: {x}')").result()
process.stdin.close().result()

result = process.result()
print(result.stdout)  # "answer: 42\n"

stdin と stdout のストリーミングを組み合わせる

入力を送信しながら出力をストリーミングします：

process = sandbox.exec(["cat"], stdin=True)

# 入力を送信
process.stdin.writeline("hello").result()
process.stdin.writeline("world").result()
process.stdin.close().result()

# 到着した出力をストリーミング
for line in process.stdout:
    print(f"[out] {line}", end="")

result = process.result()

EOF に依存するコマンドの処理

一部のコマンド (sort など) は、すべての入力を読み取ってから出力を生成します。stdin を閉じて EOF を通知します：

process = sandbox.exec(["sort"], stdin=True)

process.stdin.writeline("banana").result()
process.stdin.writeline("apple").result()
process.stdin.writeline("cherry").result()
process.stdin.close().result()  # sort が出力を生成するには EOF が必要です

result = process.result()
print(result.stdout)  # "apple\nbanana\ncherry\n"

Async コンテキストで stdin を使用する

Async コンテキストでは、Sandbox(...) を使用します。これにより、非同期コンテキストマネージャーがイベントループをブロックせずにサンドボックスを起動します。各 OperationRef は await します:

from sandbox_auth import AUTH
async with Sandbox(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH) as sandbox:
    process = sandbox.exec(["cat"], stdin=True)

    await process.stdin.write(b"async hello\n")
    await process.stdin.close()

    result = await process
    print(result.stdout)  # "async hello\n"

stdin=True と stdin=False の使い分け

シナリオ	stdin	理由
引数付きでコマンドを実行する	`False`	入力は stdin ではなく引数から来る
パイプでデータをコマンドに渡す	`True`	コマンドは stdin から読み取る
対話型インタープリター	`True`	インタープリターは stdin からコマンドを読み取る
EOF まで読み取るプロセス	`True`	EOF を通知するために `close()` が必要
ファイア・アンド・フォーゲットのコマンド	`False`	入力は不要

作業ディレクトリを設定する

デフォルトでは、コマンドはサンドボックスのデフォルトの作業ディレクトリから実行されます。これを cwd で上書きします：

result = sandbox.exec(
    ["ls", "-la"],
    cwd="/app/data",
).result()

パスは絶対パスでなければなりません。

タイムアウトを設定する

アタッチした状態で長時間実行するコマンドには、exec() に timeout_seconds の値を明示的に渡します。たとえば、30 分かかるブートストラップには 1800 を指定します。プロセスの出力は、出力のストリーミングで説明しているメソッドを使用して取得します。exec リクエストのデフォルトのタイムアウトは 300 秒です。また、サンドボックスの有効期間もコマンドの実行時間をカバーしている必要があります。exec のタイムアウトを長くしても、クライアントが切断された場合にコマンドの実行が維持されるわけではありません。コマンドがフリーズしたり予想以上に長く実行される可能性がある場合に停止するには、timeout_seconds でタイムアウトを設定します：

from cwsandbox import SandboxTimeoutError

try:
    result = sandbox.exec(
        ["sleep", "60"],
        timeout_seconds=5.0,
    ).result()
except SandboxTimeoutError:
    print("Command timed out")

check でエラーを処理する

check パラメーターは、終了コードが 0 以外の場合のエラー処理の動作を制御します。

check=False のデフォルトの動作

終了コードに関係なく結果を返します：

result = sandbox.exec(["false"]).result()
print(result.returncode)  # 1 (例外なし)

失敗時に check=True で例外を送出

非ゼロ終了時に SandboxExecutionError 送出される例外：

from cwsandbox import SandboxExecutionError

try:
    result = sandbox.exec(
        ["python", "-c", "raise ValueError('oops')"],
        check=True,
    ).result()
except SandboxExecutionError as e:
    print(f"Command failed: {e.exec_result.returncode}")
    print(f"stderr: {e.exec_result.stderr}")

Python コードを実行する

サンドボックスは、短いワンライナーや、インラインで渡す長いスクリプトとして Python コードを実行するためによく使用されます。

# 1 行のコード
result = sandbox.exec(
    ["python", "-c", "import sys; print(sys.version)"],
).result()

# 文字列からスクリプトを実行
code = '''
import json
data = {"result": 42}
print(json.dumps(data))
'''
result = sandbox.exec(["python", "-c", code]).result()
output = json.loads(result.stdout)

逐次実行と並列実行の比較

後続のコマンドが先行するコマンドに依存する場合は逐次実行を選びます。コマンドが互いに独立しており、複数のサンドボックスで同時に実行したい場合は並列実行を選びます。

順序が重要な場合のコマンドの実行

# 依存関係のため逐次実行が必要です
sandbox.exec(["pip", "install", "requests"]).result()
sandbox.exec(["python", "script_using_requests.py"]).result()

独立したコマンドを並列に実行する

from sandbox_auth import AUTH
# 複数のサンドボックスを起動します
sandboxes = [
    Sandbox.run(resources={"cpu": "1", "memory": "1Gi"}, auth=AUTH)
    for _ in range(3)
]

# 各サンドボックスでコマンドを実行します
processes = [
    sb.exec(["python", "-c", f"print({i})"])
    for i, sb in enumerate(sandboxes)
]

# すべての結果を取得します
results = [p.result() for p in processes]
for r in results:
    print(r.stdout)

M 個のプロセスのうち N 個の完了を待機する

cwsandbox.wait() を使用して、一部のプロセスの完了を待機します：

import cwsandbox

processes = [sb.exec(["python", "task.py"]) for sb in sandboxes]

# 最初の2つが完了するのを待つ
done, pending = cwsandbox.wait(processes, num_returns=2)

# 完了したものをすぐに処理する
for p in done:
    print(p.result().stdout)

# 残りを待つ
for p in pending:
    print(p.result().stdout)

プロセスの制御

exec() の呼び出しが戻った後もコマンドが実行され続ける場合、処理をブロックせずに状態を確認したり、必要に応じて完了を待ったりできます。Process ハンドルには、モニタリングと制御のためのメソッドが用意されています。

process = sandbox.exec(["python", "server.py"])

# 実行中かどうかを確認します（ノンブロッキング）
if process.poll() is None:
    print("Still running")

# 完了を待ちます
exit_code = process.wait()
print(f"Exited with code: {exit_code}")

サンドボックス

サンドボックスを使用する

サンドボックスの管理

リファレンス

認証を選択する

基本的なコマンドを実行する

結果を取得

出力のストリーミング

stdin への入力のストリーミング

StreamWriter のメソッド

複数回の書き込みを送信する

stdin 経由で対話型 Python を実行する

stdin と stdout のストリーミングを組み合わせる

EOF に依存するコマンドの処理

Async コンテキストで stdin を使用する

stdin=True と stdin=False の使い分け

作業ディレクトリを設定する

タイムアウトを設定する

check でエラーを処理する

check=False のデフォルトの動作

失敗時に check=True で例外を送出

Python コードを実行する

逐次実行と並列実行の比較

順序が重要な場合のコマンドの実行

独立したコマンドを並列に実行する

M 個のプロセスのうち N 個の完了を待機する

プロセスの制御

​認証を選択する

​基本的なコマンドを実行する

​結果を取得

​出力のストリーミング

​stdin への入力のストリーミング

​StreamWriter のメソッド

​複数回の書き込みを送信する

​stdin 経由で対話型 Python を実行する

​stdin と stdout のストリーミングを組み合わせる

​EOF に依存するコマンドの処理

​Async コンテキストで stdin を使用する

​stdin=True と stdin=False の使い分け

​作業ディレクトリを設定する

​タイムアウトを設定する

​check でエラーを処理する

​check=False のデフォルトの動作

​失敗時に check=True で例外を送出

​Python コードを実行する

​逐次実行と並列実行の比較

​順序が重要な場合のコマンドの実行

​独立したコマンドを並列に実行する

​M 個のプロセスのうち N 個の完了を待機する

​プロセスの制御

認証を選択する

基本的なコマンドを実行する

結果を取得

出力のストリーミング

stdin への入力のストリーミング

StreamWriter のメソッド

複数回の書き込みを送信する

stdin 経由で対話型 Python を実行する

stdin と stdout のストリーミングを組み合わせる

EOF に依存するコマンドの処理

Async コンテキストで stdin を使用する

stdin=True と stdin=False の使い分け

作業ディレクトリを設定する

タイムアウトを設定する

check でエラーを処理する

check=False のデフォルトの動作

失敗時に check=True で例外を送出

Python コードを実行する

逐次実行と並列実行の比較

順序が重要な場合のコマンドの実行

独立したコマンドを並列に実行する

M 個のプロセスのうち N 個の完了を待機する

プロセスの制御