Error Handling

CMDOPError (base)
├── ConnectionError
│   ├── AgentNotRunningError
│   ├── StalePortFileError
│   ├── ConnectionTimeoutError
│   └── ConnectionLostError
├── AuthenticationError
│   ├── InvalidAPIKeyError
│   ├── PermissionDeniedError
│   └── TokenExpiredError
├── MethodNotFoundError
├── AgentError
│   ├── AgentOfflineError
│   ├── AgentBusyError
│   └── FeatureNotAvailableError
├── SessionError
│   ├── SessionNotFoundError
│   ├── SessionClosedError
│   └── SessionInterruptedError
├── FileError
│   ├── FileNotFoundError
│   ├── FilePermissionError
│   └── FileTooLargeError
├── BrowserError
│   ├── BrowserSessionClosedError
│   ├── BrowserNavigationError
│   └── BrowserElementNotFoundError
└── RateLimitError

from cmdop import CMDOPClient
from cmdop.exceptions import AgentNotRunningError
 
try:
    # Attempt to connect to the local CMDOP agent
    client = CMDOPClient.local()
except AgentNotRunningError as e:
    # Error message includes instructions to start the agent
    print(e)
    # Output:
    # CMDOP agent is not running.
    #
    # To fix, run one of:
    #   - cmdop serve
    #   - Open CMDOP Desktop

from cmdop.exceptions import StalePortFileError
 
try:
    client = CMDOPClient.local()
except StalePortFileError as e:
    # Remove the stale discovery file left by a dead agent
    e.cleanup()
    # Retry connection after cleanup
    client = CMDOPClient.local()

from cmdop.exceptions import ConnectionTimeoutError
 
try:
    # Attempt remote connection (subject to timeout)
    client = CMDOPClient.remote(api_key="cmd_xxx")
except ConnectionTimeoutError as e:
    # Access the timeout duration from the exception
    print(f"Connection timed out after {e.timeout_seconds}s")

from cmdop.exceptions import ConnectionLostError
 
try:
    # Stream events from a long-running command
    async for event in client.terminal.stream("long-command"):
        process(event)
except ConnectionLostError:
    # Connection dropped mid-stream; reconnect and resume
    print("Connection lost during stream")

from cmdop.exceptions import InvalidAPIKeyError
 
try:
    # Server validates the API key on connection
    client = CMDOPClient.remote(api_key="invalid")
except InvalidAPIKeyError:
    # Key passed format check but was rejected by the server
    print("API key format valid but rejected by server")

from cmdop.exceptions import PermissionDeniedError
 
try:
    # Unix socket connections verify process UID matches
    client = CMDOPClient.local()
except PermissionDeniedError as e:
    # Shows the UID mismatch between agent and caller
    print(f"Agent UID: {e.agent_uid}, Your UID: {e.caller_uid}")

from cmdop.exceptions import TokenExpiredError
 
try:
    result = await client.terminal.execute("ls")
except TokenExpiredError:
    # Token expired mid-session; refresh and retry
    await client.refresh_token()

from cmdop.exceptions import AgentOfflineError
 
try:
    result = await client.agent.run("task")
except AgentOfflineError as e:
    # Access agent metadata from the exception
    print(f"Agent {e.agent_id} is offline")
    print(f"Last seen: {e.last_seen}")  # Timestamp of last heartbeat

from cmdop.exceptions import AgentBusyError
 
try:
    result = await client.agent.run("task")
except AgentBusyError:
    # Agent is processing another request; wait before retrying
    await asyncio.sleep(5)

from cmdop.exceptions import FeatureNotAvailableError
 
try:
    # Some features are only available in specific connection modes
    result = await client.browser.create_session()
except FeatureNotAvailableError as e:
    # Exception includes which feature and which mode caused the error
    print(f"Feature '{e.feature}' not available in {e.mode} mode")

from cmdop.exceptions import MethodNotFoundError
 
try:
    skills = await client.skills.list()
except MethodNotFoundError:
    # Server is running an older version or needs to be restarted
    print("Server doesn't support this method. Restart with: make grpc")

from cmdop.exceptions import SessionNotFoundError
 
try:
    # Retrieve an existing session by its ID
    session = await client.terminal.get_session(session_id)
except SessionNotFoundError as e:
    # Session may have been garbage-collected after grace period
    print(f"Session {e.session_id} not found")

from cmdop.exceptions import SessionClosedError
 
try:
    await session.execute("ls")
except SessionClosedError as e:
    # Session was terminated; create a new one to continue
    print(f"Session {e.session_id} is closed")

from cmdop.exceptions import SessionInterruptedError
 
try:
    # Stream events from command execution
    async for event in session.stream("command"):
        process(event)
except SessionInterruptedError:
    # Stream was interrupted; reconnect to resume
    print("Stream interrupted")

from cmdop.exceptions import FileNotFoundError
 
