cmdop 0.1.15

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.

---

Install

pip install cmdop

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")

---

Part 1: Remote Control

Terminal

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

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

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

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

Agent

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!

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

Types: chat, terminal, command, router, planner

Real World Examples

Deploy with typed result:

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")

---

Part 2: Web Parsing

Browser

DOM extraction:

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

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

JS fetch injection (bypass CORS):

with client.browser.create_session() as b:
    b.navigate("https://site.com")  # Get cookies/session

    # Single API call
    data = b.fetch_json("https://api.site.com/v1/items")

    # Parallel fetch via Promise.all
    results = b.fetch_all({
        "users": "https://api.site.com/v1/users",
        "orders": "https://api.site.com/v1/orders",
    })
    # → {"users": [...], "orders": [...]}

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]
fetch_json(url)	JS fetch → dict
fetch_all(urls)	Parallel fetch → dict[str, Any]
execute_script(js)	Run JS
screenshot()	PNG
get_cookies() / set_cookies()	Cookies

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"

    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"])

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"

Helpers

from cmdop import json_to_yaml, flatten_json

# Convert to YAML
yaml_str = json_to_yaml({"name": "test", "items": [1, 2, 3]})

# Flatten nested dict
flat = flatten_json({"user": {"name": "John", "address": {"city": "NYC"}}})
# → {"user.name": "John", "user.address.city": "NYC"}

Parsing Examples

Scrape with validation:

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"])

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

Scrape with SDKBaseModel:

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

Parallel API fetching:

with client.browser.create_session() as b:
    b.navigate("https://api.example.com")

    # Fetch 10 pages in parallel
    urls = {f"page_{i}": f"https://api.example.com/items?page={i}" for i in range(10)}
    results = b.fetch_all(urls)

    items = []
    for key, data in results.items():
        if data:
            items.extend(data.get("items", []))

JS fetch for protected APIs:

with client.browser.create_session() as b:
    b.navigate("https://site.com")  # Get session cookies

    # Fetch JSON API (inherits cookies)
    data = b.fetch_json("https://api.site.com/v1/data")

    # Parallel fetch multiple endpoints
    results = b.fetch_all({
        "inspection": f"https://api.site.com/v1/car/{car_id}/inspection",
        "options": f"https://api.site.com/v1/car/{car_id}/options",
    })

---

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