ephemeris-mcp/src/ephemeris_mcp/server.py

"""
Ephemeris MCP server — FastAPI + FastMCP entry point.

All MCP tools are defined here. The ephemeris engine and cache are in
ephemeris.py and storage.py respectively.
"""

from __future__ import annotations

import logging
from datetime import datetime, timezone

from fastapi import FastAPI
from mcp.server.fastmcp import FastMCP
from mcp.server.transport_security import TransportSecuritySettings

from . import config
from .ephemeris import (
    BODIES,
    close as ephemeris_close,
    get_constellation_at_ecliptic as _compute_constellation_at_ecliptic,
    get_lunar_state as _compute_lunar_state,
    get_planetary_positions as _compute_planetary_positions,
    get_sidereal_time as _compute_sidereal_time,
    get_solar_events as _compute_solar_events,
    init as ephemeris_init,
)
from .storage import cache_key, get_cache

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

# ── FastMCP + FastAPI ──────────────────────────────────────────────

cache = get_cache()
mcp = FastMCP(
    "ephemeris-mcp",
    transport_security=TransportSecuritySettings(
        enable_dns_rebinding_protection=False,
    ),
)


# ── Helpers ─────────────────────────────────────────────────────────

def _now_jd() -> float:
    from swisseph import julday

    now = datetime.now(timezone.utc)
    return julday(now.year, now.month, now.day, now.hour + now.minute / 60 + now.second / 3600)


def _parse_datetime(dt_str: str | None) -> float:
    """Parse ISO datetime string to Julian Day. Defaults to now."""
    if dt_str is None:
        return _now_jd()
    from swisseph import julday

    dt = datetime.fromisoformat(dt_str.replace("Z", "+00:00"))
    if dt.tzinfo is None:
        dt = dt.replace(tzinfo=timezone.utc)
    return julday(dt.year, dt.month, dt.day, dt.hour + dt.minute / 60 + dt.second / 3600)


def _parse_date(date_str: str) -> float:
    """Parse ISO date string to Julian Day at noon."""
    from swisseph import julday

    dt = datetime.fromisoformat(date_str)
    return julday(dt.year, dt.month, dt.day, 12.0)


def _default_location(lat: float | None = None, lon: float | None = None) -> tuple[float, float]:
    resolved_lat = getattr(config, "DEFAULT_LAT", None) if lat is None else lat
    resolved_lon = getattr(config, "DEFAULT_LON", None) if lon is None else lon
    return (resolved_lat if resolved_lat is not None else 0.0, resolved_lon if resolved_lon is not None else 0.0)


def _tool_names() -> list[str]:
    return [
        "get_planetary_positions",
        "get_solar_events",
        "get_lunar_state",
        "get_moon_phase",
        "get_sidereal_time",
        "get_constellation_at_ecliptic",
        "list_available_bodies",
    ]


# ── Group A — Celestial Mechanics ──────────────────────────────────

@mcp.tool()
def get_planetary_positions(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
    elevation: float = 0.0,
    geocentric: bool = False,
) -> dict:
    """
    Get positions of all major solar system bodies.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.
        elevation: Observer elevation in meters.
        geocentric: If True, return geocentric positions instead of topocentric.

    Returns:
        Object with 'input' (echoed params), 'timestamp', and 'bodies' array.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_datetime(datetime)

    ck = cache_key("get_planetary_positions", datetime=datetime or "now", lat=lat, lon=lon, geocentric=geocentric)
    cached = cache.get(ck)
    if cached:
        logger.info(f"cache hit: {ck}")
        return {"input": {"datetime": datetime, "lat": lat, "lon": lon, "geocentric": geocentric}, **cached}

    positions = _compute_planetary_positions(jd, lat, lon, elevation, geocentric)

    result = {
        "input": {"datetime": datetime, "lat": lat, "lon": lon, "geocentric": geocentric},
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        "bodies": positions,
    }
    cache.set(
        ck,
        {"timestamp_utc": result["timestamp_utc"], "julian_day": result["julian_day"], "bodies": positions},
        config.CACHE_PLANETARY,
    )
    return result


@mcp.tool()
def get_solar_events(
    date: str,
    lat: float | None = None,
    lon: float | None = None,
) -> dict:
    """
    Get solar events (sunrise, sunset, solar noon, twilight) for a date and location.

    Args:
        date: ISO date string (YYYY-MM-DD).
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.

    Returns:
        Object with all event times as ISO datetime strings and day length.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_date(date)

    ck = cache_key("get_solar_events", date=date, lat=lat, lon=lon)
    cached = cache.get(ck)
    if cached:
        return {"input": {"date": date, "lat": lat, "lon": lon}, **cached}

    events = _compute_solar_events(jd, lat, lon)

    result = {
        "input": {"date": date, "lat": lat, "lon": lon},
        "date": date,
        "julian_day_base": round(jd, 6),
        "events_jd": events,
    }
    cache.set(
        ck,
        {"date": date, "events_jd": events, "julian_day_base": round(jd, 6)},
        config.CACHE_SOLAR,
    )
    return result


