Configuration

# Connection
export CMDOP_API_KEY=cmd_xxx
export CMDOP_SERVER_ADDRESS=grpc.cmdop.com:443
export CMDOP_WORKSPACE=my-workspace
 
# Timeouts
export CMDOP_CONNECT_TIMEOUT=15.0
export CMDOP_REQUEST_TIMEOUT=60.0
 
# Retry
export CMDOP_RETRY_ATTEMPTS=5
export CMDOP_RETRY_TIMEOUT=30.0
 
# Circuit Breaker
export CMDOP_CIRCUIT_FAIL_MAX=5
export CMDOP_CIRCUIT_RESET_TIMEOUT=60.0
 
# Logging
export CMDOP_LOG_LEVEL=INFO
export CMDOP_LOG_JSON=true

from cmdop import get_settings, configure_settings
 
# Retrieve the current SDK settings instance
settings = get_settings()
print(f"Timeout: {settings.connect_timeout}")   # Current connection timeout value
print(f"Retries: {settings.retry_attempts}")     # Current max retry attempts
 
# Override settings programmatically (applies globally)
configure_settings(
    connect_timeout=15.0,   # Set connection timeout to 15 seconds
    retry_attempts=3,       # Limit retries to 3
    log_json=False,         # Disable JSON log format
)

from cmdop import CMDOPClient, ConnectionConfig, KeepaliveConfig, RetryConfig
 
config = ConnectionConfig(
    # Timeouts — control how long the client waits
    connect_timeout_seconds=15.0,    # Max wait for initial connection
    request_timeout_seconds=60.0,    # Max wait for a single request
    stream_timeout_seconds=0.0,      # 0 = unlimited (for long-running streams)
 
    # Keepalive — detect dead connections via periodic pings
    keepalive=KeepaliveConfig(
        time_ms=30_000,              # Send ping every 30 seconds
        timeout_ms=5_000,            # Wait 5 seconds for pong response
        permit_without_calls=True,   # Ping even when no active RPCs
    ),
 
    # Retry — automatic retry with exponential backoff
    retry=RetryConfig(
        max_attempts=3,              # Total attempts before giving up
        initial_backoff_seconds=1.0, # Delay before first retry
        max_backoff_seconds=30.0,    # Cap on backoff delay
        backoff_multiplier=2.0,      # Multiply delay each retry (1s, 2s, 4s...)
        jitter_fraction=0.1,         # Add +/- 10% randomness to avoid thundering herd
        retryable_codes=("UNAVAILABLE", "RESOURCE_EXHAUSTED"),  # gRPC codes to retry
    ),
 
    # Limits — constrain message sizes
    max_message_size_mb=50,          # Max gRPC message size in megabytes
)
 
# Create a remote client using the custom connection config
client = CMDOPClient.remote(api_key="cmd_xxx", config=config)

keepalive = KeepaliveConfig(
    time_ms=30_000,           # Ping every 30s
    timeout_ms=5_000,         # Wait 5s for pong
    permit_without_calls=True, # Ping even when idle
)

retry = RetryConfig(
    max_attempts=3,                    # Try 3 times
    initial_backoff_seconds=1.0,       # First retry after 1s
    max_backoff_seconds=30.0,          # Max delay 30s
    backoff_multiplier=2.0,            # Double each retry
    jitter_fraction=0.1,               # +/- 10% random
    retryable_codes=(                  # Retry these errors
        "UNAVAILABLE",
        "RESOURCE_EXHAUSTED",
        "DEADLINE_EXCEEDED",
    ),
)

from cmdop.logging import setup_logging, get_logger
 
# Initialize logging at application startup
setup_logging(
    level="DEBUG",            # Minimum log level to capture
    log_to_file=True,         # Write logs to disk
    log_to_console=True,      # Also print logs to stdout
    app_name="my-app",        # Used in log file name
    rich_tracebacks=True,     # Pretty-print exception tracebacks
)
 
# Create a module-scoped logger instance
log = get_logger(__name__)
log.info("Starting operation")          # Informational message
log.debug("Details: %s", data)          # Debug-level detail (lazy formatting)
log.error("Failed!", exc_info=True)     # Error with full stack trace

# Enable structured JSON output instead of human-readable text
setup_logging(log_json=True)

{
  "timestamp": "2026-02-14T10:30:00Z",
  "level": "INFO",
  "logger": "cmdop.client",
  "message": "Connected to server"
}

from cmdop import CMDOPClient
 
# Default paths searched:
# 1. ~/.cmdop/agent.info
# 2. /tmp/cmdop-{uid}/agent.info
 
