meatbag/meatbag.py

#!/usr/bin/env python3
import sys
import os
import json
import urllib.request
import urllib.error
import argparse

BACKEND_URL = "http://localhost:8080"

def get_api_key():
    if os.environ.get("MEATBAG_API_KEY"):
        return os.environ.get("MEATBAG_API_KEY")
    
    paths = [".meatbag_key", "backend/.meatbag_key", "../.meatbag_key"]
    for path in paths:
        if os.path.exists(path):
            try:
                with open(path, "r") as f:
                    return f.read().strip()
            except Exception:
                pass
    return None

def send_request(path, method="GET", payload=None, query_params=None):
    url = f"{BACKEND_URL}{path}"
    if query_params:
        params_str = urllib.parse.urlencode(query_params)
        url = f"{url}?{params_str}"
        
    headers = {}
    api_key = get_api_key()
    if api_key:
        headers["X-Meatbag-API-Key"] = api_key

    data = None
    if payload is not None:
        data = json.dumps(payload).encode("utf-8")
        headers["Content-Type"] = application_json = "application/json"

    req = urllib.request.Request(url, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(req) as response:
            res_body = response.read().decode("utf-8")
            if not res_body:
                return {}
            return json.loads(res_body)
    except urllib.error.HTTPError as e:
        err_msg = e.read().decode("utf-8")
        try:
            err_json = json.loads(err_msg)
            print(f"Error: {err_json.get('error', err_msg)}", file=sys.stderr)
        except Exception:
            print(f"Error: {e.code} - {err_msg}", file=sys.stderr)
        sys.exit(1)
    except Exception as e:
        print(f"Failed to connect to backend at {BACKEND_URL}: {str(e)}", file=sys.stderr)
        sys.exit(1)

def handle_ask(args):
    payload = {
        "type": "ApprovalRequest",
        "id": f"cli_{os.urandom(4).hex()}",
        "prompt": args.prompt,
        "target": args.target
    }
    query_params = {}
    if args.urgent:
        query_params["urgent"] = "true"
    if args.attention:
        query_params["attentionTier"] = args.attention

    print(f"Submitting approval request: '{args.prompt}'...", file=sys.stderr)
    result = send_request("/api/request", method="POST", payload=payload, query_params=query_params)
    approved = result.get("approved", False)
    
    print(f"Response: {'APPROVED' if approved else 'REJECTED'}", file=sys.stderr)
    if approved:
        sys.exit(0)
    else:
        sys.exit(1)

def handle_keys(args):
    if args.subcommand == "create":
        res = send_request("/api/keys", method="POST", query_params={"agentName": args.name})
        print(f"Created API key for {args.name}:")
        print(res.get("apiKey"))
    elif args.subcommand == "list":
        keys = send_request("/api/keys")
        print(f"{'ID':<40} | {'Agent Name':<20} | {'Status':<10}")
        print("-" * 76)
        for key in keys:
            status = "Revoked" if key.get("revoked") else "Active"
            print(f"{key.get('id'):<40} | {key.get('agentName'):<20} | {status:<10}")
    elif args.subcommand == "revoke":
        res = send_request(f"/api/keys/{args.id}", method="DELETE")
        print(res.get("status", "Revoked successfully"))

def handle_devices(args):
    if args.subcommand == "list":
        devices = send_request("/api/devices")
        print(f"{'Device ID':<40} | {'Type':<15} | {'Paired At':<20}")
        print("-" * 81)
        for dev in devices:
            import datetime
            paired_date = datetime.datetime.fromtimestamp(dev.get("pairedAtMs") / 1000.0).strftime('%Y-%m-%d %H:%M:%S')
            print(f"{dev.get('deviceId'):<40} | {dev.get('deviceType'):<15} | {paired_date:<20}")
    elif args.subcommand == "unpair":
        res = send_request(f"/api/devices/{args.id}", method="DELETE")
        print(res.get("status", "Unpaired successfully"))

def main():
    parser = argparse.ArgumentParser(description="meatbag CLI wrapper for human-in-the-loop approvals")
    subparsers = parser.add_subparsers(dest="command", required=True)

    # ask subcommand
    ask_parser = subparsers.add_parser("ask", help="Ask for human approval")
    ask_parser.add_argument("prompt", help="The approval prompt text")
    ask_parser.add_argument("--target", default="system-agent", help="The target tier/agent identifier")
    ask_parser.add_argument("--urgent", action="store_true", help="Mark the request as urgent")
    ask_parser.add_argument("--attention", choices=["SILENT_HAPTIC", "BACKGROUND_NOTIFICATION", "OVERRIDE_SCREEN"], help="Specify preferred attention level")

    # keys subcommand
    keys_parser = subparsers.add_parser("keys", help="Manage API keys")
    keys_sub = keys_parser.add_subparsers(dest="subcommand", required=True)
    
    create_key = keys_sub.add_parser("create", help="Create a new API key")
    create_key.add_argument("--name", required=True, help="Agent name for the key")
    
    keys_sub.add_parser("list", help="List API keys")
    
    revoke_key = keys_sub.add_parser("revoke", help="Revoke an API key")
    revoke_key.add_argument("id", help="The API key ID to revoke")

    # devices subcommand
    devices_parser = subparsers.add_parser("devices", help="Manage paired client devices")
    devices_sub = devices_parser.add_subparsers(dest="subcommand", required=True)
    
    devices_sub.add_parser("list", help="List paired devices")
    
    unpair_device = devices_sub.add_parser("unpair", help="Unpair a device")
    unpair_device.add_argument("id", help="The device ID to unpair")

    args = parser.parse_args()

    if args.command == "ask":
        handle_ask(args)
    elif args.command == "keys":
        handle_keys(args)
    elif args.command == "devices":
        handle_devices(args)

if __name__ == "__main__":
    main()