astro-mcp/src/astro_mcp/server.py

from __future__ import annotations

import logging
from pathlib import Path

from fastapi import FastAPI, HTTPException
from fastapi.responses import Response
from fastapi.staticfiles import StaticFiles
from fastapi.templating import Jinja2Templates
from mcp.server.fastmcp import FastMCP
from mcp.server.fastmcp.prompts.base import Message
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
from . import chart_resources  # 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",
    ]


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")


@mcp.resource("astro://guides/server-guide")
def server_guide() -> str:
    """Server usage guide — architecture, timezone rules, person management, tool reference."""
    return _read_guide("server-guide.md")


# ── Prompts ──────────────────────────────────────────────────────────

def _msg(role: str, content: str) -> Message:
    return Message(content=content, role=role)


@mcp.prompt()
def natal_reading(person_id: str) -> list[Message]:
    """Full natal chart reading workflow — Big Three, elements, aspects, karmic indicators."""
    return [
        _msg("user", (
            f"Please perform a complete natal chart reading for person {person_id}.\n\n"
            "Steps:\n"
            f"1. Call `calculate_natal_chart_by_id` with person_id=`{person_id}`, "
            "include_overview=true, include_patterns=true, include_karmic=true\n"
            "2. Fetch the guide at `astro://guides/natal-astrology`\n"
            "3. Interpret using the methodology in that guide\n"
            "4. Structure your output as described below"
        )),
        _msg("assistant", (
            "I'll structure the reading as follows:\n"
            "- **The Big Three** — Sun (core identity), Moon (emotional nature), Ascendant (outer expression)\n"
            "- **Elemental & Modal Balance** — dominant element and modality\n"
            "- **Chart Ruler** — the ruler of the Ascendant and its condition\n"
            "- **Key Planetary Aspects** — up to 5 significant aspects with interpretation\n"
            "- **Aspect Patterns** — any T-squares, Grand Trines, Grand Crosses, Yods\n"
            "- **Karmic Indicators** — nodal axis, Saturn, Pluto Polarity Point\n"
            "- **Summary** — a synthesized paragraph integrating all themes\n\n"
            "Fetching chart data now..."
        )),
    ]


@mcp.prompt()
def karmic_reading(person_id: str) -> list[Message]:
    """Karmic-focused natal reading — nodal axis, Saturn, Pluto, retrogrades, 12th house."""
    return [
        _msg("user", (
            f"Please perform a karmic natal reading for person {person_id}.\n\n"
            "Steps:\n"
            f"1. Call `calculate_natal_chart_by_id` with person_id=`{person_id}`, "
            "include_overview=true, include_karmic=true\n"
            "2. Fetch the guide at `astro://guides/karmic-astrology`\n"
            "3. Interpret using the karmic methodology in that guide\n"
            "4. Structure your output as described below"
        )),
        _msg("assistant", (
            "I'll approach this as a karmic reading, leading with the nodal axis and Saturn "
            "before addressing the Big Three. The chart will be read as a record of soul memory "
            "and current-life developmental themes.\n\n"
            "Structure:\n"
            "- **Nodal Axis** — South Node (past-life patterns), North Node (growth direction), house placement\n"
            "- **Saturn** — sign, house, hard aspects to personal planets (karmic lessons)\n"
            "- **Pluto & Polarity Point** — evolutionary desires and the balancing point\n"
            "- **Retrograde Planets** — unfinished past-life business\n"
            "- **12th House** — spiritual karma, unconscious patterns\n"
            "- **Synthesis** — the soul's core developmental arc in this lifetime\n\n"
            "Fetching chart data now..."
        )),
    ]


