openclaw-plus

Multi-capability dev skill for chained workflows involving Python execution, package management, git, HTTP requests, file operations, process management, sub-agents, or webhook notifications. Use when the task combines two or more of these — or when any single one requires disciplined error handling.

OpenClaw+ 🚀

A modular super-skill for development and automation workflows. Chain capabilities together or use them individually.

Capabilities

Capability	Use for
run_python	Execute Python code or scripts
install_package	Install pip/system packages
git_status / git_commit	Version control operations
http_request	Fetch URLs or call APIs (GET, POST, PUT, DELETE)
file_ops	Read, write, move, copy, delete, diff files
process	Start, monitor, and stop background processes
sub_agent	Delegate tasks to a spawned agent
notify	Send webhook notifications (Slack, Discord, ntfy, custom)

---

Core Capabilities

1. Python Execution (run_python)

python

# Simple execution
run_python("print('Hello, world!')")

# Multi-line with packages
run_python("""
import pandas as pd
df = pd.read_csv('data.csv')
print(df.describe())
""")

Notes:

• Use absolute paths for files
• For scripts >50 lines, write to a .py file first then execute
• Capture both stdout and stderr

---

2. Package Installation (install_package)

bash

install_package("pandas")                    # single
install_package("numpy==1.24.0")             # pinned version
install_package("requests beautifulsoup4")   # multiple
install_package("-r requirements.txt")       # from file

Implementation: pip install <package> --break-system-packages

Always check if a package is already installed before installing. Handle version conflicts explicitly.

---

3. Git Operations (git_status, git_commit)

bash

git_status()                                         # check current state
git_status("/path/to/repo")                          # specific directory
git_commit("feat: add data pipeline")                # commit staged files
git_commit("fix: resolve parsing error", stage_all=True)  # stage all + commit

Conventional commit types: feat, fix, docs, style, refactor, test, chore

Keep first line under 50 characters. Always run git_status() before committing. Configure user.email and user.name if git commit fails with identity errors.

---

4. HTTP Requests (http_request)

Unified capability for both URL fetching and API calls.

python

# GET request
data = http_request("https://api.example.com/users")

# POST with JSON body
result = http_request(
    "https://api.example.com/create",
    method="POST",
    json={"name": "Alice", "role": "admin"}
)

# With authentication
data = http_request(
    "https://api.example.com/protected",
    headers={"Authorization": "Bearer TOKEN"}
)

# GraphQL
result = http_request(
    "https://api.example.com/graphql",
    method="POST",
    json={"query": "{ users { id name } }"}
)

Notes:

• Default timeout: 30s — always set explicitly for slow endpoints
• Call response.raise_for_status() before parsing
• Handle rate limiting with exponential backoff for retries

---

5. File Operations (file_ops)

Read, write, move, copy, delete, and diff files without shelling out manually.

python

# Read
content = file_ops("read", "data/report.txt")
data = file_ops("read", "config.json", parse_json=True)

# Write
file_ops("write", "output/results.txt", content="Analysis complete.")
file_ops("write", "data/config.json", content={"key": "value"}, as_json=True)

# Append
file_ops("append", "logs/run.log", content="[2026-05-28] Step complete\n")

# Copy / Move
file_ops("copy", "report.txt", dest="archive/report_v1.txt")
file_ops("move", "temp_output.csv", dest="final/output.csv")

# Delete
file_ops("delete", "temp/scratch.txt")
file_ops("delete", "temp/", recursive=True)  # delete directory

# Diff
diff = file_ops("diff", "old_config.json", "new_config.json")

# List directory
entries = file_ops("list", "data/", pattern="*.csv")

Best practices:

• Always check if a file exists before reading or deleting
• Use absolute paths in scripts to avoid working-directory surprises
• Prefer append over write for log files to avoid overwriting
• Clean up temp files at the end of workflows

---

6. Process Management (process)

Start background processes, check their status, tail output, and stop them.

python

# Start a background process
pid = process("start", "python3 server.py", cwd="/app", capture_output=True)

# Start and get a handle
job = process("start", "npm run build", env={"NODE_ENV": "production"})

