mcp.py

from __future__ import annotations

import atexit
import os
import shutil
import socket
import subprocess
import sys
import tempfile
import time
from dataclasses import dataclass, field
from pathlib import Path

import httpx


OPENDOTA_MCP_HTTP_URL = "https://opendota-mcp-server-jylza6gata-ew.a.run.app/mcp"
LOCAL_OPENDOTA_MCP_HOST = "127.0.0.1"
LOCAL_OPENDOTA_MCP_PORT = int(os.getenv("OPENDOTA_MCP_PORT", "8080"))
_managed_process: subprocess.Popen[bytes] | None = None
_stderr_log: tempfile._TemporaryFileWrapper[bytes] | None = None


@dataclass(frozen=True)
class OpenDotaMCPServer:
    """Configuration for the OpenDota MCP server.

    The repo intentionally owns this file as `mcp.py`, so the agent uses a
    lightweight JSON-RPC client instead of importing the Python package named
    `mcp`, which would collide with this module name.
    """

    transport: str = field(default_factory=lambda: os.getenv("OPENDOTA_MCP_TRANSPORT", "http"))
    url: str = field(default_factory=lambda: os.getenv("OPENDOTA_MCP_URL", OPENDOTA_MCP_HTTP_URL))

    @property
    def is_http(self) -> bool:
        return self.transport.lower() == "http"


def initialize_opendota_mcp_server() -> OpenDotaMCPServer:
    """Configure the OpenDota MCP HTTP server for the Gradio app.

    The Hugging Face Space uses the hosted OpenDota MCP server by default.
    Set OPENDOTA_MCP_LOCAL=1 to launch a local subprocess instead.
    """

    use_local = os.getenv("OPENDOTA_MCP_LOCAL", "").lower() in {"1", "true", "yes", "on"}
    if os.getenv("OPENDOTA_MCP_URL") or not use_local:
        return OpenDotaMCPServer()

    port = _available_port(LOCAL_OPENDOTA_MCP_PORT)
    url = f"http://{LOCAL_OPENDOTA_MCP_HOST}:{port}/mcp"
    os.environ["OPENDOTA_MCP_TRANSPORT"] = "http"
    os.environ["OPENDOTA_MCP_URL"] = url

    if _healthcheck(port):
        return OpenDotaMCPServer()

    command = _opendota_command()
    env = os.environ.copy()
    env["MCP_TRANSPORT"] = "http"
    env["PORT"] = str(port)
    if os.getenv("OPENDOTA_API_KEY"):
        env["OPENDOTA_API_KEY"] = os.environ["OPENDOTA_API_KEY"]

    global _managed_process, _stderr_log
    _stderr_log = tempfile.NamedTemporaryFile(prefix="opendota-mcp-", suffix=".log")
    _managed_process = subprocess.Popen(
        command,
        cwd="/tmp",
        env=env,
        stdout=subprocess.DEVNULL,
        stderr=_stderr_log,
    )
    atexit.register(shutdown_opendota_mcp_server)
    _wait_for_server(port)
    return OpenDotaMCPServer()


def shutdown_opendota_mcp_server() -> None:
    global _managed_process, _stderr_log
    if _managed_process is None or _managed_process.poll() is not None:
        _managed_process = None
        if _stderr_log is not None:
            _stderr_log.close()
            _stderr_log = None
        return
    _managed_process.terminate()
    try:
        _managed_process.wait(timeout=5)
    except subprocess.TimeoutExpired:
        _managed_process.kill()
        _managed_process.wait(timeout=5)
    _managed_process = None
    if _stderr_log is not None:
        _stderr_log.close()
        _stderr_log = None


def _opendota_command() -> list[str]:
    configured_command = os.getenv("OPENDOTA_MCP_COMMAND", "")
    if configured_command:
        return [configured_command, *_split_args(os.getenv("OPENDOTA_MCP_ARGS", ""))]

    executable = shutil.which("opendota-mcp")
    if executable:
        return [executable]

    venv_executable = Path(sys.executable).with_name("opendota-mcp")
    if venv_executable.exists():
        return [str(venv_executable)]

    return [sys.executable, "-m", "opendota_mcp.server"]


def _split_args(raw_args: str) -> list[str]:
    return [item for item in raw_args.split(" ") if item]


def _available_port(preferred_port: int) -> int:
    if _port_is_available(preferred_port):
        return preferred_port
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        sock.bind((LOCAL_OPENDOTA_MCP_HOST, 0))
        return int(sock.getsockname()[1])


def _port_is_available(port: int) -> bool:
    with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
        return sock.connect_ex((LOCAL_OPENDOTA_MCP_HOST, port)) != 0


def _healthcheck(port: int) -> bool:
    try:
        response = httpx.get(
            f"http://{LOCAL_OPENDOTA_MCP_HOST}:{port}/health",
            timeout=0.5,
        )
        return response.status_code == 200
    except httpx.HTTPError:
        return False


def _wait_for_server(port: int) -> None:
    deadline = time.monotonic() + 30
    while time.monotonic() < deadline:
        if _managed_process is not None and _managed_process.poll() is not None:
            stderr = _read_stderr_log()
            raise RuntimeError(f"OpenDota MCP server exited during startup. {stderr}")
        if _healthcheck(port):
            return
        time.sleep(0.25)
    raise TimeoutError("Timed out waiting for the OpenDota MCP server to start.")


def _read_stderr_log() -> str:
    if _stderr_log is None:
        return ""
    _stderr_log.flush()
    with open(_stderr_log.name, "rb") as log_file:
        return log_file.read().decode("utf-8", errors="replace")