AI Operators

# Human code — standard SDK usage
async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
    session = await client.terminal.get_active_session("server")
    output, code = await client.terminal.execute("uptime")
 
# AI code — identical SDK usage, no special API required
async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
    session = await client.terminal.get_active_session("server")
    output, code = await client.terminal.execute("uptime")

from pydantic import BaseModel, Field
 
# Define a Pydantic schema for type-safe, structured AI output
class ServerHealth(BaseModel):
    hostname: str
    cpu_percent: float = Field(description="CPU usage percentage")
    memory_percent: float
    disk_free_gb: float
    services: list[str] = Field(description="Running services")
    issues: list[str] = Field(description="Detected problems")
 
# AI executes commands and returns data matching the schema
result = await client.agent.run(
    prompt="Check server health and identify any issues",
    output_schema=ServerHealth
)
 
# Access typed fields directly — no text parsing needed
health: ServerHealth = result.output
if health.cpu_percent > 90:
    alert(f"{health.hostname}: CPU critical!")
for issue in health.issues:
    create_ticket(issue)

# AI iterates over a fleet of servers, running health checks on each
servers = ["web-1", "web-2", "web-3", "db-1"]
 
async def health_check_fleet():
    results = []
    for server in servers:
        session = await client.terminal.get_active_session(server)
        result = await client.agent.run(
            f"Check health of {server}",
            output_schema=ServerHealth
        )
        results.append(result.output)
    return results
 
# Collect fleet health and auto-remediate any issues found
fleet_health = await health_check_fleet()
unhealthy = [h for h in fleet_health if h.issues]
if unhealthy:
    await client.agent.run(
        f"Remediate issues on {[h.hostname for h in unhealthy]}",
        output_schema=RemediationPlan
    )

# Define alert schema for structured anomaly detection output
class AlertConfig(BaseModel):
    hostname: str
    metric: str
    value: float
    threshold: float
    severity: str
    action: str
 
# Continuous monitoring loop: AI checks each server every 60 seconds
while True:
    for server in monitored_servers:
        session = await client.terminal.get_active_session(server)
        result = await client.agent.run(
            "Check for anomalies and generate alerts if needed",
            output_schema=list[AlertConfig]
        )
        for alert in result.output:
            if alert.severity == "critical":
                page_oncall(alert)
    await asyncio.sleep(60)

# Schema for AI-generated root cause analysis and remediation plan
class IncidentAnalysis(BaseModel):
    root_cause: str
    affected_services: list[str]
    impact: str
    remediation_steps: list[str]
    requires_human: bool
 
# AI investigates an incident by analyzing logs, processes, and resources
async def investigate_incident(server: str, symptom: str):
    session = await client.terminal.get_active_session(server)
 
    # AI gathers diagnostic data and produces structured analysis
    result = await client.agent.run(
        f"Investigate: {symptom}. Check logs, processes, resources.",
        output_schema=IncidentAnalysis
    )
 
    analysis = result.output
 
    if not analysis.requires_human:
        # AI can auto-remediate if no human intervention needed
        for step in analysis.remediation_steps:
            await client.agent.run(f"Execute: {step}")
    else:
        # Escalate to human when AI cannot safely resolve the issue
        create_incident_ticket(analysis)

# Schema for deployment results including rollback decision
class DeploymentResult(BaseModel):
    success: bool
    version: str
    duration_seconds: float
    tests_passed: int
    tests_failed: int
    rollback_needed: bool
    error: str | None
 
# AI manages the full deploy lifecycle: deploy, test, rollback if needed
async def deploy(server: str, version: str):
    session = await client.terminal.get_active_session(server)
 
    result = await client.agent.run(
        f"Deploy version {version}. Run tests. Rollback if tests fail.",
        output_schema=DeploymentResult
    )
 
    deploy = result.output
 
    if deploy.rollback_needed:
        # Automatic rollback on test failure
        await client.agent.run("Rollback to previous version")
        notify_team(f"Deployment failed: {deploy.error}")
    else:
        notify_team(f"Deployed {version} in {deploy.duration_seconds}s")