# Custom paths
client = CMDOPClient.local(
    discovery_paths=[
        "/custom/path/agent.info",
        "~/.my-cmdop/agent.info",
    ],
    use_defaults=False,  # Skip default paths
)

from cmdop.discovery import discover_agent, require_agent
 
# Search for a local agent on this machine
result = discover_agent(
    custom_paths=["/custom/agent.info"],  # Additional paths to check
    verify_alive=True,                    # Confirm agent process is running
)
 
if result.found:
    print(f"Agent: {result.agent_info}")  # AgentInfo with version, PID, address
else:
    print(f"Not found: {result.reason}")  # Reason discovery failed
 
# Or require — raises AgentNotFoundError if no agent is found
agent_info, path = require_agent()

from cmdop import AsyncCMDOPClient
 
# Fetch all registered agents from the remote API
agents = await AsyncCMDOPClient.list_agents(api_key="cmd_xxx")
 
for agent in agents:
    print(f"{agent.name}: {agent.status}")  # Status: ONLINE, OFFLINE, or BUSY
 
# Filter to only agents currently online
online = await AsyncCMDOPClient.get_online_agents(api_key="cmd_xxx")

from cmdop.discovery import AgentInfo
 
# AgentInfo holds metadata about a locally discovered agent
info: AgentInfo
print(f"Version: {info.version}")        # Agent software version
print(f"PID: {info.pid}")                # OS process ID
print(f"Transport: {info.transport}")    # Connection type: unix, pipe, or tcp
print(f"Address: {info.address}")        # Socket path or host:port
print(f"Started: {info.started_at}")     # Agent start timestamp

from cmdop.discovery import RemoteAgentInfo
 
# RemoteAgentInfo holds metadata about a cloud-registered agent
agent: RemoteAgentInfo
print(f"Agent ID: {agent.agent_id}")      # Unique agent identifier
print(f"Name: {agent.name}")              # User-assigned agent name
print(f"Hostname: {agent.hostname}")      # Machine hostname
print(f"Platform: {agent.platform}")      # OS platform (linux, darwin, windows)
print(f"Status: {agent.status}")          # ONLINE, OFFLINE, or BUSY
print(f"Last Seen: {agent.last_seen}")    # Last heartbeat timestamp
print(f"Is Online: {agent.is_online}")    # Convenience bool for status == ONLINE

from cmdop import ExtractOptions
 
options = ExtractOptions(
    model="openai/gpt-4o",        # LLM model
    temperature=0.0,              # 0.0 = deterministic
    max_tokens=4096,              # Max response tokens
    max_retries=3,                # Retries on validation failure
    timeout_seconds=60,           # Total timeout
    working_directory="/app",     # Working dir for tools
    enabled_tools=["read_file", "list_dir"],  # Limit tools
)
 
# Run the extraction and return a validated Pydantic model
result = await client.extract.run(
    output_model=MyModel,             # Pydantic model defining output schema
    prompt="Extract data from...",    # Instruction for the LLM
    options=options,                  # ExtractOptions configured above
)

from cmdop import AgentRunOptions, AgentType
 
options = AgentRunOptions(
    model="openai/gpt-4o",
    max_turns=10,           # Max conversation turns
    max_retries=3,          # Max retries on error
    timeout_seconds=300,    # Total timeout
)
 
# Start an agent run with the specified type and options
result = await client.agent.run(
    prompt="Check disk usage",           # Task instruction for the agent
    agent_type=AgentType.TERMINAL,       # Agent type (see available types below)
    options=options,                     # AgentRunOptions configured above
)

from cmdop import AgentType
 
# Simple Q&A
AgentType.CHAT
 
# Execute terminal commands
AgentType.TERMINAL
 
# Direct command execution
AgentType.COMMAND
 
# Delegate to specialists
AgentType.ROUTER
 
# Multi-step planning
AgentType.PLANNER

from cmdop.services.browser.models import WaitUntil
 
# Create a new browser session with specified options
session = await client.browser.create_session(
    headless=True,              # Run without visible browser window
    devtools=False,             # Do not open DevTools panel
    window_size=(1920, 1080),   # Set viewport width x height in pixels
)
 
# Navigate to a URL and wait for network activity to finish
await session.navigate(
    "https://example.com",
    wait_until=WaitUntil.NETWORKIDLE,  # Wait until no pending network requests
)

from cmdop import reset_settings
 
# Reset all SDK settings back to their default values (useful in tests)
reset_settings()