コマンドの実行 - CoreWeave Docs

The examples on this page omit auth. To authenticate with your W&B API key, install cwsandbox[wandb] and pass auth=AuthStrategy.WANDB to Sandbox.run(), Sandbox(), or Sandbox.session(). See Get started.

This guide shows you how to run commands in sandboxes using the exec() method. Use it when you need to run shell commands, scripts, or interactive interpreters inside a sandbox and want to capture their output, stream it in real time, send input on stdin, or control how non-zero exit codes are handled. The guide is for developers who already work with the cwsandbox Python client.

Run a basic command

Start with a single command that runs to completion and returns its captured output. The exec() method returns a Process handle:

from cwsandbox import Sandbox

with Sandbox.run() as sandbox:
    # Run a command and get the result
    result = sandbox.exec(["echo", "Hello, World!"]).result()

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

Get results

exec() returns immediately so you can decide how to wait for the command. Call .result() on the Process handle to block for the output:

# Returns Process immediately
process = sandbox.exec(["python", "-c", "print('hello')"])

# Block for result
result = process.result()
print(result.stdout)      # "hello\n"
print(result.stderr)      # ""
print(result.returncode)  # 0

# One-liner pattern
result = sandbox.exec(["ls", "-la"]).result()

Stream output

Blocking on .result() waits for the command to finish before you see any output. To observe progress as it happens, iterate over process.stdout before calling .result():

# Returns Process immediately
process = sandbox.exec(["python", "long_script.py"])

# Stream stdout line by line
for line in process.stdout:
    print(f"[stdout] {line}", end="")

# Get final result
result = process.result()
print(f"Exit code: {result.returncode}")

Use streaming when you need to:

Monitor long-running processes.
Process output as it arrives.
Implement progress indicators.

Stream input to stdin

Some commands need input written to stdin while they run, such as pipelines, interactive interpreters, or tools that read until EOF. Send input to running commands by enabling stdin with stdin=True:

with Sandbox.run() 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 methods

When stdin=True, process.stdin is a StreamWriter with three methods:

write(data: bytes): Write raw bytes. Returns OperationRef[None].
writeline(text: str): Write text with a trailing newline (encodes to UTF-8). Returns OperationRef[None].
close(): Signal EOF. The system completes pending writes first. Returns OperationRef[None].

When stdin=False (the default), process.stdin is None.

Send multiple writes

Send data incrementally before closing:

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"

Run interactive Python through stdin

Feed Python code to an interactive interpreter:

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"

Combine stdin and stdout streaming

Stream output while sending input:

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

# Send input
process.stdin.writeline("hello").result()
process.stdin.writeline("world").result()
process.stdin.close().result()

# Stream output as it arrives
for line in process.stdout:
    print(f"[out] {line}", end="")

result = process.result()

Handle EOF-dependent commands

Some commands (like sort) read all input before producing output. Close stdin to signal 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 requires EOF before producing output

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

Use stdin in async contexts

In async contexts, await each OperationRef directly:

async with Sandbox.run() 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"

When to use stdin=True compared with stdin=False

Scenario	stdin	Reason
Run a command with arguments	`False`	Input comes from args, not stdin
Pipe data into a command	`True`	The command reads from stdin
Interactive interpreter	`True`	The interpreter reads commands from stdin
Process that reads until EOF	`True`	Requires `close()` to signal EOF
Fire-and-forget command	`False`	No input needed

Set the working directory

By default, commands run from the sandbox’s default working directory. Override that with cwd:

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

The path must be absolute.

Set a timeout

To stop a command that might freeze or run longer than you expect, set a timeout with timeout_seconds:

from cwsandbox import SandboxTimeoutError

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

Handle errors with check

The check parameter controls error behavior for non-zero exit codes:

Default behavior with check=False

Returns the result regardless of exit code:

result = sandbox.exec(["false"]).result()
print(result.returncode)  # 1 (no exception)

Raise on failure with check=True

Raises SandboxExecutionError on non-zero exit:

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}")

Run Python code

Sandboxes are commonly used to run Python code, either as short one-liners or as longer scripts passed inline:

# One-liner
result = sandbox.exec(
    ["python", "-c", "import sys; print(sys.version)"],
).result()

# Script from string
code = '''
import json
data = {"result": 42}
print(json.dumps(data))
'''
result = sandbox.exec(["python", "-c", code]).result()
output = json.loads(result.stdout)

Sequential compared with parallel execution

Choose sequential execution when later commands depend on earlier ones, and parallel execution when commands are independent and you want them to run concurrently across sandboxes.

Run commands sequentially when order matters

# Dependencies require sequential execution
sandbox.exec(["pip", "install", "requests"]).result()
sandbox.exec(["python", "script_using_requests.py"]).result()

Run independent commands in parallel

# Start multiple sandboxes
sandboxes = [Sandbox.run() for _ in range(3)]

# Start commands on each
processes = [
    sb.exec(["python", "-c", f"print({i})"])
    for i, sb in enumerate(sandboxes)
]

# Collect all results
results = [p.result() for p in processes]
for r in results:
    print(r.stdout)

Wait for N of M processes to complete

Use cwsandbox.wait() to wait for a subset of processes:

import cwsandbox

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

# Wait for first 2 to complete
done, pending = cwsandbox.wait(processes, num_returns=2)

# Process completed ones immediately
for p in done:
    print(p.result().stdout)

# Wait for remaining
for p in pending:
    print(p.result().stdout)

Control processes

When a command keeps running after the exec() call returns, you can check on it without blocking or wait for it to finish on your own terms. The Process handle provides methods for monitoring and control:

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

# Check if running (non-blocking)
if process.poll() is None:
    print("Still running")

# Wait for completion
exit_code = process.wait()
print(f"Exited with code: {exit_code}")

​Run a basic command

​Get results

​Stream output

​Stream input to stdin

​StreamWriter methods

​Send multiple writes

​Run interactive Python through stdin

​Combine stdin and stdout streaming

​Handle EOF-dependent commands

​Use stdin in async contexts

​When to use stdin=True compared with stdin=False

​Set the working directory

​Set a timeout

​Handle errors with check

​Default behavior with check=False

​Raise on failure with check=True

​Run Python code

​Sequential compared with parallel execution

​Run commands sequentially when order matters

​Run independent commands in parallel

​Wait for N of M processes to complete

​Control processes

Run a basic command

Get results

Stream output

Stream input to stdin

StreamWriter methods

Send multiple writes

Run interactive Python through stdin

Combine stdin and stdout streaming

Handle EOF-dependent commands

Use stdin in async contexts

When to use stdin=True compared with stdin=False

Set the working directory

Set a timeout

Handle errors with check

Default behavior with check=False

Raise on failure with check=True

Run Python code

Sequential compared with parallel execution

Run commands sequentially when order matters

Run independent commands in parallel

Wait for N of M processes to complete

Control processes