コンテンツにスキップ

プロセス

class Process()

サンドボックス内でのプロセスおよびコード実行を扱います。

Process.__init__

def __init__(sandbox_id: str, code_toolbox: SandboxPythonCodeToolbox,
             toolbox_api: ToolboxApi, get_root_dir: Callable[[], str])

新しい Process インスタンスを初期化します。

引数:

  • sandbox_id str - サンドボックスの ID。
  • code_toolbox SandboxPythonCodeToolbox - 言語固有のコード実行ツールボックス。
  • toolbox_api ToolboxApi - サンドボックス操作用の API クライアント。
  • get_root_dir Callable[[], str] - サンドボックスの既定のルートディレクトリを取得する関数。

Process.exec

@intercept_errors(message_prefix="Failed to execute command: ")
def exec(command: str,
         cwd: Optional[str] = None,
         env: Optional[Dict[str, str]] = None,
         timeout: Optional[int] = None) -> ExecuteResponse

サンドボックスでシェルコマンドを実行します。

引数:

  • command str - 実行するシェルコマンド。
  • cwd Optional[str] - コマンド実行時の作業ディレクトリ。指定しない場合はサンドボックスのルートディレクトリを使用します。既定はユーザーのルートディレクトリです。
  • env Optional[Dict[str, str]] - コマンドに設定する環境変数。
  • timeout Optional[int] - コマンドの完了を待機する最大時間（秒）。0 は無期限待機を意味します。

戻り値:

  • ExecuteResponse - コマンド実行結果を含みます:
    • exit_code: コマンドの終了ステータス
    • result: コマンドの標準出力
    • artifacts: stdout（result と同じ）および charts（matplotlib のチャートのメタデータ）を含む ExecutionArtifacts オブジェクト

例:

# シンプルなコマンド
response = sandbox.process.exec("echo 'Hello'")
print(response.artifacts.stdout)  # 出力: Hello


# 作業ディレクトリを指定したコマンド
result = sandbox.process.exec("ls", cwd="workspace/src")


# タイムアウト付きのコマンド
result = sandbox.process.exec("sleep 10", timeout=5)

Process.code_run

def code_run(code: str,
             params: Optional[CodeRunParams] = None,
             timeout: Optional[int] = None) -> ExecuteResponse

適切な言語ランタイムを用いてサンドボックスでコードを実行します。

引数:

  • code str - 実行するコード。
  • params Optional[CodeRunParams] - コード実行のパラメータ。
  • timeout Optional[int] - コードの完了を待機する最大時間（秒）。0 は無期限待機を意味します。

戻り値:

  • ExecuteResponse - コード実行結果を含みます:
    • exit_code: 実行の終了ステータス
    • result: コードの標準出力
    • artifacts: stdout（result と同じ）および charts（matplotlib のチャートのメタデータ）を含む ExecutionArtifacts オブジェクト

例:

# Python コードを実行
response = sandbox.process.code_run('''
    x = 10
    y = 20
    print(f"Sum: {x + y}")
''')
print(response.artifacts.stdout)  # 出力: Sum: 30

Matplotlib のチャートは自動検出され、ExecutionArtifacts オブジェクトの charts フィールドで返されます。

code = '''
import matplotlib.pyplot as plt
import numpy as np


x = np.linspace(0, 10, 30)
y = np.sin(x)


plt.figure(figsize=(8, 5))
plt.plot(x, y, 'b-', linewidth=2)
plt.title('Line Chart')
plt.xlabel('X-axis (seconds)')
plt.ylabel('Y-axis (amplitude)')
plt.grid(True)
plt.show()
'''


response = sandbox.process.code_run(code)
chart = response.artifacts.charts[0]


print(f"Type: {chart.type}")
print(f"Title: {chart.title}")
if chart.type == ChartType.LINE and isinstance(chart, LineChart):
    print(f"X Label: {chart.x_label}")
    print(f"Y Label: {chart.y_label}")
    print(f"X Ticks: {chart.x_ticks}")
    print(f"X Tick Labels: {chart.x_tick_labels}")
    print(f"X Scale: {chart.x_scale}")
    print(f"Y Ticks: {chart.y_ticks}")
    print(f"Y Tick Labels: {chart.y_tick_labels}")
    print(f"Y Scale: {chart.y_scale}")
    print("Elements:")
    for element in chart.elements:
        print(f"Label: {element.label}")
        print(f"Points: {element.points}")

Process.create_session

@intercept_errors(message_prefix="Failed to create session: ")
def create_session(session_id: str) -> None

サンドボックスに新しい常駐型のバックグラウンドセッションを作成します。

セッションはバックグラウンドプロセスで、コマンド間の状態を維持します。関連する複数のコマンド実行や環境の恒常的なセットアップが必要なシナリオに最適です。長時間実行のコマンドを走らせ、プロセスの状態を監視できます。

引数:

  • session_id str - 新規セッションの一意な識別子。

例:

# Create a new session
session_id = "my-session"
sandbox.process.create_session(session_id)
session = sandbox.process.get_session(session_id)
# Do work...
sandbox.process.delete_session(session_id)

Process.get_session

@intercept_errors(message_prefix="Failed to get session: ")
def get_session(session_id: str) -> Session

サンドボックス内のセッションを取得します。

引数:

  • session_id str - 取得するセッションの一意な識別子。

戻り値:

  • Session - 次の情報を含むセッション情報:
    • session_id: セッションの一意な識別子
    • commands: セッションで実行されたコマンドの一覧