try:
    # Read a file from a remote machine
    content = await client.files.read("server", "/nonexistent")
except FileNotFoundError as e:
    # Exception includes the path that was not found
    print(f"File not found: {e.path}")

from cmdop.exceptions import FilePermissionError
 
try:
    # Attempt to write to a restricted file
    await client.files.write("server", "/etc/passwd", "data")
except FilePermissionError as e:
    # Exception includes both the failed operation and the file path
    print(f"Permission denied: {e.operation} on {e.path}")

from cmdop.exceptions import FileTooLargeError
 
try:
    content = await client.files.read("server", "/large-file")
except FileTooLargeError as e:
    # Compare actual size vs allowed maximum; use streaming for large files
    print(f"File too large: {e.size_bytes} > {e.max_bytes}")

from cmdop.exceptions import BrowserSessionClosedError
 
try:
    await session.navigate("https://example.com")
except BrowserSessionClosedError as e:
    # error_detail includes instructions for restarting the session
    print(e.error_detail)

from cmdop.exceptions import BrowserNavigationError
 
try:
    await session.navigate("https://invalid-url")
except BrowserNavigationError as e:
    # Exception includes the target URL and a detailed error description
    print(f"Failed to navigate to {e.url}: {e.error_detail}")

from cmdop.exceptions import BrowserElementNotFoundError
 
try:
    await session.click("button.nonexistent")
except BrowserElementNotFoundError as e:
    # Exception includes the selector and debugging tips
    print(f"Element not found: {e.selector}")

from cmdop.exceptions import RateLimitError
import asyncio
 
try:
    result = await client.agent.run("task")
except RateLimitError as e:
    # Access rate limit metadata from the exception
    print(f"Rate limited. Retry after {e.retry_after_seconds}s")
    print(f"Limit: {e.limit}, Remaining: {e.remaining}")
 
    if e.retry_after_seconds:
        # Wait the server-specified duration before retrying
        await asyncio.sleep(e.retry_after_seconds)

from cmdop import CMDOPClient
from cmdop.exceptions import (
    AgentNotRunningError,
    StalePortFileError,
    ConnectionLostError,
)
from cmdop.helpers import ensure_desktop_running
 
def get_client_with_recovery():
    """Get client with automatic recovery."""
    try:
        return CMDOPClient.local()
    except StalePortFileError as e:
        # Remove stale port file and try launching the desktop app
        e.cleanup()
        if ensure_desktop_running():
            return CMDOPClient.local()
        raise  # Re-raise if desktop app failed to start
    except AgentNotRunningError:
        # Attempt to start the desktop app automatically
        if ensure_desktop_running():
            return CMDOPClient.local()
        raise  # Re-raise if desktop app failed to start

from cmdop.helpers import with_auto_restart
 
# Decorator auto-restarts the function if the agent crashes mid-execution
@with_auto_restart
def scrape_data(client):
    """Function auto-restarts if agent crashes."""
    with client.browser.create_session() as session:
        session.navigate("https://example.com")
        return session.dom.soup()  # Returns a BeautifulSoup object

import asyncio
from cmdop.exceptions import CMDOPError, RateLimitError
 
async def execute_with_retry(client, command, max_retries=3):
    for attempt in range(max_retries):
        try:
            return await client.terminal.execute(command)
        except RateLimitError as e:
            # Use server-provided delay for rate limit errors
            if e.retry_after_seconds:
                await asyncio.sleep(e.retry_after_seconds)
            continue
        except CMDOPError as e:
            # On last attempt, re-raise instead of retrying
            if attempt == max_retries - 1:
                raise
            # Exponential backoff: 1s, 2s, 4s between retries
            await asyncio.sleep(2 ** attempt)

from cmdop import CMDOPClient
from cmdop.exceptions import (
    CMDOPError,
    ConnectionError,
    AuthenticationError,
    AgentError,
    SessionError,
    FileError,
    RateLimitError,
)
 
async def safe_execute(client, command):
    """Execute a command with automatic recovery for common errors."""
    try:
        return await client.terminal.execute(command)
 
    except AuthenticationError:
        # Token expired or invalid; refresh and retry
        await client.refresh_token()
        return await client.terminal.execute(command)
 
    except ConnectionError:
        # Connection dropped; re-establish and retry
        await client.reconnect()
        return await client.terminal.execute(command)
 
    except RateLimitError as e:
        # Wait the server-specified delay (or 60s default) before retry
        await asyncio.sleep(e.retry_after_seconds or 60)
        return await client.terminal.execute(command)
 
    except SessionError:
        # Session was lost; create a fresh session and retry
        await client.terminal.create_session()
        return await client.terminal.execute(command)
 
    except CMDOPError as e:
        # Catch-all for unrecoverable CMDOP errors; log and propagate
        logger.error(f"CMDOP error: {e}")
        raise