Jyotisha/scripts/vedastro_official_mcp_bridge.py

#!/usr/bin/env python3
"""Thin bridge to the official public VedAstro MCP endpoint.

This bridge is intentionally narrow. It proves that the official public MCP
surface is reachable and callable from the local workspace without letting
external responses silently override local adjudication.
"""

from __future__ import annotations

import argparse
import json
import urllib.error
import urllib.request
from typing import Any


DEFAULT_ENDPOINT = "https://mcp.vedastro.org/api/mcp/public"
DEFAULT_PROTOCOL_VERSION = "2025-06-18"
DEFAULT_TIMEOUT_SECONDS = 60


def _post_json(endpoint: str, payload: dict[str, Any], timeout: float = DEFAULT_TIMEOUT_SECONDS) -> tuple[dict[str, Any], dict[str, str]]:
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    req = urllib.request.Request(
        endpoint,
        data=body,
        method="POST",
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
        },
    )
    with urllib.request.urlopen(req, timeout=timeout) as resp:
        raw = resp.read().decode("utf-8")
        headers = {key.lower(): value for key, value in resp.headers.items()}
    return json.loads(raw), headers


def _initialize(endpoint: str) -> dict[str, Any]:
    payload = {
        "jsonrpc": "2.0",
        "id": 1,
        "method": "initialize",
        "params": {
            "protocolVersion": DEFAULT_PROTOCOL_VERSION,
            "capabilities": {},
            "clientInfo": {
                "name": "jyotish-local-bridge",
                "version": "1.0.0",
            },
        },
    }
    result, headers = _post_json(endpoint, payload)
    session_id = headers.get("mcp-session-id") or headers.get("x-mcp-session-id")
    return {
        "endpoint": endpoint,
        "available": True,
        "status": "ok",
        "operation": "initialize",
        "session_id": session_id,
        "result": result.get("result") or result,
        "source": "official_public_mcp",
    }


def _tools_list(endpoint: str) -> dict[str, Any]:
    init = _initialize(endpoint)
    payload = {
        "jsonrpc": "2.0",
        "id": 2,
        "method": "tools/list",
        "params": {},
    }
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
    }
    if init.get("session_id"):
        headers["Mcp-Session-Id"] = str(init["session_id"])
    req = urllib.request.Request(endpoint, data=body, method="POST", headers=headers)
    with urllib.request.urlopen(req, timeout=DEFAULT_TIMEOUT_SECONDS) as resp:
        raw = resp.read().decode("utf-8")
        response_headers = {key.lower(): value for key, value in resp.headers.items()}
    result = json.loads(raw)
    tools = (result.get("result") or {}).get("tools") or []
    return {
        "endpoint": endpoint,
        "available": True,
        "status": "ok",
        "operation": "tools_list",
        "session_id": init.get("session_id") or response_headers.get("mcp-session-id"),
        "tool_count": len(tools),
        "tool_names": [tool.get("name") for tool in tools if isinstance(tool, dict) and tool.get("name")],
        "result": result.get("result") or result,
        "source": "official_public_mcp",
    }


def _call_tool(endpoint: str, tool_name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    init = _initialize(endpoint)
    payload = {
        "jsonrpc": "2.0",
        "id": 3,
        "method": "tools/call",
        "params": {"name": tool_name, "arguments": arguments},
    }
    body = json.dumps(payload, ensure_ascii=False).encode("utf-8")
    headers = {
        "Content-Type": "application/json",
        "Accept": "application/json, text/event-stream",
    }
    if init.get("session_id"):
        headers["Mcp-Session-Id"] = str(init["session_id"])
    req = urllib.request.Request(endpoint, data=body, method="POST", headers=headers)
    with urllib.request.urlopen(req, timeout=DEFAULT_TIMEOUT_SECONDS) as resp:
        raw = resp.read().decode("utf-8")
        response_headers = {key.lower(): value for key, value in resp.headers.items()}
    result = json.loads(raw)
    return {
        "endpoint": endpoint,
        "available": True,
        "status": "ok",
        "operation": "call_tool",
        "tool_name": tool_name,
        "session_id": init.get("session_id") or response_headers.get("mcp-session-id"),
        "result": result.get("result") or result,
        "source": "official_public_mcp",
    }


def schema() -> dict[str, Any]:
    return {
        "bridge": "vedastro_official_mcp_bridge",
        "role": "official_public_mcp_thin_bridge",
        "endpoint": DEFAULT_ENDPOINT,
        "operations": ["initialize", "tools_list", "call_tool"],
        "response_contract": [
            "endpoint",
            "available",
            "status",
            "operation",
            "result",
            "source",
        ],
        "boundaries": [
            "Use for official MCP reachability, tool discovery, and targeted tool calls.",
            "Do not let raw MCP results directly override local score/dominant_label/payout_label.",
            "Promote only through explicit local contracts and tests.",
        ],
    }


def _error_result(endpoint: str, operation: str, exc: Exception) -> dict[str, Any]:
    reason = str(exc)
    if isinstance(exc, urllib.error.HTTPError):
        reason = f"http_{exc.code}: {reason}"
    return {
        "endpoint": endpoint,
        "available": False,
        "status": "mcp_request_failed",
        "operation": operation,
        "reason": reason,
        "source": "official_public_mcp",
    }


def main() -> int:
    parser = argparse.ArgumentParser(description="Official public VedAstro MCP bridge")
    parser.add_argument("--print-schema", action="store_true")
    parser.add_argument("--endpoint", default=DEFAULT_ENDPOINT)
    parser.add_argument("--operation", choices=["initialize", "tools_list", "call_tool"], default="tools_list")
    parser.add_argument("--tool")
    parser.add_argument("--arguments-json", default="{}")
    args = parser.parse_args()

    if args.print_schema:
        result = schema()
    else:
        try:
            if args.operation == "initialize":
                result = _initialize(args.endpoint)
            elif args.operation == "call_tool":
                if not args.tool:
                    raise ValueError("--tool is required when --operation call_tool")
                arguments = json.loads(args.arguments_json)
                if not isinstance(arguments, dict):
                    raise ValueError("--arguments-json must decode to an object")
                result = _call_tool(args.endpoint, args.tool, arguments)
            else:
                result = _tools_list(args.endpoint)
        except Exception as exc:  # noqa: BLE001
            result = _error_result(args.endpoint, args.operation, exc)

    print(json.dumps(result, ensure_ascii=False, indent=2, sort_keys=True))
    return 0


if __name__ == "__main__":
    raise SystemExit(main())