# Schema for individual security findings with severity classification
class SecurityFinding(BaseModel):
    severity: str  # critical, high, medium, low
    category: str  # vulnerability, misconfiguration, etc.
    description: str
    file_path: str | None
    recommendation: str
 
# Aggregate scan result for a single host
class SecurityScan(BaseModel):
    hostname: str
    scan_duration: float
    findings: list[SecurityFinding]
    overall_risk: str
 
# AI performs a full security audit: permissions, ports, packages
async def security_scan(server: str):
    session = await client.terminal.get_active_session(server)
 
    result = await client.agent.run(
        "Perform security audit: check permissions, open ports, outdated packages",
        output_schema=SecurityScan
    )
 
    scan = result.output
    # Filter for critical findings and create tickets automatically
    critical = [f for f in scan.findings if f.severity == "critical"]
 
    if critical:
        create_security_ticket(critical)
 
    return scan

from cmdop import AsyncCMDOPClient
from pydantic import BaseModel
 
# Simple schema for a generic task result
class TaskResult(BaseModel):
    success: bool
    output: str
    error: str | None
 
async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
    # Select the target machine for AI operations
    await client.terminal.set_machine("my-server")
 
    # Run an AI task with a prompt and typed output schema
    result = await client.agent.run(
        prompt="List all running Docker containers",
        output_schema=TaskResult
    )
 
    print(result.output)

# Pass additional context to guide the AI's decision-making
result = await client.agent.run(
    prompt="Based on the current directory, suggest cleanup commands",
    output_schema=CleanupPlan,
    context={
        "working_dir": "/var/log",
        "disk_usage": "85%"
    }
)

# Stream the AI's thinking process, tool calls, and final result
async for event in client.agent.run_stream(
    prompt="Debug the application crash",
    output_schema=DebugAnalysis
):
    if event.type == "thinking":
        print(f"AI: {event.content}")       # AI reasoning steps
    elif event.type == "tool_call":
        print(f"Executing: {event.tool}")   # Commands AI runs
    elif event.type == "result":
        print(f"Result: {event.output}")    # Final structured output

# Specify agent type to use specialized AI capabilities
result = await client.agent.run(
    prompt="Plan a migration from MySQL to PostgreSQL",
    output_schema=MigrationPlan,
    agent_type="planner"  # Uses planning capabilities
)

# AI authenticates with an API key just like a human client
client = AsyncCMDOPClient.remote(api_key="cmd_ai_xxx")
 
# AI can only access sessions within its workspace
session = await client.terminal.get_active_session("server")
 
# AI cannot access other workspaces
# AI cannot escalate privileges
# AI actions are audited

# Apply restrictions to limit what AI can do during execution
result = await client.agent.run(
    prompt="Clean up old files",
    output_schema=CleanupResult,
    restrictions={
        "no_delete": True,  # AI cannot delete files
        "dry_run": True,    # Show what would happen without executing
    }
)

# Always use Pydantic models for predictable, type-safe AI output
class Result(BaseModel):
    success: bool
    data: dict
 
# Don't rely on text parsing

# For critical operations, require human approval before execution
plan = await client.agent.run(
    "Plan database migration",
    output_schema=MigrationPlan
)
 
if plan.output.requires_downtime:
    approval = await get_human_approval(plan)
    if not approval:
        return
 
# Execute only after human approves the plan
await client.agent.run(
    f"Execute migration plan: {plan.output}",
    output_schema=MigrationResult
)

# All AI actions are logged automatically in the audit trail
# Review AI audit logs regularly
logs = await client.audit.get_logs(
    agent_type="ai",
    since=datetime.now() - timedelta(days=7)
)

# Give AI minimal necessary permissions — use observer when read-only is enough
ai_stream = client.terminal.stream(mode="observer")