cmdop 0.1.12

Python SDK for CMDOP agent interaction

cmdop

Any machine. One API.

from cmdop import CMDOPClient

with CMDOPClient.remote(api_key="cmd_xxx") as server:
    server.terminal.execute("docker restart app")
    server.files.write("/etc/nginx/nginx.conf", new_config)
    logs = server.files.read("/var/log/app.log")

No SSH. No VPN. No open ports.

---

How

Your Code ──── Cloud Relay ──── Agent (on server)
                    │
        Outbound only, works through any NAT/firewall

Agent connects out. Your code connects to relay. Done.

---

What You Get

Terminal:

session = server.terminal.create()
server.terminal.send_input(session.session_id, "kubectl get pods\n")
output = server.terminal.get_history(session.session_id)

Files:

server.files.list("/var/log")
server.files.read("/etc/nginx/nginx.conf")
server.files.write("/tmp/config.json", b'{"key": "value"}')

Browser — scrape without Selenium/Playwright bullshit:

with server.browser.create_session() as browser:
    browser.navigate("https://shop.com/products")

    # One call → structured data
    products = browser.extract_data(
        ".product-card",
        '{"name": "h2", "price": ".price", "url": {"selector": "a", "attr": "href"}}',
        limit=100
    )["items"]
    # → [{"name": "iPhone", "price": "$999", "url": "/p/123"}, ...]

AI Agent — typed responses:

from pydantic import BaseModel

class Health(BaseModel):
    status: str
    cpu: float
    issues: list[str]

result = server.agent.run("Check server health", output_schema=Health)
health: Health = result.output  # Typed!

---

Real World

AI Agent + Typed Output:

class DeployResult(BaseModel):
    success: bool
    version: str
    errors: list[str]

result = server.agent.run(
    "Deploy myapp:v2.1, verify containers healthy",
    output_schema=DeployResult
)
if not result.output.success:
    rollback(result.output.errors)

Fleet Update (1000 devices):

async def update_fleet(keys: list[str], config: bytes):
    async with asyncio.TaskGroup() as tg:
        for key in keys:
            tg.create_task(update_one(key, config))

async def update_one(key: str, config: bytes):
    async with AsyncCMDOPClient.remote(api_key=key) as dev:
        await dev.files.write("/etc/app/config.yml", config)
        await dev.terminal.execute("systemctl restart app")

Debug Customer Machine:

with CMDOPClient.remote(api_key=customer_key) as m:
    m.terminal.send_input(sid, "ps aux\n")
    logs = m.files.read("~/Library/Logs/MyApp/error.log")
    m.terminal.send_input(sid, "df -h\n")

Scrape Products:

class Product(SDKBaseModel):
    __base_url__ = "https://amazon.com"
    title: str = ""
    price: int = 0   # "$1,299" → 1299
    url: str = ""    # "/dp/..." → "https://amazon.com/dp/..."

with client.browser.create_session(headless=True) as b:
    b.navigate("https://amazon.com/s?k=laptop")
    raw = b.extract_data(".s-result-item", '{"title": "h2", "price": ".a-price-whole", "url": {"selector": "a", "attr": "href"}}', limit=50)
    products = Product.from_list(raw["items"])  # clean + dedupe + filter

---

Install

pip install cmdop

Usage

from cmdop import CMDOPClient, AsyncCMDOPClient

# Remote (via cloud relay)
with CMDOPClient.remote(api_key="cmd_xxx") as client:
    client.files.list("/home")

# Local (direct IPC)
with CMDOPClient.local() as client:
    client.terminal.execute("ls -la")

# Async
async with AsyncCMDOPClient.remote(api_key="cmd_xxx") as client:
    await client.files.read("/etc/hostname")

---

API

Terminal

Method	Description
create(shell)	Start session
send_input(id, data)	Send commands
get_history(id)	Get output
resize(id, cols, rows)	Resize
send_signal(id, signal)	SIGINT/SIGTERM
close(id)	End session

Files

Method	Description
list(path)	List dir
read(path)	Read file
write(path, content)	Write file
delete(path)	Delete
copy(src, dst)	Copy
move(src, dst)	Move
mkdir(path)	Create dir
info(path)	Metadata

Browser

Method	Description
create_session(headless)	Start browser
navigate(url)	Go to URL
click(selector)	Click
type(selector, text)	Type
wait_for(selector, timeout_ms)	Wait
extract(selector, attr)	Get text/attr
extract_regex(pattern)	Regex matches
validate_selectors(item, fields)	Check selectors
extract_data(item, fields, limit)	Bulk extract → list[dict]
execute_script(js)	Run JS
screenshot()	PNG
get_cookies() / set_cookies()	Cookies

Scraping workflow:

with client.browser.create_session() as b:
    b.navigate("https://cars.com/listings")

    # 1. Validate (fail fast if site changed)
    v = b.validate_selectors(".item", {"title": "h2", "price": ".price"})
    if not v["valid"]:
        raise Exception(v["errors"])  # also has counts, samples

    # 2. Extract
    cars = b.extract_data(
        ".item",
        '{"title": "h2", "price": {"selector": ".price", "regex": "\\\\d+"}}',
        limit=200
    )["items"]

SDKBaseModel

Auto-cleaning Pydantic model for scraped data. No more manual .strip(), regex, URL joining.

from cmdop import SDKBaseModel

class Product(SDKBaseModel):
    __base_url__ = "https://shop.com"  # for relative URLs

    name: str = ""    # "  iPhone 15  \n" → "iPhone 15"
    price: int = 0    # "$1,299.00" → 1299
    rating: float = 0 # "4.5 stars" → 4.5
    url: str = ""     # "/p/123" → "https://shop.com/p/123"

# Batch parse with auto dedupe + filter
products = Product.from_list(raw["items"])

What it does:

Type	Input	Output
str	" text \n\t "	"text"
int	"$27,471"	27471
float	"4.5 out of 5"	4.5
str (url field)	"/path"	"https://base.com/path"

Usage with extract_data:

raw = browser.extract_data(".product", fields, limit=100)
products = Product.from_list(raw["items"])  # clean + dedupe by url + filter empty

Agent

Method	Description
run(prompt, output_schema)	Run agent, get typed result

Types: chat, terminal, command, router, planner

---

Security

• TLS everywhere
• Outbound only — no open ports
• API key scoping
• Audit logs

Requirements

• Python 3.10+
• CMDOP agent on target

Links

cmdop.com

License

MIT