@mcp.tool()
def get_lunar_state(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
) -> dict:
    """
    Get current lunar phase and position.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude in decimal degrees.
        lon: Observer longitude in decimal degrees.

    Returns:
        Object with phase name, illumination fraction, age, and position.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_datetime(datetime)

    ck = cache_key("get_lunar_state", datetime=datetime or "now", lat=lat, lon=lon)
    cached = cache.get(ck)
    cached_next_major_phase = cached.get("lunar_state", {}).get("next_major_phase") if cached else None
    if cached_next_major_phase and cached_next_major_phase.get("phase_name") and cached_next_major_phase.get("at_utc") and cached_next_major_phase.get("in_text"):
        return {"input": {"datetime": datetime, "lat": lat, "lon": lon}, **cached}
    if cached:
        cache.delete(ck)

    state = _compute_lunar_state(jd, lat, lon)

    result = {
        "input": {"datetime": datetime, "lat": lat, "lon": lon},
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        "lunar_state": state,
    }
    cache.set(
        ck,
        {"timestamp_utc": result["timestamp_utc"], "julian_day": result["julian_day"], "lunar_state": state},
        config.CACHE_LUNAR,
    )
    return result


@mcp.tool()
def get_sidereal_time(
    datetime: str | None = None,
    lat: float | None = None,
    lon: float | None = None,
) -> dict:
    """
    Get sidereal time and obliquity of the ecliptic.

    Args:
        datetime: ISO 8601 datetime (UTC). Defaults to now.
        lat: Observer latitude.
        lon: Observer longitude.

    Returns:
        Object with Greenwich and local sidereal time, and obliquity.
    """
    lat, lon = _default_location(lat, lon)
    jd = _parse_datetime(datetime)

    ck = cache_key("get_sidereal_time", datetime=datetime or "now", lat=lat, lon=lon)
    cached = cache.get(ck)
    if cached:
        return {"input": {"datetime": datetime, "lat": lat, "lon": lon}, **cached}

    result_st = _compute_sidereal_time(jd, lat, lon)

    result = {
        "input": {"datetime": datetime, "lat": lat, "lon": lon},
        "timestamp_utc": datetime or __import__("datetime").datetime.now(timezone.utc).isoformat(),
        "julian_day": round(jd, 6),
        **result_st,
    }
    cache.set(
        ck,
        {"timestamp_utc": result["timestamp_utc"], "julian_day": result["julian_day"], **result_st},
        config.CACHE_SIDEREAL,
    )
    return result


@mcp.tool()
def get_constellation_at_ecliptic(ecliptic_lon: float) -> dict:
    """
    Look up IAU constellation at a given ecliptic longitude.

    Args:
        ecliptic_lon: Ecliptic longitude in degrees (0-360).

    Returns:
        Object with constellation abbreviation, full name, and position within.
    """
    result = _compute_constellation_at_ecliptic(ecliptic_lon)
    return {"input": {"ecliptic_lon": ecliptic_lon}, **result}


@mcp.tool()
def list_available_bodies(category: str | None = None) -> dict:
    """
    List all computable celestial bodies.

    This is a lightweight discovery tool for the v0.1 core slice.
    """
    bodies = []
    for name in BODIES:
        bodies.append(
            {
                "name": name,
                "full_name": name.replace("_", " ").title(),
                "available": True,
            }
        )

    if category:
        category_lower = category.lower()
        if category_lower == "planets":
            bodies = [b for b in bodies if b["name"] in {"sun", "mercury", "venus", "mars", "jupiter", "saturn", "uranus", "neptune", "pluto"}]
        elif category_lower == "lunar":
            bodies = [b for b in bodies if b["name"] == "moon"]
        elif category_lower == "major":
            bodies = [b for b in bodies if b["name"] in {"sun", "moon", "mercury", "venus", "mars", "jupiter", "saturn"}]

    return {"bodies": bodies}


@mcp.tool()
def get_moon_phase(datetime: str | None = None, lat: float | None = None, lon: float | None = None) -> dict:
    """
    Convenience lunar alias for mcporter users.

    Phase naming is driven by lunar age, while illumination is reported as
    a separate descriptive value.
    """
    lat, lon = _default_location(lat, lon)
    state = get_lunar_state(datetime=datetime, lat=lat, lon=lon)
    return {
        "input": {"datetime": datetime, "lat": lat, "lon": lon},
        "phase_name": state["lunar_state"]["phase_name"],
        "illumination_fraction": state["lunar_state"]["illumination_fraction"],
        "age_days": state["lunar_state"]["age_days"],
        "next_major_phase": state["lunar_state"]["next_major_phase"],
        "ecliptic_lon": state["lunar_state"]["ecliptic_lon"],
        "ecliptic_lat": state["lunar_state"]["ecliptic_lat"],
        "distance_km": state["lunar_state"]["distance_km"],
    }


def create_app() -> FastAPI:
    """
    Build the FastAPI app for the v0.1 core slice.
    """
    logging.basicConfig(
        filename=str(config.LOG_DIR / "server.log"),
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    )

    ephemeris_init()
    app = FastAPI(title="ephemeris-mcp")
    app.mount("/mcp", mcp.sse_app())

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

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

    return app