astro-mcp/src/astro_mcp/server.py

from __future__ import annotations

import logging
from pathlib import Path

from fastapi import FastAPI
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

from . import config

logger = logging.getLogger("astro-mcp")

mcp = FastMCP(
    "astro-mcp",
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=False,
    ),
)

# Import tools module to register all @mcp.tool() handlers
from . import tools  # noqa: E402, F401

# Templates and static files
TEMPLATES_DIR = Path(__file__).resolve().parent.parent.parent / "templates"
STATIC_DIR = Path(__file__).resolve().parent.parent.parent / "static"

templates = Jinja2Templates(directory=str(TEMPLATES_DIR))


def _tool_names() -> list[str]:
    return [
        "get_planetary_positions",
        "calculate_natal_chart",
        "calculate_transit_chart",
        "calculate_synastry_chart",
        "calculate_composite_chart",
        "get_transit_preview",
        "person_manage",
        "list_house_systems",
        "calculate_natal_chart_by_id",
        "calculate_transit_chart_by_id",
        "calculate_synastry_chart_by_id",
        "calculate_composite_chart_by_id",
        "get_transit_preview_by_id",
    ]


def create_app() -> FastAPI:
    config.LOG_DIR.mkdir(parents=True, exist_ok=True)
    logging.basicConfig(
        filename=str(config.LOG_DIR / "server.log"),
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

    app = FastAPI(title="astro-mcp")

    # Mount static files
    if STATIC_DIR.exists():
        app.mount("/static", StaticFiles(directory=str(STATIC_DIR)), name="static")

    # Mount MCP SSE
    app.mount("/mcp", mcp.sse_app())

    # Dashboard routes
    from .dashboard import router as dashboard_router  # noqa: E402
    app.include_router(dashboard_router)

    @app.get("/health")
    def health() -> dict:
        return {"ok": True, "server": "astro-mcp", "port": config.PORT}

    @app.get("/")
    def root() -> dict:
        return {
            "server": "astro-mcp",
            "status": "ready",
            "tools": _tool_names(),
            "mcp": {
                "sse": "/mcp/sse",
                "messages": "/mcp/messages",
            },
        }

    return app