chris/OSVauco
1
0
Fork 0
Code Issues 1 Pull Requests Actions 2 Packages Projects Releases Wiki Activity
3b6e4e4890
OSVauco/opax-mcp/server.py
Chris Christiansen 3b6e4e4890 docs/code: clarify A2H2A prototype status and disable execution 
- Add Implementation Status section to A2H2A spec

- Clarify Google Chat is notification-only

- Add target approval architecture requirements

- Add critical A2H2A safety principle to RUNBOOK.md

- Remove unsafe approval/rejection endpoints from server.py

- Enforce server-side parameter_hash calculation

- Disable execution for all tools in A2H2A_TOOL_ALLOW_LIST
2026-09-04 12:00:07 +00:00

804 lines
41 KiB
Python
Raw Blame History

"""opax-mcp — MCP Streamable HTTP server for OPAX/Vauco
Transport: MCP Streamable HTTP (JSON-RPC 2.0) on POST /
Auth: Authorization: Bearer <secret> OR X-MCP-Secret: <secret> OR api-key: <secret>
"""
import os
import json
import uuid
import httpx
import base64
import google.auth
import google.auth.transport.requests
import google.oauth2.id_token
from googleapiclient.discovery import build
from google.cloud import secretmanager
from google.oauth2 import service_account
from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse, StreamingResponse
from fastapi.middleware.cors import CORSMiddleware
import asyncio
import secrets
# --- Tool Imports (with fault tolerance) ---
try:
from tyr.tools.scan_tyr_surface import scan_tyr_surface
except Exception as e:
print(f"Failed to load scan_tyr_surface: {e}")
scan_tyr_surface = None
try:
from tyr.tools.eval_tyr_identity import eval_tyr_identity
except Exception as e:
print(f"Failed to load eval_tyr_identity: {e}")
eval_tyr_identity = None
try:
from tyr.tools.get_tyr_forecast import get_tyr_forecast
except Exception as e:
print(f"Failed to load get_tyr_forecast: {e}")
get_tyr_forecast = None
try:
from tyr.tools.get_tyr_user_risk import get_tyr_user_risk
except Exception as e:
print(f"Failed to load get_tyr_user_risk: {e}")
get_tyr_user_risk = None
try:
from modules.factory.tools.provision_new_mcp_module import provision_new_mcp_module
except Exception as e:
print(f"Failed to load provision_new_mcp_module: {e}")
provision_new_mcp_module = None
from email.mime.text import MIMEText
from email.mime.text import MIMEText
from datetime import datetime, timezone, timedelta
from typing import Any, Optional, Dict, List
import logging
from pydantic import BaseModel, Field
import hashlib
# --- A2H2A Pydantic Models ---
class TicketSource(BaseModel):
reporter: str
trigger: str
affected_service: str
project_id: str
region: str
class TicketContext(BaseModel):
summary: str
evidence_logs: List[str]
class ProposedAction(BaseModel):
action_type: str
runbook_reference: str
execution_tool: str
parameters: Dict
parameter_hash: Optional[str] = None
rollback_plan: str
class Governance(BaseModel):
approval_status: str = "PENDING"
authorized_approver: str
requires_mfa: bool = True
timeout_minutes: int = 15
class A2H2ATicket(BaseModel):
ticket_id: str
timestamp: datetime
severity: str # CRITICAL, HIGH, MEDIUM, LOW
category: str
source: TicketSource
context: TicketContext
proposed_action: ProposedAction
governance: Governance
def calculate_parameter_hash(self) -> str:
"""Beregn SHA-256 hash av parameters i kanonisk form."""
param_json = json.dumps(self.proposed_action.parameters, sort_keys=True, separators=(",", ":"))
return hashlib.sha256(param_json.encode()).hexdigest()
def is_expired(self) -> bool:
"""Sjekk om ticket er utløpt"""
from datetime import datetime, timedelta
expiry = self.timestamp + timedelta(minutes=self.governance.timeout_minutes)
return datetime.utcnow() > expiry.replace(tzinfo=None)
class ConfirmationRequired(BaseModel):
message: str
preview: dict
requires_confirmation: bool = True
def require_confirmation(confirm: bool, preview: dict, message: str = "This action requires confirmation"):
if not confirm:
raise HTTPException(
status_code=400,
detail=ConfirmationRequired(message=message, preview=preview).model_dump()
)
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
app = FastAPI(title="opax-mcp", version="3.6.0") # BUMPED
app.add_middleware(
CORSMiddleware,
allow_origins=["https://opax-mcp-357036551735.us-central1.run.app"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
@app.get("/")
async def root_get():
return {"status": "healthy", "service": "OPAX MCP Server", "version": "3.6.0"}
async def sse_event_stream(request: Request):
"""Yields server-sent events for MCP protocol."""
# MCP requires an immediate connect event
yield f"id: {uuid.uuid4()}\nevent: connect\ndata: {json.dumps({'protocolVersion': '2024-11-05'})}\n\n"
while True:
# Send keepalive pings every 15s
yield f"id: {uuid.uuid4()}\nevent: ping\ndata: {json.dumps({'time': datetime.now(timezone.utc).isoformat()})}\n\n"
await asyncio.sleep(15)
@app.get("/sse")
async def sse_endpoint(request: Request):
"""Handles Server-Sent Events for external clients like Perplexity."""
return StreamingResponse(sse_event_stream(request), media_type="text/event-stream")
# --- A2H2A API Endpoints ---
# --- A2H2A Secure Design Placeholder ---
#
# The following logic is a placeholder for a secure A2H2A (Agent-to-Human-to-Agent)
# approval workflow. The key principles are:
#
# 1. NOTIFICATION, NOT ACTION: Google Chat messages are for notification only. They
# do not carry any identity or authorization guarantee.
# 2. VERIFIED IDENTITY: Human approval must happen through a protected surface that
# verifies the user's identity (e.g., a web app protected by Google IAP).
# 3. ATOMIC, IMMUTABLE ACTIONS: The ticket stored in the backend (Firestore) is the
# source of truth. The proposed action, its parameters, and the server-generated
# parameter hash are immutable.
# 4. ONE-TIME USE TOKENS: The approval UI should be accessed via a link containing a
# cryptographically random, single-use token that is invalidated after the first
# approval or rejection, or after the ticket's TTL expires.
# 5. DECOUPLED EXECUTION: The approval action only changes the ticket's state to
# 'APPROVED'. A separate, secure worker process will poll for approved tickets and
# execute them, ensuring a clean separation of concerns. This worker validates the
# ticket's integrity (non-expired, parameter hash match) before execution.
#
# This implementation only covers the safe recording of a proposed ticket.
# Execution is explicitly disabled.
# A strict allow-list of tools that can be proposed in an A2H2A ticket.
# `execution_enabled: False` means the tool can be proposed but not executed.
A2H2A_TOOL_ALLOW_LIST = {
"example_tool_1": {"execution_enabled": False, "description": "A sample tool that does something harmless."},
"example_tool_2": {"execution_enabled": False, "description": "Another sample tool."},
# Add real, vetted tools here with execution_enabled: True ONLY after extensive security review.
}
@app.post("/api/v1/a2h2a/tickets")
async def create_a2h2a_ticket(ticket: A2H2ATicket, request: Request):
"""
Opprett ny A2H2A ticket og send til Google Chat for godkjenning.
Dette er kun en prototype for sikker registrering, ingen verktøy blir kjørt.
"""
try:
await _verify_auth(request)
# Security check: Reject if caller provides a parameter_hash.
if ticket.proposed_action.parameter_hash is not None:
raise HTTPException(status_code=400, detail="Do not provide a parameter_hash. The server will generate it.")
# Security check: Ensure the proposed tool is in the allow-list.
tool_name = ticket.proposed_action.execution_tool
if tool_name not in A2H2A_TOOL_ALLOW_LIST:
raise HTTPException(status_code=400, detail=f"Tool '{tool_name}' is not on the approved A2H2A allow-list.")
# Generate parameter hash server-side for integrity.
ticket.proposed_action.parameter_hash = ticket.calculate_parameter_hash()
from google.cloud import firestore
db = firestore.Client(project=GOOGLE_CLOUD_PROJECT)
ticket_data = ticket.model_dump(mode='json')
ticket_data['created_at'] = datetime.utcnow().isoformat()
ticket_data['status'] = 'PENDING'
doc_ref = db.collection('a2h2a_tickets').document(ticket.ticket_id)
doc_ref.set(ticket_data)
# The approval URL now points to a conceptual, IAP-protected UI.
approval_ui_url = f"https://opax.vauco.no/a2h2a/review/{ticket.ticket_id}"
google_chat_webhook = os.getenv('A2H2A_GOOGLE_CHAT_WEBHOOK')
if google_chat_webhook:
chat_payload = {
"cards": [
{
"header": {
"title": f"🚨 A2H2A Proposal: {ticket.severity}",
"subtitle": "OSVxCC Security (Prototype - No Execution)",
"image": {"imageUrl": "https://www.gstatic.com/images/branding/googlelogo/2x/googlelogo_color_92x30dp.png"}
},
"sections": [
{
"widgets": [
{
"textParagraph": {
"text": f"<b>Category:</b> {ticket.category}<br>"
f"<b>Service:</b> {ticket.source.affected_service}<br>"
f"<b>Tool:</b> {ticket.proposed_action.execution_tool}<br>"
f"<b>Summary:</b> {ticket.context.summary}"
}
},
{
"buttons": [
{
"textButton": {
"text": "REVIEW IN APPROVAL UI",
"onClick": {"openLink": {"url": approval_ui_url}}
}
}
]
}
]
}
]
}
]
}
import httpx
async with httpx.AsyncClient() as client:
await client.post(google_chat_webhook, json=chat_payload)
return {"status": "success", "ticket_id": ticket.ticket_id, "parameter_hash": ticket.proposed_action.parameter_hash}
except Exception as e:
logger.error(f"Feil ved opprettelse av A2H2A ticket: {e}")
raise HTTPException(status_code=500, detail=str(e))
# ── Service discovery ───────────────────────────────────────────────────────
OSVAUCO_AGENT_URL = os.environ.get("OSVAUCO_AGENT_URL", "") # f.eks. https://osvauco-agent-....run.app
MCP_SECRET = os.environ.get("MCP_SECRET", "")
INTERNAL_API_KEY = os.environ.get("INTERNAL_API_KEY", "")
GITEA_URL = os.environ.get("GITEA_URL", "http://34.59.131.162:3000")
GITEA_TOKEN = os.environ.get("GITEA_TOKEN", "")
GITEA_REPO = os.environ.get("GITEA_REPO", "chris/OSVauco")
GOOGLE_CLOUD_PROJECT = os.environ.get("GOOGLE_CLOUD_PROJECT", "propane-will-491900-m5")
WORKSPACE_ADMIN_USER = "chris.christiansen@vauco.no"
# Emma-vm Ollama — direkte tilkobling
OLLAMA_BASE_URL = os.environ.get("OLLAMA_BASE_URL", "http://34.13.238.133:11434")
EMMA_MODEL = os.environ.get("EMMA_MODEL", "gemma3:27b")
EMMA_FAST_MODEL = os.environ.get("EMMA_FAST_MODEL", "gemma3:4b")
EMMA_LIGHT_MODEL = os.environ.get("EMMA_LIGHT_MODEL", "qwen2.5:3b")
logger.info(f"OSVAUCO_AGENT_URL: {OSVAUCO_AGENT_URL}")
logger.info(f"OLLAMA_BASE_URL: {OLLAMA_BASE_URL}")
# ---------------------------------------------------------------------------
# Auth (kun for innkommende kall til opax-mcp, f.eks. fra Gemini TUI)
# ---------------------------------------------------------------------------
async def _verify_auth(request: Request) -> None:
secret = os.getenv("MCP_SECRET")
if not secret:
logger.error("MCP_SECRET is not configured on the server.")
raise HTTPException(status_code=500, detail="MCP_SECRET not configured")
# Perplexity sends 'api-key', others might send 'X-MCP-Secret' or 'Authorization: Bearer ...'
token = request.headers.get("api-key")
if not token:
token = request.headers.get("X-MCP-Secret")
if not token:
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:]
if not token or not secrets.compare_digest(token.encode(), secret.encode()):
raise HTTPException(status_code=401, detail="Invalid or missing API key")
# ---------------------------------------------------------------------------
# Backend Agent helpers (for kall VIDERE til osvauco-agent)
# ---------------------------------------------------------------------------
def _agent_headers() -> dict:
"""Headers for maskin-til-maskin kall videre til osvauco-agent via X-Internal-Key."""
return {
"X-Internal-Key": INTERNAL_API_KEY, # Bruker den nye delte nøkkelen
"Content-Type": "application/json"
}
async def _agent_get(path: str) -> Any:
"""GET-kall til osvauco-agent."""
url = f"{OSVAUCO_AGENT_URL}{path}"
async with httpx.AsyncClient(timeout=30) as c:
r = await c.get(url, headers=_agent_headers())
r.raise_for_status()
return r.json()
async def _agent_post(path: str, body: dict) -> Any:
"""POST-kall til osvauco-agent."""
url = f"{OSVAUCO_AGENT_URL}{path}"
async with httpx.AsyncClient(timeout=45) as c:
r = await c.post(url, json=body, headers=_agent_headers())
r.raise_for_status()
return r.json()
# ---------------------------------------------------------------------------
# Google Workspace Helpers
# ---------------------------------------------------------------------------
def _get_workspace_service(api: str, version: str, scopes: list):
"""
Creates an authenticated Google API service object using a service account
key from Secret Manager and impersonating the admin user.
"""
try:
client = secretmanager.SecretManagerServiceClient()
secret_name = "workspace-sa-key"
version_name = f"projects/{GOOGLE_CLOUD_PROJECT}/secrets/{secret_name}/versions/latest"
response = client.access_secret_version(request={"name": version_name})
secret_payload = response.payload.data.decode("UTF-8")
secret_info = json.loads(secret_payload)
creds = service_account.Credentials.from_service_account_info(
secret_info, scopes=scopes
)
delegated_creds = creds.with_subject(WORKSPACE_ADMIN_USER)
return build(api, version, credentials=delegated_creds, cache_discovery=False)
except Exception as e:
logger.error(f"Failed to get Workspace service: {e}", exc_info=True)
raise
async def create_email_alias(p: dict) -> dict:
"""Creates a new email alias for a Google Workspace user."""
user_key = p.get("user_key")
alias_email = p.get("alias")
if not user_key or not alias_email:
raise ValueError("Missing required parameters: 'user_key' and 'alias'")
service = _get_workspace_service('admin', 'directory_v1', ['https://www.googleapis.com/auth/admin.directory.user'])
alias_body = {'alias': alias_email}
logger.info(f"Creating alias '{alias_email}' for user '{user_key}'...")
result = service.users().aliases().insert(userKey=user_key, body=alias_body).execute()
logger.info(f"Successfully created alias.")
return {"status": "success", "alias": result}
async def list_user_aliases(p: dict) -> dict:
"""Lists all aliases for a Google Workspace user."""
user_key = p.get("user_key")
if not user_key:
raise ValueError("Missing required parameter: 'user_key'")
service = _get_workspace_service('admin', 'directory_v1', ['https://www.googleapis.com/auth/admin.directory.user'])
logger.info(f"Listing aliases for user '{user_key}'...")
result = service.users().aliases().list(userKey=user_key).execute()
logger.info(f"Successfully listed aliases.")
return {"status": "success", "aliases": result.get("aliases", [])}
async def delete_email_alias(p: dict) -> dict:
"""Deletes an email alias from a Google Workspace user."""
user_key = p.get("user_key")
alias_email = p.get("alias")
if not user_key or not alias_email:
raise ValueError("Missing required parameters: 'user_key' and 'alias'")
service = _get_workspace_service('admin', 'directory_v1', ['https://www.googleapis.com/auth/admin.directory.user'])
logger.info(f"Deleting alias '{alias_email}' for user '{user_key}'...")
service.users().aliases().delete(userKey=user_key, alias=alias_email).execute()
logger.info(f"Successfully deleted alias.")
return {"status": "success", "detail": f"Alias '{alias_email}' deleted."}
async def send_email_as(p: dict) -> dict:
"""Sends an email as a user or their alias, on their behalf."""
from_address, to_address, subject, body = p.get("from"), p.get("to"), p.get("subject"), p.get("body")
if not all([from_address, to_address, subject, body]):
raise ValueError("Missing required parameters: 'from', 'to', 'subject', 'body'")
service = _get_workspace_service('gmail', 'v1', ['https://www.googleapis.com/auth/gmail.send'])
message = MIMEText(body)
message['to'], message['from'], message['subject'] = to_address, from_address, subject
encoded_message = base64.urlsafe_b64encode(message.as_bytes()).decode()
create_message = {'raw': encoded_message}
logger.info(f"Sending email from '{from_address}' to '{to_address}'...")
send_message = service.users().messages().send(userId='me', body=create_message).execute()
logger.info(f"Successfully sent email with ID: {send_message['id']}")
return {"status": "success", "message_id": send_message['id']}
async def get_workspace_user(p: dict) -> dict:
"""Gets detailed information about a single user in Google Workspace."""
user_key = p.get("user_key")
if not user_key: raise ValueError("Missing required parameter: 'user_key'")
service = _get_workspace_service('admin', 'directory_v1', ['https://www.googleapis.com/auth/admin.directory.user'])
logger.info(f"Getting info for user '{user_key}'...")
user = service.users().get(userKey=user_key).execute()
logger.info(f"Successfully retrieved user info.")
return {"name": user.get("name", {}).get("fullName"), "email": user.get("primaryEmail"), "aliases": user.get("aliases", []), "suspended": user.get("suspended"), "lastLoginTime": user.get("lastLoginTime")}
async def list_workspace_users(p: dict) -> dict:
"""Lists all users in the Google Workspace domain."""
service = _get_workspace_service('admin', 'directory_v1', ['https://www.googleapis.com/auth/admin.directory.user'])
logger.info("Listing all workspace users...")
result = service.users().list(domain='vauco.no', maxResults=50, orderBy='email').execute()
users = result.get('users', [])
logger.info(f"Found {len(users)} users.")
return {"users": [{"email": user.get("primaryEmail"), "name": user.get("name", {}).get("fullName")} for user in users]}
async def get_emails(p: dict) -> dict:
"""Searches a user's mailbox and retrieves a list of emails."""
user_key, query, max_results = p.get('user_key'), p.get('query', ''), p.get('max_results', 10)
if not user_key: raise ValueError("Missing required parameter: 'user_key'")
service = _get_workspace_service('gmail', 'v1', ['https://www.googleapis.com/auth/gmail.modify'])
logger.info(f"Searching emails for user '{user_key}' with query '{query}'...")
list_result = service.users().messages().list(userId=user_key, q=query, maxResults=max_results).execute()
messages = list_result.get('messages', [])
if not messages: return {"messages": []}
email_details = []
for msg in messages:
detail = service.users().messages().get(userId=user_key, id=msg['id'], format='metadata', metadataHeaders=['subject', 'from']).execute()
headers = {h['name']: h['value'] for h in detail['payload']['headers']}
email_details.append({"id": msg['id'], "snippet": detail.get('snippet'), "subject": headers.get('Subject'), "from": headers.get('From')})
logger.info(f"Successfully retrieved {len(email_details)} emails.")
return {"messages": email_details}
async def list_calendar_events(p: dict) -> dict:
"""Lists events from a user's calendar."""
user_key, days_ahead = p.get('user_key'), p.get('days_ahead', 7)
if not user_key: raise ValueError("Missing required parameter: 'user_key'")
service = _get_workspace_service('calendar', 'v3', ['https://www.googleapis.com/auth/calendar'])
logger.info(f"Fetching calendar events for '{user_key}' for the next {days_ahead} days...")
now, time_min = datetime.now(timezone.utc), (datetime.now(timezone.utc)).isoformat()
time_max = (now + timedelta(days=days_ahead)).isoformat()
events_result = service.events().list(calendarId=user_key, timeMin=time_min, timeMax=time_max, singleEvents=True, orderBy='startTime').execute()
events = events_result.get('items', [])
formatted_events = [{"summary": event.get('summary'), "start": event.get('start', {}).get('dateTime', event.get('start', {}).get('date')), "end": event.get('end', {}).get('dateTime', event.get('end', {}).get('date')), "organizer": event.get('organizer', {}).get('email')} for event in events]
logger.info(f"Found {len(formatted_events)} events.")
return {"events": formatted_events}
async def trigger_build(p: dict) -> dict:
"""Triggers a Cloud Build manually by its trigger ID."""
from google.cloud.devtools import cloudbuild_v1
project_id = GOOGLE_CLOUD_PROJECT
trigger_id = p.get("trigger_id")
branch = p.get("branch", "main")
substitutions = p.get("substitutions", {})
if not trigger_id:
raise ValueError("Missing required parameter: 'trigger_id'")
try:
client = cloudbuild_v1.CloudBuildClient()
source = cloudbuild_v1.RepoSource(branch_name=branch, substitutions=substitutions)
response = client.run_build_trigger(
project_id=project_id,
trigger_id=trigger_id,
source=source,
)
build_id = response.metadata.build.id
logger.info(f"Successfully triggered Cloud Build. Build ID: {build_id}")
return {"status": "success", "build_id": build_id}
except Exception as e:
logger.error(f"Failed to trigger Cloud Build: {e}", exc_info=True)
raise HTTPException(status_code=500, detail=f"Failed to trigger Cloud Build: {e}")
# ---------------------------------------------------------------------------
# Ollama helpers — direkte mot emma-gpu-vm
# ---------------------------------------------------------------------------
async def _ollama_chat(model: str, prompt: str, system: str = "") -> dict:
# ... (beholdt uendret)
messages = []
if system:
messages.append({"role": "system", "content": system})
messages.append({"role": "user", "content": prompt})
payload = { "model": model, "messages": messages, "stream": False }
try:
async with httpx.AsyncClient(timeout=120) as c:
r = await c.post(f"{OLLAMA_BASE_URL}/api/chat", json=payload)
r.raise_for_status()
data = r.json()
return { "model": model, "response": data.get("message", {}).get("content", ""), "done": data.get("done", True), "total_duration_ms": round(data.get("total_duration", 0) / 1e6) }
except Exception as e:
logger.error(f"[OLLAMA ERROR] model={model} {type(e).__name__}: {e}")
raise
async def _ollama_models() -> list:
async with httpx.AsyncClient(timeout=10) as c:
r = await c.get(f"{OLLAMA_BASE_URL}/api/tags")
r.raise_for_status()
return r.json().get("models", [])
# ---------------------------------------------------------------------------
# Gitea helpers
# ---------------------------------------------------------------------------
def _gitea_headers() -> dict:
return {"Authorization": f"token {GITEA_TOKEN}", "Content-Type": "application/json", "Accept": "application/json"}
async def _gitea_get(path: str) -> Any:
# ... (beholdt uendret)
async with httpx.AsyncClient(timeout=30) as c:
r = await c.get(f"{GITEA_URL}/api/v1{path}", headers=_gitea_headers())
r.raise_for_status()
return r.json()
async def _gitea_post(path: str, body: dict) -> Any:
# ... (beholdt uendret)
async with httpx.AsyncClient(timeout=30) as c:
r = await c.post(f"{GITEA_URL}/api/v1{path}", json=body, headers=_gitea_headers())
r.raise_for_status()
return r.json()
async def _gitea_put(path: str, body: dict) -> Any:
# ... (beholdt uendret)
async with httpx.AsyncClient(timeout=30) as c:
r = await c.put(f"{GITEA_URL}/api/v1{path}", json=body, headers=_gitea_headers())
r.raise_for_status()
return r.json()
# ---------------------------------------------------------------------------
# Tool implementations (refaktorert til å bruke _agent_get/_agent_post)
# ---------------------------------------------------------------------------
async def get_billing_summary(p): return await _agent_get("/billing/summary")
async def get_billing_forecast(p): return await _agent_get("/billing/forecast")
async def get_billing_credits(p): return await _agent_get("/billing/credits")
async def get_billing_anomalies(p): return await _agent_get("/billing/anomalies")
async def get_billing_history(p): return await _agent_get("/billing/history")
async def get_billing_budget(p): return await _agent_get("/billing/budget")
async def get_billing_tokens_by_module(p): return await _agent_get("/telemetry/history")
async def set_billing_budget(p): return await _agent_post("/billing/budget", {"budget": p.get("amount", 500)})
async def create_invite(p):
return await _agent_post("/onboard/invite", {
"company": p.get("company", p.get("name", "")), "email": p.get("email"), "tier": p.get("tier", "starter"),
})
async def list_customers(p): return await _agent_get("/state")
async def send_webhook(p):
return await _agent_post("/notify/webhook", {
"url": p.get("url", ""), "event": p.get("event", "custom"),
"title": p.get("title", "OPAX varsel"), "body": p.get("message", p.get("body", "")),
})
async def send_email(p): return await _agent_post("/notify/email", {"to": p.get("to"), "subject": p.get("subject"), "event": p.get("event", "digest")})
async def send_sms(p): return await _agent_post("/notify/sms", {"to": p.get("to"), "body": p.get("message", p.get("body", "")), "event": p.get("event", "spike"), "tier": p.get("tier", "guard")})
async def get_notify_channels(p): return await _agent_get("/notify/channels")
async def get_health(p): return await _agent_get("/health")
async def get_build_status(p): return await _agent_get("/opax/build-status")
async def get_state(p): return await _agent_get("/state")
async def get_telemetry(p): return await _agent_get("/telemetry/history")
async def run_terminal(p): return await _agent_post("/terminal/exec", {"cmd": p.get("command", p.get("cmd", "help"))})
async def tui_command(p): return await _agent_post("/tui-command", p)
# --- Uendrede funksjoner ---
async def run_jason(p):
"""Kaller /run på osvauco-agent, som nå har sin egen JASON_BACKEND-logikk."""
return await _agent_post("/run", {"message": p.get("prompt", p.get("message", "")), "user_id": p.get("user_id", "opax"), "session_id": p.get("session_id", "mcp"), "mode": p.get("mode", "light")})
async def run_emma(p): return await _ollama_chat("gemma3:4b", p.get("prompt", p.get("message", "")), p.get("system", "Du er Emma Vauger..."))
async def run_emma_fast(p): return await _ollama_chat(EMMA_FAST_MODEL, p.get("prompt", p.get("message", "")), p.get("system", "Du er en rask og konsis AI-assistent..."))
async def run_qwen(p): return await _ollama_chat(EMMA_LIGHT_MODEL, p.get("prompt", p.get("message", "")))
async def list_emma_models(p): return await _ollama_models()
async def list_commits(p): return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/commits?limit={p.get('limit', 10)}")
async def get_file(p): return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/contents/{p.get('path', '')}?ref={p.get('ref', 'main')}")
async def list_open_issues(p): return await _gitea_get(f"/repos/{p.get('repo', GITEA_REPO)}/issues?state=open&limit=20")
async def create_issue(p): return await _gitea_post(f"/repos/{p.get('repo', GITEA_REPO)}/issues", {"title": p.get("title"), "body": p.get("body", "")})
async def push_file(p):
# ... (beholdt uendret)
repo, path = p.get("repo", GITEA_REPO), p.get("path")
content = base64.b64encode(p.get("content", "").encode()).decode()
body = {"message": p.get("message", f"chore: update {path} via opax-mcp"), "content": content, "branch": p.get("branch", "main")}
if not p.get("sha"):
try:
existing = await _gitea_get(f"/repos/{repo}/contents/{path}?ref={p.get('branch','main')}")
body["sha"] = existing["sha"]
except Exception: pass
else: body["sha"] = p["sha"]
if "sha" in body: return await _gitea_put(f"/repos/{repo}/contents/{path}", body)
return await _gitea_post(f"/repos/{repo}/contents/{path}", body)
# ---------------------------------------------------------------------------
# Memory Bank & Deployment Tools
# ---------------------------------------------------------------------------
PROJECT_DIR = os.path.abspath("project")
PROD_BUILD_TRIGGER_ID = "cf42cbc7-a5e8-4724-a569-dbaac1b46a0b"
def _resolve_memory_path(file_name: str) -> str:
"""Resolves a file path and ensures it is within the project/ directory."""
if not file_name or '..' in file_name or file_name.startswith('/'):
raise ValueError("Invalid file name.")
abs_path = os.path.abspath(os.path.join(PROJECT_DIR, file_name))
if not abs_path.startswith(PROJECT_DIR):
raise ValueError("Security violation: Path is outside the allowed memory bank directory.")
return abs_path
async def read_memory_bank(p: dict) -> dict:
"""Reads content from a file in the memory bank (project/ directory)."""
file_name = p.get("file_name")
if not file_name:
raise ValueError("Missing required parameter: 'file_name'")
file_path = _resolve_memory_path(file_name)
try:
with open(file_path, 'r', encoding='utf-8') as f:
content = f.read()
return {"file_path": file_name, "content": content}
except FileNotFoundError:
raise ValueError(f"File not found: {file_name}")
except Exception as e:
logger.error(f"Error reading memory bank file {file_name}: {e}", exc_info=True)
raise
async def write_memory_bank(p: dict) -> dict:
"""Writes or overwrites content to a file in the memory bank (project/ directory)."""
file_name = p.get("file_name")
content = p.get("content")
if not file_name or content is None:
raise ValueError("Missing required parameters: 'file_name' and 'content'")
file_path = _resolve_memory_path(file_name)
try:
os.makedirs(PROJECT_DIR, exist_ok=True)
with open(file_path, 'w', encoding='utf-8') as f:
f.write(content)
logger.info(f"Successfully wrote to memory bank file: {file_name}")
return {"status": "success", "file_path": file_name}
except Exception as e:
logger.error(f"Error writing to memory bank file {file_name}: {e}", exc_info=True)
raise
async def build_and_deploy_service(p: dict) -> dict:
"""
Triggers the production build and deployment pipeline for the main service.
This action requires manual approval in the Google Cloud Console.
"""
branch_to_deploy = p.get("branch", "main")
logger.info(f"Triggering production deployment for branch: {branch_to_deploy}")
result = await trigger_build({
"trigger_id": PROD_BUILD_TRIGGER_ID,
"branch": branch_to_deploy,
"substitutions": p.get("substitutions", {})
})
return result
# ---------------------------------------------------------------------------
# Tool registry + MCP schema
# ---------------------------------------------------------------------------
TOOLS = {
# Billing
"get_billing_summary": (get_billing_summary, "Hent billing-sammendrag for OPAX", {}),
"get_billing_forecast": (
get_billing_forecast,
"Get billing forecast",
{},
),
"get_billing_credits": (
get_billing_credits,
"Get available billing credits",
{},
),
"get_billing_anomalies": (
get_billing_anomalies,
"Get billing anomalies",
{},
),
"get_billing_history": (
get_billing_history,
"Get billing history",
{},
),
"get_billing_budget": (
get_billing_budget,
"Get the configured billing budget",
{},
),
"get_telemetry": (
get_billing_tokens_by_module,
"Get token and telemetry history by module",
{},
),
"set_billing_budget": (set_billing_budget, "Sett månedlig budsjett", {"type":"object","properties":{"amount":{"type":"number"}}}),
# Onboarding
"create_invite": (create_invite, "Inviter ny kunde til OPAX", {"type":"object","properties":{"company":{"type":"string"},"email":{"type":"string"},"tier":{"type":"string"}},"required":["email"]}),
# Notifications
"send_webhook": (send_webhook, "Send webhook-varsling", {"type":"object","properties":{"url":{"type":"string"},"message":{"type":"string"}}}),
"send_email": (send_email, "Send e-post via ekstern tjeneste", {"type":"object","properties":{"to":{"type":"string"},"subject":{"type":"string"}},"required":["to","subject"]}),
"send_sms": (send_sms, "Send SMS", {"type":"object","properties":{"to":{"type":"string"},"message":{"type":"string"}},"required":["to","message"]}),
"get_notify_channels": (
get_notify_channels,
"List configured notification channels",
{},
),
# AI Agents
"run_jason": (run_jason, "Kjør Jason-agenten med en prompt", {"type":"object","properties":{"prompt":{"type":"string"},"mode":{"type":"string"}},"required":["prompt"]}),
"run_emma": (run_emma, "Emma Vauger (gemma3:27b) — primær lokal AI", {"type":"object","properties":{"prompt":{"type":"string"}}}),
# Local models — read-only discovery
"list_emma_models": (
list_emma_models,
"List locally available OSVx model names",
{},
),
# System & Ops
"get_health": (get_health, "Hent helsestatus for OPAX", {}),
"get_build_status": (get_build_status, "Hent siste build-status", {}),
"trigger_build": (trigger_build, "Trigger en Cloud Build manuelt", {"type":"object","properties":{"repo":{"type":"string"},"branch":{"type":"string"},"config":{"type":"string"}}}),
"get_state": (get_state, "Hent platform-tilstand", {}),
"list_customers": (list_customers, "List alle kunder (alias for get_state)", {}),
"run_terminal": (run_terminal, "Kjør terminalkommando på VM", {"type":"object","properties":{"command":{"type":"string"}}}),
# Gitea / VCS
"list_commits": (list_commits, "List siste commits i Gitea-repo", {}),
"get_file": (get_file, "Hent fil fra Gitea-repo", {"type":"object","properties":{"path":{"type":"string"}},"required":["path"]}),
# Google Workspace
"create_email_alias": (create_email_alias, "Opprett et nytt e-postalias", {"type":"object","properties":{"user_key":{"type":"string"},"alias":{"type":"string"}},"required":["user_key","alias"]}),
"list_user_aliases": (list_user_aliases, "List en brukers e-postaliaser", {"type":"object","properties":{"user_key":{"type":"string"}},"required":["user_key"]}),
"delete_email_alias": (delete_email_alias, "Slett et e-postalias", {"type":"object","properties":{"user_key":{"type":"string"},"alias":{"type":"string"}},"required":["user_key","alias"]}),
"send_email_as": (send_email_as, "Send en e-post på vegne av en bruker/alias", {"type":"object","properties":{"from":{"type":"string"},"to":{"type":"string"},"subject":{"type":"string"},"body":{"type":"string"}},"required":["from","to","subject","body"]}),
"get_workspace_user": (get_workspace_user, "Hent detaljer for en Workspace-bruker", {"type":"object","properties":{"user_key":{"type":"string"}},"required":["user_key"]}),
"list_workspace_users": (list_workspace_users, "List alle brukere i Workspace-domenet", {}),
"get_emails": (get_emails, "Søk i en brukers e-post", {"type":"object","properties":{"user_key":{"type":"string"},"query":{"type":"string"}},"required":["user_key"]}),
"list_calendar_events": (list_calendar_events, "List en brukers kalenderhendelser", {"type":"object","properties":{"user_key":{"type":"string"},"days_ahead":{"type":"integer"}},"required":["user_key"]}),
# Memory Bank
"read_memory_bank": (read_memory_bank, "Reads a file from the project memory bank.", {"type": "object", "properties": {"file_name": {"type": "string"}}, "required": ["file_name"]}),
"write_memory_bank": (write_memory_bank, "Writes a file to the project memory bank.", {"type": "object", "properties": {"file_name": {"type": "string"}, "content": {"type": "string"}}, "required": ["file_name", "content"]}),
"build_and_deploy_service": (build_and_deploy_service, "Triggers the production build & deployment pipeline.", {"type": "object", "properties": {"branch": {"type": "string"}}, "required": []}),
}
if provision_new_mcp_module:
TOOLS["provision_new_mcp_module"] = (provision_new_mcp_module, "Provisions a new MCP module", {"type":"object","properties":{"module_name":{"type":"string"},"tool_name":{"type":"string"},"tool_spec":{"type":"object"}},"required":["module_name","tool_name","tool_spec"]})
# ---------------------------------------------------------------------------
# JSON-RPC Endpoint
# ---------------------------------------------------------------------------
def _jsonrpc_ok(req_id, result):
return {"jsonrpc": "2.0", "id": req_id, "result": result}
def _jsonrpc_err(req_id, code, message):
return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}
@app.post("/")
async def mcp_handler(request: Request):
await _verify_auth(request)
try:
body = await request.json()
except Exception: return JSONResponse(_jsonrpc_err(None, -32700, "Parse error"), status_code=400)
method, req_id, params = body.get("method", ""), body.get("id"), body.get("params", {})
if method == "initialize":
return JSONResponse(_jsonrpc_ok(req_id, {"protocolVersion": "2024-11-05", "capabilities": {"tools": {}},"serverInfo": {"name": "opax-mcp", "version": "3.6.0"}})) # BUMPED
if method == "tools/list":
return JSONResponse(_jsonrpc_ok(req_id, {"tools": [
{"name": name, "description": desc, "inputSchema": schema if schema else {"type": "object", "properties": {}}}
for name, (_, desc, schema) in TOOLS.items()
]}))
if method == "tools/call":
tool_name, tool_args = params.get("name") or params.get("tool"), params.get("arguments", params.get("params", {}))
entry = TOOLS.get(tool_name)
if not entry: return JSONResponse(_jsonrpc_err(req_id, -32601, f"Unknown tool: {tool_name}"))
handler, _, _ = entry
try:
result = await handler(tool_args)
return JSONResponse(_jsonrpc_ok(req_id, {"content": [{"type": "text", "text": json.dumps(result, ensure_ascii=False)}]}))
except Exception as e:
logger.error(f"[MCP HANDLER ERROR] tool={tool_name} {type(e).__name__}: {e}", exc_info=True)
return JSONResponse(_jsonrpc_err(req_id, -32000, str(e)))
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)
# ---------------------------------------------------------------------------
# Health (keepalive)
# ---------------------------------------------------------------------------
@app.get("/health")
async def health():
return {"status": "ok", "service": "opax-mcp", "version": "3.6.0", "ollama": OLLAMA_BASE_URL} # BUMPED
Reference in New Issue View Git Blame Copy Permalink