144 lines
5.1 KiB
Python
144 lines
5.1 KiB
Python
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()
|
|
|
|
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
|
|
}
|
|
|
|
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={"Content-Type": "application/json"}
|
|
)
|
|
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()
|