# Check status
status = process("status", pid)
# Returns: { "pid": 1234, "running": True, "exit_code": None }

# Tail output
logs = process("logs", pid, lines=50)

# Wait for completion (blocks)
result = process("wait", pid, timeout=120)
# Returns: { "exit_code": 0, "stdout": "...", "stderr": "..." }

# Kill
process("kill", pid)
process("kill", pid, signal="SIGTERM")  # graceful shutdown

Common patterns:

python

# Start a server, wait for it to be ready, then run tests
server_pid = process("start", "./start_server.sh")
time.sleep(2)  # brief startup wait

result = process("wait", test_pid, timeout=60)
if result["exit_code"] != 0:
    logs = process("logs", test_pid)
    print(f"Tests failed:\n{logs}")
finally:
    process("kill", server_pid)

Notes:

• Always kill background processes when done — don't leave orphans
• Use process("wait", pid, timeout=N) rather than time.sleep for predictable completion
• Check exit_code after wait — a 0 is success, anything else needs handling
• Redirect stdout/stderr to files for long-running processes

---

7. Sub-Agents (sub_agent)

Delegate a task to a spawned agent and get the result back. Use when a task is large enough to benefit from parallel work or clean separation of concerns.

python

# Spawn an agent with a specific task
result = sub_agent(
    task="Analyze the CSV at data/sales.csv and return a JSON summary with total revenue, top 5 products, and monthly breakdown.",
    context={"file": "data/sales.csv"}
)

# Spawn multiple agents in parallel
results = sub_agent(
    tasks=[
        "Summarize the Q1 report at reports/q1.pdf",
        "Summarize the Q2 report at reports/q2.pdf",
        "Summarize the Q3 report at reports/q3.pdf"
    ],
    parallel=True
)

# Agent with specific tools/skills
result = sub_agent(
    task="Generate a bar chart of monthly revenue from data/revenue.json and save it as charts/revenue.png",
    skills=["run_python", "file_ops"]
)

When to use sub-agents:

• Tasks that can be parallelized (summarizing multiple documents, processing batches)
• Isolating a complex sub-task so failures don't cascade
• Tasks that need a fresh context window (e.g., processing a large file)
• Delegating specialized work (one agent for data, one for formatting)

When NOT to use sub-agents:

• Simple sequential steps that take seconds — overhead isn't worth it
• When shared state between steps is required — sub-agents don't share memory
• When you need tight control over each intermediate result

Notes:

• Always give the sub-agent a clear, self-contained task — include all context it needs
• Specify expected output format explicitly (e.g., "return JSON with these fields")
• Handle the case where a sub-agent returns an error or unexpected format
• For parallel agents, define how to merge/reconcile their outputs

---

8. Webhook Notifications (notify)

Send notifications when a workflow step completes, fails, or needs attention.

python

# Slack
notify(
    "slack",
    webhook_url=os.environ["SLACK_WEBHOOK"],
    message="✅ Data pipeline complete. 4,821 records processed.",
    channel="#data-alerts"
)

# Discord
notify(
    "discord",
    webhook_url=os.environ["DISCORD_WEBHOOK"],
    message="Build failed on main branch.",
    username="OpenClaw Bot",
    embed={
        "title": "Build Failure",
        "description": "Step 3 exited with code 1",
        "color": 0xFF0000
    }
)

# ntfy (self-hosted push notifications)
notify(
    "ntfy",
    topic="my-agent-alerts",
    message="Scraping job finished: 320 products saved.",
    priority="default",
    tags=["white_check_mark"]
)

# Generic webhook (any HTTP endpoint)
notify(
    "webhook",
    url="https://hooks.example.com/alert",
    payload={"event": "pipeline_complete", "records": 4821, "status": "ok"},
    headers={"X-Secret": os.environ["WEBHOOK_SECRET"]}
)

Best practices:

• Always store webhook URLs in environment variables — never hardcode
• Send notifications at meaningful checkpoints: job start, completion, and on failure
• Include actionable context in the message: what ran, how many records, what failed
• For failure notifications, include the error message and the step that failed
• Use notify at the end of long workflows so the user doesn't have to watch

Failure notification pattern:

python

