meatbag/mcp_server.py

import sys
import json
import urllib.request
import urllib.error
import uuid

def log(msg):
    sys.stderr.write(f"[Meatbag MCP] {msg}\n")
    sys.stderr.flush()

import os

def get_api_key():
    if os.environ.get("MEATBAG_API_KEY"):
        return os.environ.get("MEATBAG_API_KEY")
    
    script_dir = os.path.dirname(os.path.abspath(__file__))
    paths = [
        os.path.join(script_dir, ".meatbag_key"),
        os.path.join(script_dir, "backend", ".meatbag_key"),
        os.path.join(script_dir, "..", ".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 handle_request_approval(prompt, target):
    log(f"Handling approval request. Target: {target}, Prompt: {prompt}")
    request_id = f"mcp_{uuid.uuid4().hex[:8]}"
    
    payload = {
        "type": "ApprovalRequest",
        "id": request_id,
        "prompt": prompt,
        "target": target
    }
    
    headers = {"Content-Type": "application/json"}
    api_key = get_api_key()
    if api_key:
        headers["X-Meatbag-API-Key"] = api_key
        log(f"Using API Key: {api_key[:6]}...")
    else:
        log("Warning: No API key found.")
        
    try:
        url = "http://localhost:8080/api/request?attentionTier=OVERRIDE_SCREEN"
        log(f"Sending POST to {url}")
        req = urllib.request.Request(
            url,
            data=json.dumps(payload).encode("utf-8"),
            headers=headers
        )
        with urllib.request.urlopen(req) as response:
            res = json.loads(response.read().decode("utf-8"))
            log(f"Response received from backend: {res}")
            approved = res.get("approved", False)
            return {
                "content": [{"type": "text", "text": f"approved: {approved}"}],
                "isError": False
            }
        except Exception as e:
            log(f"Error communicating with Meatbag backend: {str(e)}")
            return {
                "content": [{"type": "text", "text": f"Error contacting Meatbag backend: {str(e)}"}],
                "isError": True
            }

def main():
    log("Starting Meatbag MCP Server...")
    while True:
        try:
            line = sys.stdin.readline()
            if not line:
                break
            
            req = json.loads(line)
            method = req.get("method")
            req_id = req.get("id")
            
            log(f"Received request: {method} (id: {req_id})")
            
            if method == "initialize":
                res = {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "protocolVersion": "2024-11-05",
                        "capabilities": {
                            "tools": {}
                        },
                        "serverInfo": {
                            "name": "meatbag-mcp",
                            "version": "1.0.0"
                        }
                    }
                }
            elif method == "tools/list":
                res = {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "result": {
                        "tools": [
                          {
                            "name": "request_approval",
                            "description": "Request human-in-the-loop approval on the user's phone/watch before executing a high-risk action. Blocks until the user responds on their device.",
                            "inputSchema": {
                              "type": "object",
                              "properties": {
                                "prompt": {
                                  "type": "string",
                                  "description": "The description of the action being requested (e.g. 'Delete prod database')"
                                },
                                "target": {
                                  "type": "string",
                                  "description": "The system or resource target (e.g. 'prod-db')"
                                }
                              },
                              "required": ["prompt", "target"]
                            }
                          }
                        ]
                    }
                }
            elif method == "tools/call":
                params = req.get("params", {})
                tool_name = params.get("name")
                args = params.get("arguments", {})
                
                if tool_name == "request_approval":
                    result = handle_request_approval(args.get("prompt"), args.get("target"))
                    res = {
                        "jsonrpc": "2.0",
                        "id": req_id,
                        "result": result
                    }
                else:
                    res = {
                        "jsonrpc": "2.0",
                        "id": req_id,
                        "error": {
                            "code": -32601,
                            "message": f"Method not found: {tool_name}"
                        }
                    }
            elif method.startswith("notifications/"):
                # Handle notification without response
                continue
            else:
                res = {
                    "jsonrpc": "2.0",
                    "id": req_id,
                    "error": {
                        "code": -32601,
                        "message": f"Method not found: {method}"
                    }
                }
            
            # Send response to stdout
            sys.stdout.write(json.dumps(res) + "\n")
            sys.stdout.flush()
            
        except Exception as e:
            log(f"Exception in message loop: {str(e)}")

if __name__ == "__main__":
    main()