First Automation

# Install the Python SDK from PyPI
pip install cmdop

import asyncio
from cmdop import AsyncCMDOPClient
 
async def main():
    # Connect to Control Plane with your API key
    async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
        # Set the target machine by hostname
        await client.terminal.set_machine("my-server")
 
        # Execute a command and get output + exit code
        output, exit_code = await client.terminal.execute("uptime")
        print(f"Output: {output}")
        print(f"Exit code: {exit_code}")
 
asyncio.run(main())

Output: 14:23:01 up 42 days, 3:15, 2 users, load average: 0.15, 0.10, 0.08
Exit code: 0

# Create a personal API key via CLI
cmdop auth create-key --name "my-script"
 
# Copy the key: cmd_acme_xxx...

# Connect to CMDOP cloud via API key
client = AsyncCMDOPClient.remote(api_key="cmd_xxx")

# Connect to local Control Plane for development
client = AsyncCMDOPClient.local(
    host="localhost",
    port=50051
)

# Simple one-shot command execution
output, code = await client.terminal.execute("ls -la")
 
# With timeout (seconds) for long-running scripts
output, code = await client.terminal.execute(
    "long-running-script.sh",
    timeout=300  # 5-minute timeout
)
 
# Check exit code for error handling
if code != 0:
    print(f"Command failed with code {code}")

# Get the active session for a machine
session = await client.terminal.get_active_session("my-server")
 
# Create a bidirectional stream
stream = client.terminal.stream()
 
# Register output callback — prints data as it arrives
stream.on_output(lambda data: print(data.decode(), end=""))
 
# Attach stream to the existing session
await stream.attach(session.session_id)
 
# Send commands through the stream
await stream.send_input(b"cd /var/log\n")
await stream.send_input(b"tail -f app.log\n")
 
# Keep running for 60 seconds
await asyncio.sleep(60)
 
# Detach — session continues running on server
await stream.close()

async def monitor_logs():
    session = await client.terminal.get_active_session("my-server")
    stream = client.terminal.stream()
 
    # Async callback processes each chunk of output
    async def on_output(data: bytes):
        text = data.decode()
        if "ERROR" in text:
            send_alert(text)  # Trigger alert on errors
        print(text, end="")
 
    stream.on_output(on_output)
    await stream.attach(session.session_id)
 
    # Start tailing logs on remote machine
    await stream.send_input(b"tail -f /var/log/app.log\n")
 
    # Run indefinitely until interrupted
    await asyncio.Event().wait()

# Read a file from a remote machine
content = await client.files.read("my-server", "/etc/hostname")
print(content)

# Write content to a file on the remote machine
await client.files.write(
    "my-server",
    "/tmp/config.yaml",
    "key: value\n"
)

# List directory contents on remote machine
entries = await client.files.list("my-server", "/var/log")
for entry in entries:
    print(f"{entry.name} - {entry.size} bytes")

# Download a remote file to your local machine
await client.files.download(
    "my-server",
    "/var/log/app.log",
    "./app.log"
)
 
# Upload a local file to the remote machine
await client.files.upload(
    "./config.yaml",
    "my-server",
    "/app/config.yaml"
)

from pydantic import BaseModel
 
# Define a Pydantic model for typed AI output
class ServerHealth(BaseModel):
    hostname: str
    cpu_percent: float
    memory_percent: float
    disk_percent: float
    issues: list[str]
 
# AI analyzes the machine and returns typed data
result = await client.agent.run(
    prompt="Check server health and identify any issues",
    output_schema=ServerHealth
)
 
health: ServerHealth = result.output
 
if health.cpu_percent > 90:
    send_alert(f"High CPU on {health.hostname}")
 
for issue in health.issues:
    create_ticket(issue)

# Define expected output structure for deployment
class DeployResult(BaseModel):
    success: bool
    version: str
    duration_seconds: float
    error: str | None
 
# AI agent executes multi-step deployment autonomously
result = await client.agent.run(
    prompt="Deploy version 2.1.0. Run tests. Rollback if tests fail.",
    output_schema=DeployResult
)
 
if not result.output.success:
    notify_team(f"Deploy failed: {result.output.error}")

servers = ["web-1", "web-2", "web-3"]
 
# Deploy to each server one at a time
for server in servers:
    await client.terminal.set_machine(server)
    output, code = await client.terminal.execute("./deploy.sh")
    print(f"{server}: {'OK' if code == 0 else 'FAILED'}")

import asyncio
 
async def check_server(hostname: str) -> dict:
    """Check uptime on a single server."""
    await client.terminal.set_machine(hostname)
    output, code = await client.terminal.execute("uptime")
    return {"hostname": hostname, "uptime": output, "ok": code == 0}
 
# Run all checks concurrently with asyncio.gather
servers = ["web-1", "web-2", "web-3", "db-1"]
results = await asyncio.gather(*[check_server(s) for s in servers])
 
for r in results:
    print(f"{r['hostname']}: {r['uptime']}")

from cmdop.exceptions import (
    CMDOPError,
    AuthenticationError,
    SessionNotFoundError,
    ConnectionError,
    TimeoutError
)
 
try:
    output, code = await client.terminal.execute("command")
except AuthenticationError:
    print("Invalid API key")
except SessionNotFoundError:
    print("Machine not connected")
except TimeoutError:
    print("Command timed out")
except CMDOPError as e:
    print(f"CMDOP error: {e}")

import asyncio
from cmdop import AsyncCMDOPClient
from pydantic import BaseModel
 
# Typed output model for server status
class ServerStatus(BaseModel):
    hostname: str
    uptime: str
    load: str
    memory_free: str
 
async def main():
    async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
        servers = ["prod-1", "prod-2", "staging"]
 
        for server in servers:
            print(f"\n=== {server} ===")
 
            # Set target machine
            await client.terminal.set_machine(server)
 
            # AI agent gathers and structures server metrics
            result = await client.agent.run(
                "Get server status: uptime, load average, free memory",
                output_schema=ServerStatus
            )
 
            status = result.output
            print(f"Uptime: {status.uptime}")
            print(f"Load: {status.load}")
            print(f"Memory: {status.memory_free}")
 
if __name__ == "__main__":
    asyncio.run(main())