Cerbero-Bite/src/cerbero_bite/config/mcp_endpoints.py

"""Resolve MCP service URLs and the bearer token.

Cerbero Bite runs in its own Docker container that joins the
``cerbero-suite`` network: every MCP service is reachable by the
container DNS name plus its internal port (``mcp-deribit:9011`` etc.).

The resolver supports two layers of override:

1. Per-service environment variables (``CERBERO_BITE_MCP_DERIBIT_URL``,
   ``CERBERO_BITE_MCP_MACRO_URL``…). Useful for dev when running
   outside Docker — point at ``http://localhost:9011`` etc.
2. ``CERBERO_BITE_CORE_TOKEN_FILE`` env var: path to the file that
   stores the bearer token (default
   ``/run/secrets/core_token``). The file is read at boot, the
   trailing whitespace is stripped, and the value is *not* logged.
"""

from __future__ import annotations

import os
from dataclasses import dataclass
from pathlib import Path

__all__ = [
    "DEFAULT_ENDPOINTS",
    "MCP_SERVICES",
    "McpEndpoints",
    "load_endpoints",
    "load_token",
]


# Service identifier → (default Docker DNS host, default port, env var name)
MCP_SERVICES: dict[str, tuple[str, int, str]] = {
    "deribit": ("mcp-deribit", 9011, "CERBERO_BITE_MCP_DERIBIT_URL"),
    "hyperliquid": ("mcp-hyperliquid", 9012, "CERBERO_BITE_MCP_HYPERLIQUID_URL"),
    "macro": ("mcp-macro", 9013, "CERBERO_BITE_MCP_MACRO_URL"),
    "sentiment": ("mcp-sentiment", 9014, "CERBERO_BITE_MCP_SENTIMENT_URL"),
    "telegram": ("mcp-telegram", 9017, "CERBERO_BITE_MCP_TELEGRAM_URL"),
    "portfolio": ("mcp-portfolio", 9018, "CERBERO_BITE_MCP_PORTFOLIO_URL"),
}


def _default_url(host: str, port: int) -> str:
    return f"http://{host}:{port}"


DEFAULT_ENDPOINTS: dict[str, str] = {
    name: _default_url(host, port) for name, (host, port, _) in MCP_SERVICES.items()
}


@dataclass(frozen=True)
class McpEndpoints:
    """Resolved per-service URLs."""

    deribit: str
    hyperliquid: str
    macro: str
    sentiment: str
    telegram: str
    portfolio: str

    def for_service(self, name: str) -> str:
        try:
            return getattr(self, name)  # type: ignore[no-any-return]
        except AttributeError as exc:
            raise KeyError(f"unknown MCP service '{name}'") from exc


def load_endpoints(env: dict[str, str] | None = None) -> McpEndpoints:
    """Build an :class:`McpEndpoints` honouring env-var overrides."""
    e = env if env is not None else os.environ
    resolved: dict[str, str] = {}
    for name, (host, port, env_var) in MCP_SERVICES.items():
        override = e.get(env_var)
        resolved[name] = override.rstrip("/") if override else _default_url(host, port)
    return McpEndpoints(**resolved)


_DEFAULT_TOKEN_FILE = "/run/secrets/core_token"
_TOKEN_FILE_ENV = "CERBERO_BITE_CORE_TOKEN_FILE"


def load_token(
    *,
    path: str | Path | None = None,
    env: dict[str, str] | None = None,
) -> str:
    """Read the bearer token from disk and return it stripped.

    Resolution order:
      1. explicit ``path`` argument;
      2. ``CERBERO_BITE_CORE_TOKEN_FILE`` env var;
      3. ``/run/secrets/core_token`` (Docker secrets default).
    """
    e = env if env is not None else os.environ
    target = (
        Path(path)
        if path is not None
        else Path(e.get(_TOKEN_FILE_ENV, _DEFAULT_TOKEN_FILE))
    )
    if not target.is_file():
        raise FileNotFoundError(f"core token file not found: {target}")
    token = target.read_text(encoding="utf-8").strip()
    if not token:
        raise ValueError(f"core token file is empty: {target}")
    return token