data-collector-mcp/data_collector_mcp/app.py

from __future__ import annotations

import logging
import os

import uvicorn

from .db import check_database_connection
from .mcp_app import mcp

# Import tool modules for registration side effects.
from . import bacnet_server as bacnet_server  # noqa: F401
from . import common_server as common_server  # noqa: F401
from . import modbus_server as modbus_server  # noqa: F401
from . import s7_server as s7_server  # noqa: F401


logger = logging.getLogger(__name__)


def build_mcp_http_app(path: str | None = None):
    effective_path = str(path or os.getenv("MCP_PATH", "/mcp")).strip() or "/mcp"
    return mcp.http_app(path=effective_path, transport="http", stateless_http=True)


def main() -> None:
    logging.basicConfig(level=os.getenv("LOG_LEVEL", "INFO"))
    try:
        check_database_connection()
    except Exception:
        logger.exception("Database startup check failed")
        raise SystemExit(1)

    host = os.getenv("MCP_HOST", "0.0.0.0").strip() or "0.0.0.0"
    port = int(os.getenv("MCP_PORT", "8501"))
    path = os.getenv("MCP_PATH", "/mcp").strip() or "/mcp"
    app = build_mcp_http_app(path)
    print(f"Using FastMCP app '{mcp.name}' on http://{host}:{port}{path}")
    uvicorn.run(
        app,
        host=host,
        port=port,
        lifespan="on",
        timeout_graceful_shutdown=2,
        ws="websockets-sansio",
    )