CMDOPHandler

from cmdop_bot import CMDOPHandler, Model
 
# Use as async context manager — automatically closes the client on exit
async with CMDOPHandler(
    api_key="cmdop_xxx",                     # CMDOP API key for authentication
    machine="my-server",                     # Target machine hostname
    model=Model.balanced(),                  # AI model tier (balanced performance)
) as cmdop:
    # Run an AI agent task and print the result
    result = await cmdop.run_agent("List files in /tmp")
    print(result.text)
 
    # Execute a shell command; returns raw bytes and exit code
    output, exit_code = await cmdop.execute_shell("ls -la")
    print(output.decode())
 
    # List directory contents; iterate over file entries
    files = await cmdop.list_files("/var/log")
    for f in files.entries:
        print(f.name)
 
    # Read a file; returns raw bytes
    content = await cmdop.read_file("/etc/hostname")
    print(content.decode())
 
    # List available skills on the machine
    skills = await cmdop.list_skills()
    for s in skills:
        print(f"{s.name}: {s.description}")
 
    # Show skill details
    detail = await cmdop.show_skill("code-review")
    if detail.found:
        print(detail.info.name, detail.info.description)
 
    # Run a skill with a prompt
    result = await cmdop.run_skill("code-review", "Review this PR")
    print(result.text)
 
    # Switch to a different target machine mid-session
    await cmdop.set_machine("other-server")

# Run an AI agent task and check if it succeeded
result = await cmdop.run_agent("Check server health")
if result.success:
    print(result.text)                       # .text contains the agent's response

# Define a callback to handle streaming events as they arrive
async def on_event(event):
    if event.type == "token":
        print(event.text, end="")            # Print each token as it streams in
    elif event.type == "tool_start":
        print(f"\nUsing tool: {event.name}") # Notify when the agent invokes a tool
 
# Stream the agent run — on_event fires for each token and tool event
result = await cmdop.run_agent_stream("Analyze logs", on_event=on_event)

# Execute a shell command with an explicit timeout (seconds)
output, exit_code = await cmdop.execute_shell("df -h", timeout=30.0)
print(output.decode())                       # Decode raw bytes to string
# exit_code: 0 = success, -1 = timeout, other = error

# List directory contents; returns FileListResponse with .entries
files = await cmdop.list_files("/var/log")
for f in files.entries:
    print(f.name, f.size, f.type)            # Each entry has name, size, and type

# Read file contents as raw bytes, then decode to string
content = await cmdop.read_file("/etc/nginx/nginx.conf")
print(content.decode())

# Get file metadata — returns FileInfo with path, size, type, modified_at, permissions
info = await cmdop.file_info("/etc/hostname")
print(info.size, info.modified_at)

# List all skills available on the target machine
skills = await cmdop.list_skills()
for s in skills:
    print(f"{s.name}: {s.description} [{s.origin}]")

# Get details for a specific skill by name
detail = await cmdop.show_skill("code-review")
if detail.found:
    print(detail.info.name)          # Skill name
    print(detail.info.description)   # What the skill does
    print(detail.info.author)        # Author
    print(detail.info.version)       # Version string
    print(detail.source)             # File path on the machine
    print(detail.content)            # Skill content/template

# Run a skill and check the result
result = await cmdop.run_skill("code-review", "Review this PR", timeout=300)
if result.success:
    print(result.text)               # Skill output
    print(f"Took {result.duration_seconds:.1f}s")
else:
    print(f"Error: {result.error}")

# Switch all services (terminal, files, agent, skills) to a different machine
hostname = await cmdop.set_machine("prod-server")
print(f"Connected to: {hostname}")

from cmdop_bot import CMDOPHandler
 
# Example: embed CMDOPHandler into your own custom bot class
class MyCustomBot:
    def __init__(self, cmdop_api_key: str):
        # Create a CMDOPHandler instance for CMDOP operations
        self.cmdop = CMDOPHandler(api_key=cmdop_api_key)
 
    async def handle_message(self, text: str) -> str:
        if text.startswith("/shell "):
            # Extract the command after "/shell " and execute it
            output, code = await self.cmdop.execute_shell(text[7:])
            return output.decode()
        elif text.startswith("/agent "):
            # Extract the task after "/agent " and run the AI agent
            result = await self.cmdop.run_agent(text[7:])
            return result.text
        return "Unknown command"
 
    async def close(self):
        # Clean up the CMDOP client when shutting down
        await self.cmdop.close()