try:
    run_pipeline()
    notify("slack", webhook_url=SLACK_URL, message="✅ Pipeline complete")
except Exception as e:
    notify("slack", webhook_url=SLACK_URL, message=f"❌ Pipeline failed: {e}")
    raise

---

Workflow Patterns

Data Pipeline with Notification

python

install_package("pandas requests")
data = http_request("https://api.example.com/dataset")
file_ops("write", "raw_data.json", content=data, as_json=True)
run_python("""
import pandas as pd, json
df = pd.read_json('raw_data.json')
df.dropna().to_csv('clean_data.csv', index=False)
print(f"Processed {len(df)} records")
""")
git_commit("feat: add cleaned dataset", stage_all=True)
notify("slack", webhook_url=os.environ["SLACK_WEBHOOK"], message="✅ Pipeline done")

Parallel Document Processing

python

# Delegate each document to a sub-agent in parallel
reports = ["q1.pdf", "q2.pdf", "q3.pdf", "q4.pdf"]
summaries = sub_agent(
    tasks=[f"Summarize reports/{r} and return JSON with key_findings and total_revenue" for r in reports],
    parallel=True
)
file_ops("write", "output/annual_summary.json", content=summaries, as_json=True)
notify("slack", webhook_url=os.environ["SLACK_WEBHOOK"], message="Annual summaries ready")

Background Service + Tests

python

server = process("start", "python3 -m app.server", cwd="/app")
time.sleep(2)
try:
    result = process("wait",
        process("start", "pytest tests/integration/", capture_output=True),
        timeout=120
    )
    status = "✅ All tests passed" if result["exit_code"] == 0 else "❌ Tests failed"
finally:
    process("kill", server)
    notify("slack", webhook_url=os.environ["SLACK_WEBHOOK"], message=status)

File Batch Processing with Sub-Agent

python

csv_files = file_ops("list", "data/incoming/", pattern="*.csv")
results = sub_agent(
    tasks=[f"Validate and clean data/incoming/{f}, save to data/processed/{f}" for f in csv_files],
    parallel=True,
    skills=["file_ops", "run_python"]
)
file_ops("write", "data/processing_report.json", content=results, as_json=True)
git_commit("chore: process incoming CSV batch", stage_all=True)

---

Error Handling

Always handle errors at each step — don't let a mid-workflow failure leave things in a broken state.

python

# File operations
if not file_ops("exists", "data/input.csv"):
    raise FileNotFoundError("data/input.csv not found — aborting")

# Process management
result = process("wait", pid, timeout=60)
if result["exit_code"] != 0:
    logs = process("logs", pid)
    notify("slack", webhook_url=SLACK_URL, message=f"❌ Process failed:\n{logs[-500:]}")
    raise RuntimeError(f"Process exited {result['exit_code']}")

# Sub-agent results
for i, result in enumerate(sub_agent_results):
    if result.get("error"):
        print(f"Sub-agent {i} failed: {result['error']}")

# HTTP requests
response = http_request(url)
if response.status_code != 200:
    raise RuntimeError(f"API returned {response.status_code}: {response.text}")

# Git — nothing to commit
status = git_status()
if "nothing to commit" not in status:
    git_commit("chore: update output files", stage_all=True)

---

Security

• Credentials: Always use environment variables. Never hardcode API keys, tokens, or webhook URLs.
• File paths: Validate paths before file operations. Don't allow user input to construct paths directly.
• Processes: Avoid passing unsanitized user input to process("start", ...).
• Sub-agents: Don't pass secrets in the task string — inject them via environment variables instead.
• Webhooks: Use HTTPS endpoints only. Validate with a shared secret where the service supports it.

---

Quick Reference

python

run_python(code)
install_package("name" or "-r requirements.txt")
git_status() / git_commit("message", stage_all=True)
http_request(url, method="GET", json={}, headers={})
file_ops("read|write|append|copy|move|delete|diff|list|exists", path, ...)
process("start|status|logs|wait|kill", pid_or_cmd, ...)
sub_agent(task="...", parallel=False, skills=[...])
notify("slack|discord|ntfy|webhook", webhook_url="...", message="...")