SSH Integration

Last updated: 2026-05-07 06:53:50

SSH Integration¶

The SSH integration enables pack commands to execute shell commands on user-registered SSH targets. Users register their servers (GPU clusters, custom hardware, on-premise machines), and pack authors orchestrate work on them through ctx.ssh.run().

Quick Start¶

from huitzo_sdk import command, Context

@command("check-gpu", namespace="compute")
async def check_gpu(args: Args, ctx: Context) -> dict:
    """Check GPU status on user's cluster."""
    result = await ctx.ssh.run("gpu-cluster", "nvidia-smi --query-gpu=name,memory.used --format=csv")
    return {
        "output": result.stdout,
        "exit_code": result.exit_code,
    }

ctx.ssh.run()¶

Execute a shell command on a named SSH target.

result = await ctx.ssh.run(
    target: str,           # Registered target name (e.g., "gpu-cluster")
    command: str,          # Shell command to execute
    *,
    timeout: int = 30,     # Timeout in seconds
) -> SSHResult

Parameters¶

Parameter	Type	Required	Default	Description
`target`	`str`	Yes	—	Human-readable name of the registered SSH target
`command`	`str`	Yes	—	Shell command to execute on the remote host
`timeout`	`int`	No	`30`	Maximum execution time in seconds

Return Value: SSHResult¶

@dataclass(frozen=True)
class SSHResult:
    stdout: str       # Command standard output (truncated at 1 MB)
    stderr: str       # Command standard error (truncated at 1 MB)
    exit_code: int    # Process exit code (0 = success)

Output truncation: If stdout or stderr exceeds 1 MB (1,048,576 bytes), the output is truncated and a notice is appended to stderr: "\n[OUTPUT TRUNCATED: exceeded 1MB limit]".

Examples¶

Basic Command Execution¶

@command("disk-check", namespace="infra")
async def disk_check(args: Args, ctx: Context) -> dict:
    result = await ctx.ssh.run("web-server", "df -h /")
    return {"disk_usage": result.stdout}

Checking Exit Code¶

@command("health-check", namespace="infra")
async def health_check(args: Args, ctx: Context) -> dict:
    result = await ctx.ssh.run("web-server", "systemctl is-active nginx")

    if result.exit_code == 0:
        return {"status": "healthy", "service": "nginx"}
    else:
        return {
            "status": "unhealthy",
            "service": "nginx",
            "error": result.stderr,
        }

Long-Running Command with Custom Timeout¶

@command("train-model", namespace="ml", timeout=3600)
async def train_model(args: TrainArgs, ctx: Context) -> dict:
    """Start model training on GPU cluster."""
    result = await ctx.ssh.run(
        "gpu-cluster",
        f"python /opt/ml/train.py --epochs {args.epochs} --batch-size {args.batch_size}",
        timeout=3600,
    )

    if result.exit_code != 0:
        return {"status": "failed", "error": result.stderr}

    return {"status": "completed", "output": result.stdout}

Multiple Targets¶

@command("cluster-status", namespace="infra")
async def cluster_status(args: Args, ctx: Context) -> dict:
    """Check status across multiple servers."""
    targets = ["node-1", "node-2", "node-3"]
    results = {}

    for target in targets:
        try:
            result = await ctx.ssh.run(target, "uptime")
            results[target] = {"status": "up", "uptime": result.stdout.strip()}
        except SSHError as e:
            results[target] = {"status": "down", "error": e.message}

    return {"cluster": results}

Error Handling¶

All SSH failures raise SSHError:

from huitzo_sdk.errors import SSHError

@command("run-task", namespace="compute")
async def run_task(args: Args, ctx: Context) -> dict:
    try:
        result = await ctx.ssh.run("gpu-cluster", args.command, timeout=60)
        return {"output": result.stdout, "exit_code": result.exit_code}
    except SSHError as e:
        ctx.log.error(f"SSH failed on {e.target}: {e.message}")
        return {"error": e.message, "target": e.target}

SSHError Attributes¶

Attribute	Type	Description
`host`	`str`	Hostname or IP (may be empty if target not resolved)
`target`	`str`	Target name that was requested
`message`	`str`	Human-readable error description

Common Error Scenarios¶

Scenario	Raised When
Target not allowed	Pack manifest doesn't include target in `ssh_targets.allowed`
Target not found	User hasn't registered a target with that name
No host key	Target missing `known_hosts` entry (needs verification)
Connection failed	Host unreachable, auth rejected, or network error
Command timeout	Execution exceeded the specified timeout
Command too long	Command string exceeds 4,096 bytes
Null bytes	Command contains null bytes (`\0`)

Pack Manifest Configuration¶

Packs must declare which SSH targets they can access in huitzo.yaml:

ssh_targets:
  allowed:
    - "*"                    # Allow all user-registered targets

Or restrict to specific targets:

ssh_targets:
  allowed:
    - "gpu-cluster"
    - "preprocessing-server"

If ssh_targets is omitted or allowed is empty, the pack cannot use ctx.ssh.

See Pack Manifest - SSH Targets for complete configuration reference.

Security Model¶

What Pack Authors Should Know¶

No credential handling — You never see SSH keys or passwords. The platform injects them.
Target allowlist — Your pack can only access targets declared in huitzo.yaml.
No shell injection risk — Commands are sent via SSH exec channel, not through a shell.
Output limits — stdout/stderr are truncated at 1 MB to prevent memory issues.
Timeout enforcement — Commands are killed if they exceed the timeout.

What Users Should Know¶

You control your servers — Only you can register SSH targets for your account.
Host key verification — Huitzo verifies the server identity on every connection.
Credentials encrypted — SSH keys and passwords are Fernet-encrypted at rest.
Pack permissions — Only packs that declare ssh_targets in their manifest can use SSH.

See SSH Execution Architecture for the full security model.

Best Practices¶

1. Always Handle Errors¶

# Good: Handle SSH errors gracefully
try:
    result = await ctx.ssh.run("server", "command")
except SSHError as e:
    ctx.log.error(f"SSH failed: {e.message}")
    return {"error": "Remote execution failed", "details": e.message}

2. Check Exit Codes¶

# Good: Non-zero exit code doesn't auto-raise
result = await ctx.ssh.run("server", "grep pattern /var/log/app.log")
if result.exit_code != 0:
    # grep returns 1 when no match found — not necessarily an error
    return {"matches": []}

3. Set Appropriate Timeouts¶

# Good: Match timeout to expected duration
result = await ctx.ssh.run("server", "quick-check", timeout=5)

# Good: Long timeout for heavy operations
result = await ctx.ssh.run("gpu", "python train.py", timeout=3600)

4. Keep Commands Simple¶

# Good: Single, clear command
result = await ctx.ssh.run("server", "nvidia-smi --query-gpu=name --format=csv,noheader")

# Avoid: Complex shell pipelines (harder to debug)
# result = await ctx.ssh.run("server", "ps aux | grep python | awk '{print $2}' | xargs kill")