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
from . import __version__

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"
GUIDES_DIR = Path(__file__).resolve().parent.parent.parent / "agent-guides"

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",
        "calculate_davison_chart",
        "get_transit_preview",
        "get_composite_transit_preview",
        "get_davison_transit_preview",
        "get_karmic_relationship_summary",
        "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",
        "calculate_davison_chart_by_id",
        "get_transit_preview_by_id",
        # Chart rendering tools
        "render_natal_chart",
        "render_natal_chart_by_id",
        "render_transit_chart",
        "render_transit_chart_by_id",
        "render_synastry_chart",
        "render_synastry_chart_by_id",
        "render_composite_chart",
        "render_composite_chart_by_id",
        "render_davison_chart",
        "render_davison_chart_by_id",
    ]


def _read_guide(filename: str) -> str:
    """Read an agent guide markdown file."""
    path = GUIDES_DIR / filename
    if path.exists():
        return path.read_text(encoding="utf-8")
    return f"# Error\n\nGuide not found: {filename}"


@mcp.resource("astro://guides/natal-astrology")
def natal_astrology_guide() -> str:
    """Agent interpretation guide for natal chart analysis."""
    return _read_guide("natal-astrology.md")


@mcp.resource("astro://guides/karmic-astrology")
def karmic_astrology_guide() -> str:
    """Agent interpretation guide for karmic chart analysis."""
    return _read_guide("karmic-astrology.md")


@mcp.resource("astro://guides/relationship-astrology")
def relationship_astrology_guide() -> str:
    """Agent interpretation guide for relationship chart analysis."""
    return _read_guide("relationship-astrology.md")


@mcp.resource("astro://guides/financial-astrology")
def financial_astrology_guide() -> str:
    """Agent interpretation guide for financial astrology — business cycles, stock market forecasting, planetary correlations."""
    return _read_guide("financial-astrology.md")


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", "version": __version__, "port": config.PORT}

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

    return app