quality-inspection-platform/mcp_bridge.py

"""Stdio MCP server that bridges to the local HTTP MCP gateway.

It speaks the Model Context Protocol over stdin/stdout (newline-delimited
JSON-RPC 2.0) and forwards every tool call to:

    GET  http://127.0.0.1:8000/mcp/tools
    POST http://127.0.0.1:8000/mcp/invoke

Run it with Cursor / Codex / Claude Desktop by configuring stdio MCP entry:

    {
      "command": "python3",
      "args": ["/abs/path/to/mcp_bridge.py"],
      "env": {"AI_TEST_BASE_URL": "http://127.0.0.1:8000"}
    }
"""

from __future__ import annotations

import json
import os
import sys
from typing import Any

import httpx

BASE_URL = os.environ.get("AI_TEST_BASE_URL", "http://127.0.0.1:8000").rstrip("/")
API_KEY = os.environ.get("AI_TEST_API_KEY", "").strip()
SERVER_NAME = "quality-inspection-platform"
SERVER_VERSION = "0.1.0"
PROTOCOL_VERSION = "2024-11-05"


def log(msg: str) -> None:
    sys.stderr.write(f"[mcp_bridge] {msg}\n")
    sys.stderr.flush()


def write_message(message: dict[str, Any]) -> None:
    sys.stdout.write(json.dumps(message, ensure_ascii=False) + "\n")
    sys.stdout.flush()


def fetch_tools() -> list[dict[str, Any]]:
    try:
        response = httpx.get(f"{BASE_URL}/mcp/tools", headers=_request_headers(), timeout=10.0)
        response.raise_for_status()
        payload = response.json() or {}
        tools = payload.get("tools", [])
    except Exception as exc:
        log(f"fetch tools failed: {exc}")
        return []

    converted = []
    for tool in tools:
        converted.append(
            {
                "name": tool.get("name", ""),
                "description": tool.get("description", ""),
                "inputSchema": tool.get("input_schema") or {"type": "object"},
            }
        )
    return converted


def _request_headers() -> dict[str, str]:
    headers = {"Content-Type": "application/json"}
    if API_KEY:
        headers["Authorization"] = f"Bearer {API_KEY}"
    return headers


def call_tool(name: str, arguments: dict[str, Any]) -> dict[str, Any]:
    payload: dict[str, Any] = {"tool": name, "arguments": arguments or {}}
    if API_KEY:
        payload["api_key"] = API_KEY
    try:
        response = httpx.post(
            f"{BASE_URL}/mcp/invoke",
            json=payload,
            headers=_request_headers(),
            timeout=60.0,
        )
        response.raise_for_status()
        return response.json() or {}
    except Exception as exc:
        return {"ok": False, "tool": name, "data": {}, "error": str(exc)}


def make_response(req_id: Any, result: Any) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": req_id, "result": result}


def make_error(req_id: Any, code: int, message: str) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}


def handle_initialize(req_id: Any, _: dict[str, Any]) -> dict[str, Any]:
    return make_response(
        req_id,
        {
            "protocolVersion": PROTOCOL_VERSION,
            "capabilities": {"tools": {"listChanged": False}},
            "serverInfo": {"name": SERVER_NAME, "version": SERVER_VERSION},
            "instructions": (
                "AI Auto API Test platform. Before complex tasks, read project file "
                "docs/mcp_tools_for_ai.md (tool picker, auth, recipes). "
                "Typical start: catalog_snapshot. "
                "Batch runs: workflow_batch_create → workflow_batch_update → workflow_batch_run. "
                "Writes and execution require sto- API Key (AI_TEST_API_KEY)."
            ),
        },
    )


def handle_tools_list(req_id: Any, _: dict[str, Any]) -> dict[str, Any]:
    return make_response(req_id, {"tools": fetch_tools()})


def handle_tools_call(req_id: Any, params: dict[str, Any]) -> dict[str, Any]:
    name = str(params.get("name", ""))
    arguments = params.get("arguments") or {}
    if not name:
        return make_error(req_id, -32602, "missing tool name")

    payload = call_tool(name, arguments)
    text = json.dumps(payload, ensure_ascii=False, indent=2)
    is_error = not bool(payload.get("ok", True))
    return make_response(
        req_id,
        {
            "content": [{"type": "text", "text": text}],
            "isError": is_error,
        },
    )


HANDLERS = {
    "initialize": handle_initialize,
    "tools/list": handle_tools_list,
    "tools/call": handle_tools_call,
}


def main() -> None:
    log(f"started, base_url={BASE_URL}")
    for raw_line in sys.stdin:
        line = raw_line.strip()
        if not line:
            continue
        try:
            message = json.loads(line)
        except json.JSONDecodeError as exc:
            log(f"invalid json: {exc}")
            continue

        method = message.get("method")
        req_id = message.get("id")
        params = message.get("params") or {}

        if method is None:
            continue
        if "id" not in message:
            # JSON-RPC notification, no response required.
            continue

        handler = HANDLERS.get(method)
        if handler is None:
            write_message(make_error(req_id, -32601, f"method not found: {method}"))
            continue

        try:
            response = handler(req_id, params)
        except Exception as exc:
            log(f"handler error: {exc}")
            response = make_error(req_id, -32000, str(exc))
        write_message(response)


if __name__ == "__main__":
    main()