例:

session = sandbox.process.get_session("my-session")
for cmd in session.commands:
    print(f"Command: {cmd.command}")

Process.get_session_command

@intercept_errors(message_prefix="Failed to get session command: ")
def get_session_command(session_id: str, command_id: str) -> Command

セッション内で実行された特定のコマンドに関する情報を取得します。

引数:

  • session_id str - セッションの一意な識別子。
  • command_id str - コマンドの一意な識別子。

戻り値:

  • Command - 次の情報を含むコマンド情報:
    • id: コマンドの一意な識別子
    • command: 実行されたコマンド文字列
    • exit_code: 終了ステータス（完了している場合）

例:

cmd = sandbox.process.get_session_command("my-session", "cmd-123")
if cmd.exit_code == 0:
    print(f"Command {cmd.command} completed successfully")

Process.execute_session_command

@intercept_errors(message_prefix="Failed to execute session command: ")
def execute_session_command(
        session_id: str,
        req: SessionExecuteRequest,
        timeout: Optional[int] = None) -> SessionExecuteResponse

セッション内でコマンドを実行します。

引数:

  • session_id str - 使用するセッションの一意な識別子。
  • req SessionExecuteRequest - 次を含むコマンド実行リクエスト:
    • command: 実行するコマンド
    • run_async: 非同期で実行するかどうか

戻り値:

  • SessionExecuteResponse - 次を含むコマンド実行結果:
    • cmd_id: 実行されたコマンドの一意な識別子
    • output: コマンドの出力（同期実行の場合）
    • exit_code: 終了ステータス（同期実行の場合）

例:

# Execute commands in sequence, maintaining state
session_id = "my-session"


# Change directory
req = SessionExecuteRequest(command="cd /workspace")
sandbox.process.execute_session_command(session_id, req)


# Create a file
req = SessionExecuteRequest(command="echo 'Hello' > test.txt")
sandbox.process.execute_session_command(session_id, req)


# Read the file
req = SessionExecuteRequest(command="cat test.txt")
result = sandbox.process.execute_session_command(session_id, req)
print(result.output)  # Prints: Hello

Process.get_session_command_logs

@intercept_errors(message_prefix="Failed to get session command logs: ")
def get_session_command_logs(session_id: str, command_id: str) -> str

セッションで実行したコマンドのログを取得します。セッション内で実行したコマンドの出力（stdout と stderr）を完全に取得します。

引数:

  • session_id str - セッションの一意な識別子。
  • command_id str - コマンドの一意な識別子。

戻り値:

  • str - stdout と stderr の両方を含む完全なコマンド出力。

例:

logs = sandbox.process.get_session_command_logs(
    "my-session",
    "cmd-123"
)
print(f"Command output: {logs}")

Process.get_session_command_logs_async

@intercept_errors(message_prefix="Failed to get session command logs: ")
async def get_session_command_logs_async(
        session_id: str, command_id: str, on_logs: Callable[[str],
                                                            None]) -> None

セッションで実行されたコマンドのログを、利用可能になり次第、非同期で取得して処理します。

引数:

  • session_id str - セッションの一意の識別子。
  • command_id str - コマンドの一意の識別子。
  • on_logs Callable[[str], None] - 受信したログチャンクを処理するコールバック関数。

例:

await sandbox.process.get_session_command_logs_async(
    "my-session",
    "cmd-123",
    lambda chunk: print(f"Log chunk: {chunk}")
)

Process.list_sessions

@intercept_errors(message_prefix="Failed to list sessions: ")
def list_sessions() -> List[Session]

サンドボックス内のすべてのセッションを一覧表示します。

戻り値:

  • List[Session] - サンドボックス内のすべてのセッションのリスト。

例:

sessions = sandbox.process.list_sessions()
for session in sessions:
    print(f"Session {session.session_id}:")
    print(f"  Commands: {len(session.commands)}")

Process.delete_session

@intercept_errors(message_prefix="Failed to delete session: ")
def delete_session(session_id: str) -> None

サンドボックス内のセッションを終了して削除し、関連リソースをクリーンアップします。

引数:

  • session_id str - 削除するセッションの一意の識別子。

例:

# Create and use a session
sandbox.process.create_session("temp-session")
# ... use the session ...


# Clean up when done
sandbox.process.delete_session("temp-session")

CodeRunParams

@dataclass
class CodeRunParams()

コード実行用のパラメータ。

属性:

  • argv Optional[List[str]] - コマンドライン引数
  • env Optional[Dict[str, str]] - 環境変数

SessionExecuteRequest

class SessionExecuteRequest(ApiSessionExecuteRequest,
                            AsyncApiSessionExecuteRequest)

セッションでコマンドを実行するためのリクエストです。

属性:

  • command str - 実行するコマンド。
  • run_async Optional[bool] - コマンドを非同期で実行するかどうか。
  • var_async Optional[bool] - 非推奨。代わりに run_async を使用してください。

実行成果物

class ExecutionArtifacts()

コマンド実行で得られる成果物。

属性:

  • stdout str - コマンドの標準出力。ExecuteResponse の result と同一
  • charts Optional[List[Chart]] - matplotlib のチャートメタデータのリスト

ExecuteResponse

class ExecuteResponse(ClientExecuteResponse)

コマンド実行のレスポンス。

属性:

  • exit_code int - コマンド実行の終了コード
  • result str - コマンド実行の出力
  • artifacts Optional[ExecutionArtifacts] - コマンド実行の実行成果物