@mcp.prompt()
def synastry_reading(person1_id: str, person2_id: str) -> list[Message]:
    """Relationship reading — synastry, composite, Davison, karmic overlay."""
    return [
        _msg("user", (
            f"Please perform a relationship reading for person {person1_id} and person {person2_id}.\n\n"
            "Steps:\n"
            f"1. Call `calculate_synastry_chart_by_id` with person1_id=`{person1_id}`, "
            f"person2_id=`{person2_id}`, include_davison_full=true\n"
            f"2. Call `calculate_composite_chart_by_id` with person1_id=`{person1_id}`, "
            f"person2_id=`{person2_id}`\n"
            "3. Fetch the guide at `astro://guides/relationship-astrology`\n"
            "4. Interpret using the methodology in that guide\n"
            "5. Structure your output as described below"
        )),
        _msg("assistant", (
            "I'll structure the relationship reading as follows:\n"
            "- **Synastry Overview** — top interchart aspects (tightest orbs), Venus-Mars chemistry, Sun-Moon compatibility\n"
            "- **Saturn & Node Contacts** — commitment patterns, karmic bonds, fated connections\n"
            "- **House Overlays** — which life areas each person activates in the other\n"
            "- **Composite Chart** — the relationship as its own entity (Sun, Moon, Ascendant, key aspects)\n"
            "- **Davison Chart** — inner emotional tone and long-term evolution\n"
            "- **Karmic Overlay** — Saturn-Node, Pluto-South Node contacts across charts\n"
            "- **Summary** — core narrative, strengths, growth edges\n\n"
            "Fetching chart data now..."
        )),
    ]


@mcp.prompt()
def transit_check(person_id: str, start_date: str, end_date: str) -> str:
    """Transit preview workflow — what's activating a person's natal chart in a date range.

    Args:
        person_id: Person identifier (ID or nickname).
        start_date: Start of the transit window (ISO 8601, e.g. '2026-06-01').
        end_date: End of the transit window (ISO 8601, e.g. '2026-06-30').
    """
    return (
        f"Please perform a transit analysis for person {person_id} "
        f"from {start_date} to {end_date}.\n\n"
        "Steps:\n"
        f"1. Call `get_transit_preview_by_id` with person_id=`{person_id}`, "
        f"start=`{start_date}`, end=`{end_date}`\n"
        "2. Fetch the guide at `astro://guides/natal-astrology` (see the transit section)\n"
        "3. Focus on transits to Sun, Moon, Ascendant (personal activation)\n"
        "4. Note Saturn transits (life structure changes) and Jupiter transits (growth opportunities)\n"
        "5. Highlight any exact aspects (orb <= 1°) as peak activation dates\n\n"
        "Structure:\n"
        "- **Overview** — dominant transit themes for the period\n"
        "- **Key Dates** — exact aspects with dates and interpretation\n"
        "- **Saturn/Jupiter Transits** — structural and growth-oriented movements\n"
        "- **Summary** — what the person is likely experiencing and how to work with it"
    )


@mcp.prompt()
def financial_forecast(target_date: str | None = None) -> str:
    """Financial astrology workflow — market sentiment, planetary cycles, Solar Ingress method.

    Args:
        target_date: Optional focal date (ISO 8601). Defaults to today if not provided.
    """
    date_line = f"Focus the analysis on {target_date}." if target_date else \
                "Use today's date as the focal point."
    return (
        "You are performing a financial astrology analysis using this server's tools.\n\n"
        f"{date_line}\n\n"
        "Steps:\n"
        "1. Call `get_planetary_positions` for the target date\n"
        "2. Fetch the guide at `astro://guides/financial-astrology`\n"
        "3. Identify major aspect patterns (Jupiter-Saturn, Jupiter-Uranus, Saturn-Uranus)\n"
        "4. Check the Moon's North Node position for McWhirter cycle timing\n"
        "5. Apply the Solar Ingress method for quarterly market direction\n"
        "6. Synthesize into a market sentiment outlook\n\n"
        "Structure:\n"
        "- **Active Cycles** — which planetary cycles are currently in play\n"
        "- **Key Aspects** — major configurations and their financial significance\n"
        "- **Market Sentiment** — bullish/bearish/neutral based on aspect balance\n"
        "- **Timing Windows** — dates of exact aspects as potential turning points\n\n"
        "Do not make specific investment recommendations. Frame findings as energetic tendencies."
    )


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("/charts/natal/{person_id}.{format}")
    async def natal_chart_image(person_id: str, format: str, size: int = 600):
        """Return a database-backed natal chart using the existing renderer."""
        try:
            content, content_type = await chart_resources.render_natal_artifact(
                person_id, format=format, size=size,
            )
            return Response(content=content, media_type=content_type)
        except (LookupError, ValueError) as exc:
            raise HTTPException(status_code=404 if isinstance(exc, LookupError) else 400, detail=str(exc)) from exc

    @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