chore: update opax-mcp/server.py via opax-mcp
Some checks failed
Check Python Version Consistency / Check Python Version (push) Has been cancelled

This commit is contained in:
chris 2026-07-22 02:47:29 +00:00
parent 4c4b582e25
commit 3fe6cfe5d5

View File

@ -4,7 +4,6 @@ Auth: Authorization: Bearer <secret> OR X-MCP-Secret: <secret> OR api-key: <
"""
import os
import json
import uuid
import httpx
import base64
import google.auth
@ -16,7 +15,6 @@ from google.oauth2 import service_account
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
from email.mime.text import MIMEText
from email.mime.text import MIMEText
from datetime import datetime, timezone, timedelta
from typing import Any, Optional
import logging
@ -662,35 +660,6 @@ async def mcp_handler(request: Request):
if method.startswith("notifications/"): return JSONResponse(status_code=202, content={})
return JSONResponse(_jsonrpc_err(req_id, -32601, f"Method not found: {method}"), status_code=404)
async def start_gce_instance(p: dict) -> tuple:
"""Starts a GCE instance. Assumes us-central1-b if zone is not specified."""
import subprocess
instance = p.get("instance")
zone = p.get("zone", "us-central1-b") # Default zone assumption
if not instance: raise ValueError("Missing required parameter: 'instance'")
result = subprocess.run(["gcloud", "compute", "instances", "start", instance, f"--zone={zone}", f"--project={GOOGLE_CLOUD_PROJECT}"], capture_output=True, text=True, check=False)
if result.returncode != 0: return ({"error": result.stderr}, {"status": "Error", "next_action": "Check instance name and permissions."})
raw_data = {"status": f"Start command issued for {instance}."}
custom_meta = {"status": f"Start-kommando sendt til '{instance}'.", "next_action": "Vent 30s og sjekk status med `list_gce_instances`.", "choices": [{"id": f"list_gce_instances", "label": "Sjekk status nå", "style": "primary"}]}
return (raw_data, custom_meta)
async def stop_gce_instance(p: dict) -> tuple:
"""Stops a GCE instance. Assumes us-central1-b if zone is not specified."""
import subprocess
instance = p.get("instance")
zone = p.get("zone", "us-central1-b") # Default zone assumption
if not instance: raise ValueError("Missing required parameter: 'instance'")
result = subprocess.run(["gcloud", "compute", "instances", "stop", instance, f"--zone={zone}", f"--project={GOOGLE_CLOUD_PROJECT}"], capture_output=True, text=True, check=False)
if result.returncode != 0: return ({"error": result.stderr}, {"status": "Error", "next_action": "Check instance name and permissions."})
raw_data = {"status": f"Stop command issued for {instance}."}
custom_meta = {"status": f"Stopp-kommando sendt til '{instance}'.", "next_action": "Vent 30s og sjekk status med `list_gce_instances`.", "choices": [{"id": f"list_gce_instances", "label": "Sjekk status nå", "style": "primary"}]}
return (raw_data, custom_meta)
# ---------------------------------------------------------------------------
# Health (keepalive)
# ---------------------------------------------------------------------------