Chris Christiansen
86551f2def
Some checks are pending
Check Python Version Consistency / Check Python Version (push) Waiting to run
108 lines
5.0 KiB
Python
108 lines
5.0 KiB
Python
"""Bakoverkompatibel MCP Streamable HTTP-klient for opax-mcp."""
|
|
import json
|
|
import os
|
|
import httpx
|
|
|
|
OPAX_MCP_URL = os.environ.get(
|
|
"MCP_SERVER_URL",
|
|
"https://opax-mcp-357036551735.us-central1.run.app",
|
|
).rstrip("/")
|
|
MCP_SECRET = os.environ.get("MCP_SECRET", "")
|
|
|
|
def _headers(session_id=None):
|
|
h = {
|
|
"Authorization": f"Bearer {MCP_SECRET}",
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json, text/event-stream",
|
|
"MCP-Protocol-Version": "2025-06-18",
|
|
}
|
|
if session_id:
|
|
h["Mcp-Session-Id"] = session_id
|
|
return h
|
|
|
|
def _decode(response):
|
|
raw = response.text
|
|
rows = [line[5:].strip() for line in raw.splitlines() if line.startswith("data:")]
|
|
return json.loads(rows[-1] if rows else raw)
|
|
|
|
def _post(payload, session_id=None):
|
|
with httpx.Client(timeout=60) as client:
|
|
r = client.post(OPAX_MCP_URL, headers=_headers(session_id), json=payload)
|
|
r.raise_for_status()
|
|
return _decode(r), r.headers.get("Mcp-Session-Id")
|
|
|
|
def _session():
|
|
_, sid = _post({
|
|
"jsonrpc": "2.0",
|
|
"id": 1,
|
|
"method": "initialize",
|
|
"params": {
|
|
"protocolVersion": "2025-06-18",
|
|
"capabilities": {},
|
|
"clientInfo": {"name": "osvxcc", "version": "1.0"},
|
|
},
|
|
})
|
|
_post({"jsonrpc": "2.0", "method": "notifications/initialized"}, sid)
|
|
return sid
|
|
|
|
def call_tool(tool: str, params: dict = None) -> dict:
|
|
sid = _session()
|
|
reply, _ = _post({
|
|
"jsonrpc": "2.0",
|
|
"id": 3,
|
|
"method": "tools/call",
|
|
"params": {"name": tool, "arguments": params or {}},
|
|
}, sid)
|
|
return reply.get("result", reply)
|
|
|
|
def list_tools() -> list:
|
|
sid = _session()
|
|
reply, _ = _post({
|
|
"jsonrpc": "2.0",
|
|
"id": 2,
|
|
"method": "tools/list",
|
|
"params": {},
|
|
}, sid)
|
|
return reply.get("result", {}).get("tools", [])
|
|
|
|
def get_billing_summary() -> dict: return call_tool("get_billing_summary")
|
|
def get_billing_forecast() -> dict: return call_tool("get_billing_forecast")
|
|
def get_billing_credits() -> dict: return call_tool("get_billing_credits")
|
|
def get_billing_anomalies() -> dict: return call_tool("get_billing_anomalies")
|
|
def get_billing_history() -> dict: return call_tool("get_billing_history")
|
|
def get_billing_budget() -> dict: return call_tool("get_billing_budget")
|
|
def set_billing_budget(amount: float, currency: str = "USD") -> dict: return call_tool("set_billing_budget", {"amount": amount, "currency": currency})
|
|
def create_invite(email: str, name: str = "", tier: str = "starter") -> dict: return call_tool("create_invite", {"email": email, "name": name, "tier": tier})
|
|
def list_customers() -> dict: return call_tool("list_customers")
|
|
def send_webhook(message: str, url: str = "") -> dict: return call_tool("send_webhook", {"message": message, "url": url})
|
|
def send_email(to: str, subject: str, body: str) -> dict: return call_tool("send_email", {"to": to, "subject": subject, "body": body})
|
|
def send_sms(to: str, message: str) -> dict: return call_tool("send_sms", {"to": to, "message": message})
|
|
def get_notify_channels() -> dict: return call_tool("get_notify_channels")
|
|
def run_jason(prompt: str) -> dict: return call_tool("run_jason", {"prompt": prompt})
|
|
def run_emma(prompt: str) -> dict: return call_tool("run_emma", {"prompt": prompt})
|
|
def get_health() -> dict: return call_tool("get_health")
|
|
def get_build_status() -> dict: return call_tool("get_build_status")
|
|
def get_state() -> dict: return call_tool("get_state")
|
|
def get_telemetry() -> dict: return call_tool("get_telemetry")
|
|
def run_terminal(command: str) -> dict: return call_tool("run_terminal", {"command": command})
|
|
def list_commits(repo: str = "", limit: int = 10) -> dict: return call_tool("list_commits", {"repo": repo, "limit": limit})
|
|
def get_file(path: str, repo: str = "") -> dict: return call_tool("get_file", {"path": path, "repo": repo})
|
|
def list_open_issues(repo: str = "") -> dict: return call_tool("list_open_issues", {"repo": repo})
|
|
def create_github_issue(title: str, body: str = "", repo: str = "") -> dict: return call_tool("create_github_issue", {"title": title, "body": body, "repo": repo})
|
|
def push_file(path: str, content: str, message: str = "", sha: str = "", repo: str = "") -> dict: return call_tool("push_file", {"path": path, "content": content, "message": message, "sha": sha, "repo": repo})
|
|
|
|
def get_all_function_tools() -> list:
|
|
try:
|
|
from google.adk.tools import FunctionTool
|
|
return [FunctionTool(func=f) for f in [
|
|
get_billing_summary, get_billing_forecast, get_billing_credits,
|
|
get_billing_anomalies, get_billing_history, get_billing_budget,
|
|
set_billing_budget, create_invite, list_customers, send_webhook,
|
|
send_email, send_sms, get_notify_channels, run_jason, run_emma,
|
|
get_health, get_build_status, get_state, get_telemetry,
|
|
run_terminal, list_commits, get_file, list_open_issues,
|
|
create_github_issue, push_file,
|
|
]]
|
|
except ImportError:
|
|
return []
|