Advanced Automation Patterns

#!/usr/bin/env python3
# webhook-receiver.py — minimal Flask webhook that triggers OpenClaw

from flask import Flask, request
import subprocess, json, hmac, hashlib, os

app = Flask(__name__)
SECRET = os.environ["WEBHOOK_SECRET"].encode()

def verify_signature(payload, sig_header):
    expected = "sha256=" + hmac.new(SECRET, payload, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, sig_header)

@app.route("/webhook", methods=["POST"])
def webhook():
    payload = request.get_data()
    if not verify_signature(payload, request.headers.get("X-Hub-Signature-256", "")):
        return "Unauthorized", 401

    data = json.loads(payload)
    event = request.headers.get("X-GitHub-Event")

    if event == "issues" and data.get("action") == "opened":
        issue_body = data["issue"]["body"] or ""
        result = subprocess.run(
            ["openclaw", "run",
             "Classify this issue: bug/feature/question/docs. Assign priority 1-3. Output JSON.",
             "--stdin"], input=issue_body, capture_output=True, text=True
        )
        print(f"Classified: {result.stdout.strip()}")

    return "OK", 200

if __name__ == "__main__":
    app.run(port=8080)

#!/usr/bin/env bash
# stateful-pipeline.sh

STATE_FILE=".pipeline-state.json"

# Load or initialise state
if [[ -f "$STATE_FILE" ]]; then
  STAGE=$(jq -r '.stage' "$STATE_FILE")
  echo "Resuming from stage: $STAGE"
else
  STAGE="extract"
  echo '{"stage":"extract","completed":[]}' > "$STATE_FILE"
fi

run_stage() {
  local name=$1; shift
  echo "Running stage: $name"
  if "$@"; then
    jq --arg s "$name" '.completed += [$s] | .stage = "done"' \
      "$STATE_FILE" > tmp.json && mv tmp.json "$STATE_FILE"
  else
    echo "Stage $name failed." >&2; exit 1
  fi
}

case "$STAGE" in
  extract) run_stage "extract" openclaw run "Extract key data as JSON" --file raw-data.csv --output extracted.json ;&
  transform) run_stage "transform" openclaw run "Clean and normalize this JSON" --file extracted.json --output clean.json ;&
  load)    run_stage "load" python3 scripts/load-to-db.py clean.json ;;
  done) echo "Pipeline already complete." ;;
esac

#!/usr/bin/env bash
# multi-agent-review.sh

FILE=$1

# Run three specialist agents in parallel
openclaw run "Review $FILE for security vulnerabilities only. Output JSON." \
  --file "$FILE" > security-review.json &

openclaw run "Review $FILE for performance issues only. Output JSON." \
  --file "$FILE" > perf-review.json &

openclaw run "Review $FILE for test coverage gaps only. Output JSON." \
  --file "$FILE" > test-review.json &

wait  # Wait for all three

# Synthesise
cat security-review.json perf-review.json test-review.json | \
  openclaw run "Merge these three specialist reviews into one prioritised report." \
  > final-report.md

echo "Report saved to final-report.md"

# openclaw config: rate limiting for high-volume pipelines
rate_limit:
  requests_per_minute: 40      # stay under provider limit
  tokens_per_minute: 80000
  concurrent_requests: 3       # parallelism cap
  retry:
    max_attempts: 4
    backoff: exponential
    base_delay: 2              # seconds

# Log all OpenClaw runs to a file with timestamps
openclaw run "summarise today's sales report" \
  --file daily-report.csv \
  2>&1 | tee -a /var/log/openclaw/$(date +%Y-%m-%d).log

# Check exit code and send alert on failure
if [ $? -ne 0 ]; then
  openclaw run "Send a Telegram alert: OpenClaw pipeline failed at $(date)" \
    --config telegram
fi

from openclaw.tools import register_tool

@register_tool(
    name="query_internal_db",
    description="Query the internal analytics database. Returns rows as JSON. "
                "Args: sql (str) — read-only SELECT statement only."
)
def query_internal_db(sql: str) -> str:
    # Returns rows as JSON list
    import sqlite3, json
    conn = sqlite3.connect("/data/analytics.db")
    rows = conn.execute(sql).fetchall()
    conn.close()
    return json.dumps(rows)

# Then in your automation config:
# tools: [query_internal_db, file_write, shell_exec]

# Step 1: index your docs (run once or on update)
openclaw index ./docs/ --output ./index/

# Step 2: run tasks — agent retrieves relevant chunks automatically
openclaw run "What changed in the API between v1 and v2?" \
  --rag-index ./index/ --model gpt-4o

# Step 3: scheduled nightly re-index + answer the daily digest
echo "0 2 * * * openclaw index ./docs/ -o ./index/ \
  && openclaw run 'Daily doc digest' --rag-index ./index/ \
     > /var/log/doc-digest.log 2>&1" | crontab -

# Dry run — shows planned steps and estimated API calls, no actual LLM calls
openclaw run "Summarise and classify all PRs this week" \
  --dry-run --verbose

# Estimate cost before running
openclaw run "Refactor all files in ./src" \
  --estimate-cost

# Run against a test fixture to validate output format
openclaw run "Extract action items" \
  --file ./test-fixtures/meeting-transcript.txt \
  --output-schema ./schemas/action-items.json \
  --assert-valid

# .github/workflows/openclaw-tests.yml
name: OpenClaw Workflow Tests
on: [push, pull_request]
jobs:
  test-workflows:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
      - run: pip install openclaw
      - name: Test summarisation workflow
        env:
          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
        run: |
          OUTPUT=$(openclaw run "Summarise in one sentence" \
            --file ./test-fixtures/sample.txt \
            --model gpt-4o-mini)
          [ ${#OUTPUT} -gt 20 ] || (echo "FAIL: output too short" && exit 1)
          echo "PASS: $OUTPUT"