mcp_server.py

#!/usr/bin/env python3
# /// script
# requires-python = ">=3.11"
# dependencies = [
#   "mcp[cli]>=1.9",
#   "pydantic>=2.0",
# ]
# ///
"""GPS MCP server — read-only caching tier for org and engineering data.

Gives agents and humans sub-millisecond access to people, teams, issues,
features, releases, and governance documents via MCP tools. No auth required.

Run: uv run mcp_server.py              # stdio (Claude Code / ACP)
     uv run mcp_server.py --http       # HTTP on :8000 (shared deployment)
"""

import argparse
import importlib
import json
import logging
import os
import re
import sqlite3
from datetime import date, datetime, timedelta
from difflib import get_close_matches
from pathlib import Path

from mcp.server.fastmcp import FastMCP
from starlette.requests import Request
from starlette.responses import JSONResponse, Response

REPO_ROOT = Path(__file__).resolve().parent
DATA_DIR = REPO_ROOT / "data"
DB_PATH = DATA_DIR / "gps.db"
CATALOG_PATH = DATA_DIR / "DATA_CATALOG.yaml"
VERSION_PATH = REPO_ROOT / "VERSION"
VERSION = VERSION_PATH.read_text().strip() if VERSION_PATH.exists() else "unknown"
MAX_QUERY_ROWS = 200

mcp = FastMCP(
    "gps",
    instructions=(
        "GPS is a read-only caching tier for org and engineering data. "
        "Query people, teams, issues, features, release schedules, "
        "component mappings, scrum team boards, and governance documents with sub-ms latency. "
        "Use list_scrum_team_boards for scrum team staffing data (FTE counts by role). "
        "Use list_documents/get_document/get_document_section for governance docs. "
        "Use cloud_pricing_lookup for AWS/Claude pricing data. "
        "Use get_direct_reports/get_org_tree/get_management_chain for org hierarchy queries. "
        "Use github_* tools or query with github.* prefix for GitHub data. "
        "All data is read-only."
    ),
)


# ---------------------------------------------------------------------------
# Health check
# ---------------------------------------------------------------------------


@mcp.custom_route("/health", methods=["GET"])
async def health(request: Request) -> Response:
    return JSONResponse({"status": "ok", "version": VERSION})


# ---------------------------------------------------------------------------
# DB helpers
# ---------------------------------------------------------------------------


_conn: sqlite3.Connection | None = None


def _get_conn() -> sqlite3.Connection:
    global _conn
    if _conn is not None:
        return _conn
    conn = sqlite3.connect(f"file:{DB_PATH}?mode=ro", uri=True)
    conn.row_factory = sqlite3.Row
    conn.execute("PRAGMA query_only = ON")
    conn.execute("PRAGMA mmap_size = 268435456")
    conn.execute("PRAGMA cache_size = -64000")
    conn.execute("PRAGMA temp_store = MEMORY")
    # Attach optional databases if they exist
    for db_name in ("pricing", "github"):
        db_path = DB_PATH.parent / f"{db_name}.db"
        if db_path.exists():
            conn.execute(f"ATTACH DATABASE 'file:{db_path}?mode=ro' AS {db_name}")
    _conn = conn
    return conn


def _rows_to_dicts(rows: list[sqlite3.Row]) -> list[dict]:
    return [dict(r) for r in rows]


def _db_attached(conn: sqlite3.Connection, name: str) -> bool:
    """Check if a database is attached."""
    try:
        conn.execute(f"SELECT 1 FROM {name}.sqlite_master LIMIT 1")
        return True
    except sqlite3.OperationalError:
        return False


def _attached_db_schema(db_name: str, build_hint: str) -> str:
    """Return schema DDL + row counts for an attached database."""
    conn = _get_conn()
    if not _db_attached(conn, db_name):
        return f"{db_name}.db is not attached. Run: {build_hint}"
    lines = [f"-- {db_name.upper()} TABLES\n"]
    tables = conn.execute(
        f"SELECT name, sql FROM {db_name}.sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
    ).fetchall()
    for t in tables:
        count = conn.execute(f"SELECT COUNT(*) FROM {db_name}.[{t['name']}]").fetchone()[0]
        lines.append(f"-- {t['name']}: {count:,} rows")
        lines.append(t["sql"] + ";\n")
    views = conn.execute(f"SELECT name, sql FROM {db_name}.sqlite_master WHERE type='view' ORDER BY name").fetchall()
    if views:
        lines.append(f"\n-- {db_name.upper()} VIEWS\n")
        for v in views:
            lines.append(v["sql"] + ";\n")
    return "\n".join(lines)


def _normalize_release(release: str) -> tuple[str, str]:
    """Extract (product, major.minor) from release strings in various formats.

    Examples:
        'acmeproduct-1.0'      -> ('acmeproduct', '1.0')
        'acmeproduct-1.1'      -> ('acmeproduct', '1.1')
        'acmeproduct-2.0 EA-1' -> ('acmeproduct', '2.0')
        'myproduct-2.25.2'     -> ('myproduct', '2.25')
    """
    s = release.strip().lower().replace(" ", "")
    # Split on first hyphen to get product and version
    if "-" in s:
        parts = s.split("-", 1)
        product = parts[0]
        ver = parts[1]
    else:
        product = s
        ver = ""
    # Strip EA suffix and patch version — keep only major.minor
    ver = re.sub(r"\s*ea.*$", "", ver, flags=re.IGNORECASE)
    match = re.match(r"(\d+\.\d+)", ver)
    major_minor = match.group(1) if match else ver
    return product, major_minor


def _milestone_release_key(product: str, version: str) -> tuple[str, str]:
    """Extract (product, major.minor) from milestone product+version.

    Examples:
        ('AcmeProduct', '1.0.0')   -> ('acmeproduct', '1.0')
        ('AcmeProduct', '2.0 EA1') -> ('acmeproduct', '2.0')
        ('My Product', '3.3.1')    -> ('myproduct', '3.3')
    """
    product_key = product.lower().replace(" ", "")
    ver = re.sub(r"\s*ea.*$", "", version, flags=re.IGNORECASE)
    match = re.match(r"(\d+\.\d+)", ver)
    major_minor = match.group(1) if match else ver
    return product_key, major_minor


def _parse_date(val: str, reference_year: int | None = None) -> date | None:
    """Parse dates in various formats found in the DB.

    Handles: 'YYYY-MM-DD', 'Mon-DD' (e.g. 'Apr-10'), 'Mon DD'.
    For formats without a year, uses reference_year (default: current year).
    """
    if not val or not val.strip():
        return None
    val = val.strip()
    # ISO format
    try:
        return date.fromisoformat(val)
    except ValueError:
        pass
    # Mon-DD or Mon DD
    year = reference_year or date.today().year
    for fmt in ("%b-%d", "%b %d"):
        try:
            parsed = datetime.strptime(val, fmt).replace(year=year)
            return parsed.date()
        except ValueError:
            continue
    return None


# ---------------------------------------------------------------------------
# Resources
# ---------------------------------------------------------------------------


@mcp.resource(
    "gps://schema",
    name="Database Schema",
    description="DDL and row counts for all tables and views in gps.db",
)
def schema_resource() -> str:
    conn = _get_conn()
    lines = []
    # Tables with DDL and row counts
    tables = conn.execute(
        "SELECT name, sql FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
    ).fetchall()
    lines.append("-- TABLES\n")
    for t in tables:
        count = conn.execute(f"SELECT COUNT(*) FROM [{t['name']}]").fetchone()[0]
        lines.append(f"-- {t['name']}: {count:,} rows")
        lines.append(t["sql"] + ";\n")

    # Views with DDL
    views = conn.execute("SELECT name, sql FROM sqlite_master WHERE type='view' ORDER BY name").fetchall()
    if views:
        lines.append("\n-- VIEWS\n")
        for v in views:
            lines.append(v["sql"] + ";\n")

    return "\n".join(lines)


@mcp.resource(
    "gps://catalog",
    name="Data Catalog",
    description="Source file inventory with descriptions, dates, and provenance",
)
def catalog_resource() -> str:
    if CATALOG_PATH.exists():
        return CATALOG_PATH.read_text()
    return "DATA_CATALOG.yaml not found."


@mcp.resource(
    "gps://pricing-schema",
    name="Cloud Pricing Database Schema",
    description="DDL and row counts for pricing.db (when available)",
)
def pricing_schema_resource() -> str:
    return _attached_db_schema("pricing", "uv run scripts/fetch_pricing.py")


@mcp.resource(
    "gps://github-schema",
    name="GitHub Database Schema",
    description="DDL and row counts for github.db (when available)",
)
def github_schema_resource() -> str:
    return _attached_db_schema("github", "uv run scripts/fetch_github.py --org YOUR_ORG")


# ---------------------------------------------------------------------------
# Tools
# ---------------------------------------------------------------------------


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def lookup_person(name: str | None = None, uid: str | None = None, email: str | None = None) -> str:
    """Find people in the org by name, user ID, or email.

    Searches are case-insensitive and partial-match. Provide at least one parameter.
    Returns full person detail including org, components, teams, and specialty.
    """
    if not any([name, uid, email]):
        return json.dumps({"error": "Provide at least one of: name, uid, email"})

    conditions, params = [], []
    if name:
        conditions.append("p.name LIKE ?")
        params.append(f"%{name}%")
    if uid:
        conditions.append("p.user_id LIKE ?")
        params.append(f"%{uid}%")
    if email:
        conditions.append("p.email LIKE ?")
        params.append(f"%{email}%")

    where = " AND ".join(conditions)

    conn = _get_conn()
    sql = f"""
    SELECT
        p.person_id, p.name, p.user_id, p.manager, p.manager_uid,
        o.org_name, o.org_key, s.specialty_name AS specialty,
        p.job_title, p.email, p.location, p.status, p.source, p.last_modified,
        GROUP_CONCAT(DISTINCT jc.component_name) AS components,
        GROUP_CONCAT(DISTINCT mt.miro_team_name) AS miro_teams,
        GROUP_CONCAT(DISTINCT st.team_name) AS scrum_teams
    FROM person p
    LEFT JOIN org o ON p.org_id = o.org_id
    LEFT JOIN specialty s ON p.specialty_id = s.specialty_id
    LEFT JOIN person_component pc ON p.person_id = pc.person_id
    LEFT JOIN jira_component jc ON pc.component_id = jc.component_id
    LEFT JOIN person_miro_team pmt ON p.person_id = pmt.person_id
    LEFT JOIN miro_team mt ON pmt.miro_team_id = mt.miro_team_id
    LEFT JOIN person_scrum_team pst ON p.person_id = pst.person_id
    LEFT JOIN scrum_team st ON pst.team_id = st.team_id
    WHERE {where}
    GROUP BY p.person_id
    ORDER BY p.name
    LIMIT 50
    """
    rows = conn.execute(sql, params).fetchall()
    return json.dumps({"results": _rows_to_dicts(rows), "count": len(rows)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def list_team_members(team_name: str) -> str:
    """List all members of a scrum team with their roles and details.

    Fuzzy-matches team name. Returns team info (PM, eng lead) plus full member roster.
    """
    conn = _get_conn()
    # Find matching team(s)
    teams = conn.execute(
        "SELECT team_id, team_name, pm, eng_lead FROM scrum_team WHERE team_name LIKE ?",
        (f"%{team_name}%",),
    ).fetchall()

    if not teams:
        return json.dumps(
            {
                "error": f"No team matching '{team_name}'",
                "hint": "Use lookup to find available team names",
            }
        )

    results = []
    for team in teams:
        members = conn.execute(
            """
            SELECT p.name, p.user_id, p.job_title, p.email, p.location,
                   o.org_name, s.specialty_name AS specialty,
                   GROUP_CONCAT(DISTINCT jc.component_name) AS components
            FROM person_scrum_team pst
            JOIN person p ON pst.person_id = p.person_id
            LEFT JOIN org o ON p.org_id = o.org_id
            LEFT JOIN specialty s ON p.specialty_id = s.specialty_id
            LEFT JOIN person_component pc ON p.person_id = pc.person_id
            LEFT JOIN jira_component jc ON pc.component_id = jc.component_id
            WHERE pst.team_id = ?
            GROUP BY p.person_id
            ORDER BY p.name
            """,
            (team["team_id"],),
        ).fetchall()
        results.append(
            {
                "team_name": team["team_name"],
                "pm": team["pm"],
                "eng_lead": team["eng_lead"],
                "headcount": len(members),
                "members": _rows_to_dicts(members),
            }
        )

    return json.dumps({"teams": results}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def list_scrum_team_boards(organization: str | None = None) -> str:
    """List scrum team boards with staffing breakdown by role.

    Optionally filter by organization (fuzzy match). Returns team name,
    Jira board URL, PM, and non-zero FTE counts per role.
    """
    conn = _get_conn()
    query = (
        "SELECT organization, scrum_team_name, jira_board_url, pm, "
        "agilist, architects, bff, backend_engineer, devops, manager, "
        "operations_manager, qe, staff_engineers, ui, total_staff "
        "FROM scrum_team_board"
    )
    if organization:
        query += " WHERE organization LIKE ?"
        rows = conn.execute(query + " ORDER BY total_staff DESC", (f"%{organization}%",)).fetchall()
    else:
        rows = conn.execute(query + " ORDER BY total_staff DESC").fetchall()
    role_cols = [
        "agilist",
        "architects",
        "bff",
        "backend_engineer",
        "devops",
        "manager",
        "operations_manager",
        "qe",
        "staff_engineers",
        "ui",
    ]
    teams = []
    for row in rows:
        d = dict(row)
        roles = {k: d.pop(k) for k in role_cols if d.get(k)}
        d["roles"] = roles
        teams.append(d)
    return json.dumps({"teams": teams, "count": len(teams)}, default=str)


# ---------------------------------------------------------------------------
# Org hierarchy tools
# ---------------------------------------------------------------------------


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_direct_reports(uid: str) -> str:
    """List all people who directly report to the given user ID.

    Returns each direct report with their full person detail (name, title, email,
    location, org, teams). Use this to answer 'who reports to X' questions.
    """
    conn = _get_conn()
    sql = """
    SELECT
        p.person_id, p.name, p.user_id, p.manager, p.manager_uid,
        o.org_name, o.org_key, s.specialty_name AS specialty,
        p.job_title, p.email, p.location, p.status, p.source, p.last_modified,
        GROUP_CONCAT(DISTINCT jc.component_name) AS components,
        GROUP_CONCAT(DISTINCT mt.miro_team_name) AS miro_teams,
        GROUP_CONCAT(DISTINCT st.team_name) AS scrum_teams
    FROM person p
    LEFT JOIN org o ON p.org_id = o.org_id
    LEFT JOIN specialty s ON p.specialty_id = s.specialty_id
    LEFT JOIN person_component pc ON p.person_id = pc.person_id
    LEFT JOIN jira_component jc ON pc.component_id = jc.component_id
    LEFT JOIN person_miro_team pmt ON p.person_id = pmt.person_id
    LEFT JOIN miro_team mt ON pmt.miro_team_id = mt.miro_team_id
    LEFT JOIN person_scrum_team pst ON p.person_id = pst.person_id
    LEFT JOIN scrum_team st ON pst.team_id = st.team_id
    WHERE p.manager_uid = ?
    GROUP BY p.person_id
    ORDER BY p.name
    """
    rows = conn.execute(sql, (uid,)).fetchall()
    return json.dumps(
        {"manager_uid": uid, "direct_reports": _rows_to_dicts(rows), "count": len(rows)},
        default=str,
    )


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_org_tree(uid: str, max_depth: int = 5) -> str:
    """Get the full organizational tree under a person, recursively.

    Walks down through manager_uid relationships. Returns a flattened list of
    all people in the org tree with their depth level. Use this to answer
    'who is in X's org' or 'list everyone under VP xyz'.
    max_depth limits how deep to recurse (default 5).
    """
    conn = _get_conn()
    sql = """
    WITH RECURSIVE org_tree AS (
        SELECT user_id, name, job_title, email, location, manager_uid, 0 AS depth
        FROM person
        WHERE user_id = ?
        UNION ALL
        SELECT p.user_id, p.name, p.job_title, p.email, p.location, p.manager_uid, ot.depth + 1
        FROM person p
        JOIN org_tree ot ON p.manager_uid = ot.user_id
        WHERE ot.depth < ?
    )
    SELECT user_id, name, job_title, email, location, manager_uid, depth
    FROM org_tree
    ORDER BY depth, name
    """
    rows = conn.execute(sql, (uid, max_depth)).fetchall()
    cols = ["user_id", "name", "job_title", "email", "location", "manager_uid", "depth"]
    results = [dict(zip(cols, row)) for row in rows]
    return json.dumps(
        {"root_uid": uid, "max_depth": max_depth, "people": results, "count": len(results)},
        default=str,
    )


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_management_chain(uid: str) -> str:
    """Get the management chain from a person up to the top of the org.

    Walks up through manager_uid relationships. Returns an ordered list from
    the person to the top of the org. Use this to answer 'who is the VP over X'
    or 'what is the reporting chain for X'.
    """
    conn = _get_conn()
    sql = """
    WITH RECURSIVE chain AS (
        SELECT user_id, name, job_title, email, location, manager_uid, 0 AS level
        FROM person
        WHERE user_id = ?
        UNION ALL
        SELECT p.user_id, p.name, p.job_title, p.email, p.location, p.manager_uid, c.level + 1
        FROM person p
        JOIN chain c ON c.manager_uid = p.user_id
        WHERE c.manager_uid IS NOT NULL AND c.manager_uid != ''
          AND c.level < 20
    )
    SELECT user_id, name, job_title, email, location, manager_uid, level
    FROM chain
    ORDER BY level
    """
    rows = conn.execute(sql, (uid,)).fetchall()
    cols = ["user_id", "name", "job_title", "email", "location", "manager_uid", "level"]
    results = [dict(zip(cols, row)) for row in rows]
    return json.dumps(
        {"uid": uid, "chain": results, "depth": len(results)},
        default=str,
    )


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def search_issues(
    status: str | None = None,
    priority: str | None = None,
    assignee: str | None = None,
    component: str | None = None,
    label: str | None = None,
    keyword: str | None = None,
    limit: int = 50,
) -> str:
    """Search issues with optional filters. All filters are case-insensitive partial match.

    Provide at least one filter. Use keyword to search summary text.
    Returns up to `limit` issues (default 50, max 200).
    """
    if not any([status, priority, assignee, component, label, keyword]):
        return json.dumps({"error": "Provide at least one filter"})

    limit = min(limit, MAX_QUERY_ROWS)
    conditions, params = [], []

    if status:
        conditions.append("ji.status LIKE ?")
        params.append(f"%{status}%")
    if priority:
        conditions.append("ji.priority LIKE ?")
        params.append(f"%{priority}%")
    if assignee:
        conditions.append("ji.assignee LIKE ?")
        params.append(f"%{assignee}%")
    if keyword:
        conditions.append("ji.summary LIKE ?")
        params.append(f"%{keyword}%")
    if component:
        conditions.append(
            "EXISTS (SELECT 1 FROM issue_component ic WHERE ic.issue_id = ji.issue_id AND ic.component_name LIKE ?)"
        )
        params.append(f"%{component}%")
    if label:
        conditions.append("EXISTS (SELECT 1 FROM issue_label il WHERE il.issue_id = ji.issue_id AND il.label LIKE ?)")
        params.append(f"%{label}%")

    where = " AND ".join(conditions)

    conn = _get_conn()
    sql = f"""
    SELECT ji.key, ji.summary, ji.status, ji.priority,
           ji.assignee, ji.reporter, ji.issue_type,
           ji.created, ji.updated
    FROM jira_issue ji
    WHERE {where}
    ORDER BY ji.updated DESC
    LIMIT ?
    """
    params.append(limit)
    rows = conn.execute(sql, params).fetchall()
    return json.dumps({"issues": _rows_to_dicts(rows), "count": len(rows)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_feature_status(issue_key: str | None = None, title: str | None = None) -> str:
    """Get feature details including progress, RICE score, releases, components, and teams.

    Search by exact issue key (e.g. "PROJ-1234") or fuzzy title match.
    """
    if not issue_key and not title:
        return json.dumps({"error": "Provide issue_key or title"})

    conn = _get_conn()
    if issue_key:
        features = conn.execute("SELECT * FROM feature WHERE issue_key = ?", (issue_key,)).fetchall()
    else:
        features = conn.execute("SELECT * FROM feature WHERE title LIKE ? LIMIT 20", (f"%{title}%",)).fetchall()

    if not features:
        return json.dumps({"error": f"No feature found for {'key=' + issue_key if issue_key else 'title=' + title}"})

    # Batch-load junction tables to avoid N+1 queries
    fids = [f["feature_id"] for f in features]
    ph = ",".join("?" * len(fids))
    rel_rows = conn.execute(f"SELECT feature_id, release FROM feature_release WHERE feature_id IN ({ph})", fids)
    comp_rows = conn.execute(f"SELECT feature_id, component FROM feature_component WHERE feature_id IN ({ph})", fids)
    team_rows = conn.execute(f"SELECT feature_id, team FROM feature_team WHERE feature_id IN ({ph})", fids)

    rels_by_fid: dict[int, list[str]] = {}
    for r in rel_rows:
        rels_by_fid.setdefault(r["feature_id"], []).append(r["release"])
    comps_by_fid: dict[int, list[str]] = {}
    for r in comp_rows:
        comps_by_fid.setdefault(r["feature_id"], []).append(r["component"])
    teams_by_fid: dict[int, list[str]] = {}
    for r in team_rows:
        teams_by_fid.setdefault(r["feature_id"], []).append(r["team"])

    results = []
    for f in features:
        fid = f["feature_id"]
        results.append(
            {
                "issue_key": f["issue_key"],
                "title": f["title"],
                "status": f["issue_status"],
                "priority": f["priority"],
                "progress_pct": f["progress_pct"],
                "rice_score": f["rice_score"],
                "assignee": f["assignee"],
                "target_start": f["target_start"],
                "target_end": f["target_end"],
                "due_date": f["due_date"],
                "releases": rels_by_fid.get(fid, []),
                "components": comps_by_fid.get(fid, []),
                "teams": teams_by_fid.get(fid, []),
                "product_manager": f["product_manager"],
                "tech_lead": f["tech_lead"],
                "developer": f["developer"],
                "tester": f["tester"],
                "todo_ic": f["todo_ic"],
                "in_progress_ic": f["in_progress_ic"],
                "done_ic": f["done_ic"],
                "total_ic": f["total_ic"],
            }
        )

    return json.dumps({"features": results, "count": len(results)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_release_schedule(
    product: str | None = None,
    version: str | None = None,
    milestone: str | None = None,
) -> str:
    """Get release schedule dates for any product.

    All parameters are optional and fuzzy-matched (case-insensitive, partial).
    Returns matching milestones ordered by date, including past dates.

    Args:
        product: Product name. Examples: "AcmeProduct", "acme".
        version: Version number. "1.0" also matches "1.0.1", "1.0 EA1", etc.
        milestone: Milestone type. Examples: "code freeze", "ga", "release", "planning".

    Examples:
        get_release_schedule(product="acme", version="1.0", milestone="code freeze")
        get_release_schedule(version="2.0")
        get_release_schedule(milestone="ga")
    """
    conn = _get_conn()
    today = date.today()

    # --- Query release_schedule table ---
    sched_conditions: list[str] = []
    sched_params: list[str] = []
    if product:
        sched_conditions.append("release LIKE ?")
        sched_params.append(f"%{product}%")
    if version:
        sched_conditions.append("release LIKE ?")
        sched_params.append(f"%{version}%")
    if milestone:
        sched_conditions.append("task LIKE ?")
        sched_params.append(f"%{milestone}%")

    sched_where = " AND ".join(sched_conditions) if sched_conditions else "1=1"
    sched_rows = conn.execute(
        f"""SELECT release, task AS milestone, date_start, date_finish
            FROM release_schedule
            WHERE {sched_where}
            ORDER BY date_start""",
        sched_params,
    ).fetchall()

    # --- Query release_milestone table ---
    mile_conditions: list[str] = []
    mile_params: list[str] = []
    if product:
        mile_conditions.append("product LIKE ?")
        mile_params.append(f"%{product}%")
    if version:
        mile_conditions.append("version LIKE ?")
        mile_params.append(f"%{version}%")
    if milestone:
        mile_conditions.append("event_type LIKE ?")
        mile_params.append(f"%{milestone}%")

    mile_where = " AND ".join(mile_conditions) if mile_conditions else "1=1"
    mile_rows = conn.execute(
        f"""SELECT product, version, event_type AS milestone, event_date
            FROM release_milestone
            WHERE {mile_where}
            ORDER BY event_date""",
        mile_params,
    ).fetchall()

    schedule = _rows_to_dicts(sched_rows)
    milestones_list = _rows_to_dicts(mile_rows)

    # Annotate schedule entries with past/future
    for entry in schedule:
        d = entry.get("date_finish") or entry.get("date_start")
        if d:
            parsed = _parse_date(d)
            if parsed:
                entry["status"] = "past" if parsed < today else "upcoming"
                entry["days_away"] = (parsed - today).days

    # Annotate milestone entries with past/future
    for entry in milestones_list:
        d = entry.get("event_date")
        if d:
            parsed = _parse_date(d)
            if parsed:
                entry["status"] = "past" if parsed < today else "upcoming"
                entry["days_away"] = (parsed - today).days

    if not schedule and not milestones_list:
        hints = []
        releases = conn.execute("SELECT DISTINCT release FROM release_schedule ORDER BY release").fetchall()
        if releases:
            hints = [r["release"] for r in releases]
        return json.dumps(
            {
                "error": "No matching releases found",
                "available_releases": hints,
            }
        )

    return json.dumps(
        {
            "schedule": schedule,
            "milestones": milestones_list,
            "schedule_count": len(schedule),
            "milestone_count": len(milestones_list),
            "as_of": today.isoformat(),
        },
        default=str,
    )


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def release_risk_summary(release: str | None = None, lookback_days: int = 30) -> str:
    """Assess release risk by comparing milestone dates against feature completion.

    If no release specified, analyzes all releases with recent or upcoming milestones.
    Flags features under 80% complete when milestone is within 30 days.

    Args:
        release: Filter to a specific release (fuzzy match on version). Omit for all.
        lookback_days: Include milestones up to this many days in the past (default 30).
    """
    today = date.today()
    conn = _get_conn()
    # Get all milestones (filtering done in Python due to non-ISO date formats)
    if release:
        milestones = conn.execute(
            """SELECT product, version, event_type, event_date
               FROM release_milestone
               WHERE version LIKE ?""",
            (f"%{release}%",),
        ).fetchall()
    else:
        milestones = conn.execute("SELECT product, version, event_type, event_date FROM release_milestone").fetchall()

    # Parse dates and filter to upcoming only
    releases_info = {}
    for m in milestones:
        parsed = _parse_date(m["event_date"], today.year)
        if not parsed or parsed < today - timedelta(days=lookback_days):
            continue
        key = f"{m['product']} {m['version']}"
        if key not in releases_info:
            releases_info[key] = {
                "product": m["product"],
                "version": m["version"],
                "milestones": [],
            }
        releases_info[key]["milestones"].append(
            {
                "event_type": m["event_type"],
                "event_date": m["event_date"],
                "days_away": (parsed - today).days,
            }
        )

    if not releases_info:
        return json.dumps(
            {
                "message": "No upcoming milestones found",
                "hint": "Check release_milestone table for available releases",
            }
        )

    # Build feature lookup by normalized (product, major.minor)
    all_feature_releases = conn.execute("SELECT fr.feature_id, fr.release FROM feature_release fr").fetchall()
    features_by_key: dict[tuple[str, str], list[int]] = {}
    for fr in all_feature_releases:
        norm_key = _normalize_release(fr["release"])
        features_by_key.setdefault(norm_key, []).append(fr["feature_id"])

    # For each release, find features and assess risk
    results = []
    for key, info in releases_info.items():
        info["milestones"].sort(key=lambda m: m["days_away"])
        next_milestone = info["milestones"][0]
        days_away = next_milestone["days_away"]

        # Find features targeting this release via normalized product+major.minor
        milestone_key = _milestone_release_key(info["product"], info["version"])
        feature_ids = features_by_key.get(milestone_key, [])
        if feature_ids:
            placeholders = ",".join("?" * len(feature_ids))
            features = conn.execute(
                f"""SELECT f.issue_key, f.title, f.issue_status, f.progress_pct,
                          f.rice_score, f.assignee
                   FROM feature f
                   WHERE f.feature_id IN ({placeholders})""",
                feature_ids,
            ).fetchall()
        else:
            features = []

        at_risk = []
        for f in features:
            pct = f["progress_pct"] or 0
            if pct < 80 and days_away <= 30:
                at_risk.append(
                    {
                        "issue_key": f["issue_key"],
                        "title": f["title"],
                        "status": f["issue_status"],
                        "progress_pct": pct,
                        "assignee": f["assignee"],
                    }
                )

        results.append(
            {
                "release": key,
                "next_milestone": next_milestone["event_type"],
                "milestone_date": next_milestone["event_date"],
                "days_away": days_away,
                "total_features": len(features),
                "at_risk_count": len(at_risk),
                "risk_level": "HIGH" if len(at_risk) > 5 else "MEDIUM" if at_risk else "LOW",
                "at_risk_features": at_risk,
                "all_milestones": info["milestones"],
            }
        )

    results.sort(key=lambda r: r["days_away"])
    return json.dumps({"releases": results, "assessed_on": today.isoformat()}, default=str)


# ---------------------------------------------------------------------------
# Cloud pricing tools
# ---------------------------------------------------------------------------


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def cloud_pricing_lookup(
    provider: str | None = None,
    service: str | None = None,
    instance_type: str | None = None,
    region: str | None = None,
    model_name: str | None = None,
    usage_type: str | None = None,
    limit: int = 50,
) -> str:
    """Look up cloud pricing for AWS services or Claude models.

    Filters (all optional, case-insensitive partial match):
      provider: 'aws' or 'anthropic'
      service: 'ec2', 's3', 'ebs', 'elb', 'data_transfer', 'vertex_ai'
      instance_type: e.g. 'm5.xlarge'
      region: e.g. 'us-east-1'
      model_name: e.g. 'claude-sonnet-4'
      usage_type: 'OnDemand', 'per-token-input', 'per-token-output'
    Provide at least one filter. Results capped at limit (default 50, max 200).
    """
    conn = _get_conn()
    if not _db_attached(conn, "pricing"):
        return json.dumps({"error": "pricing.db not attached. Run: uv run scripts/fetch_pricing.py"})

    if not any([provider, service, instance_type, region, model_name, usage_type]):
        return json.dumps({"error": "Provide at least one filter"})

    limit = min(limit, MAX_QUERY_ROWS)
    conditions, params = [], []

    if provider:
        conditions.append("provider = ?")
        params.append(provider.lower())
    if service:
        conditions.append("service = ?")
        params.append(service.lower())
    if instance_type:
        conditions.append("instance_type LIKE ?")
        params.append(f"%{instance_type}%")
    if region:
        conditions.append("region LIKE ?")
        params.append(f"%{region}%")
    if model_name:
        conditions.append("model_name LIKE ?")
        params.append(f"%{model_name}%")
    if usage_type:
        conditions.append("usage_type LIKE ?")
        params.append(f"%{usage_type}%")

    where = " AND ".join(conditions)
    sql = f"""
    SELECT provider, service, region, instance_type, instance_family,
           vcpu, memory_gb, storage_type, usage_type, unit,
           price_per_unit, model_name, description, effective_date
    FROM pricing.cloud_pricing
    WHERE {where}
    ORDER BY provider, service, price_per_unit
    LIMIT ?
    """
    params.append(limit)
    rows = conn.execute(sql, params).fetchall()
    return json.dumps({"pricing": _rows_to_dicts(rows), "count": len(rows)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def rosa_cluster_costs() -> str:
    """Get estimated monthly costs for OpenShift/ROSA cluster nodes.

    Joins instance types with EC2 pricing to estimate costs.
    Requires pricing.db with both AWS pricing and ROSA cluster discovery data.
    """
    conn = _get_conn()
    if not _db_attached(conn, "pricing"):
        return json.dumps({"error": "pricing.db not attached. Run: uv run scripts/fetch_pricing.py"})

    rows = conn.execute("SELECT * FROM pricing.v_rosa_estimated_cost ORDER BY estimated_monthly_cost DESC").fetchall()
    if not rows:
        return json.dumps({"error": "No ROSA cluster data. Run: uv run scripts/fetch_pricing.py (requires oc access)"})
    return json.dumps({"clusters": _rows_to_dicts(rows), "count": len(rows)}, default=str)


# ---------------------------------------------------------------------------
# GitHub tools
# ---------------------------------------------------------------------------


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def github_org_summary() -> str:
    """Get org-wide GitHub stats: repos, commits, PRs, issues, releases, contributors, LOC by language.

    Requires github.db to be built via: uv run scripts/fetch_github.py --org YOUR_ORG
    """
    conn = _get_conn()
    if not _db_attached(conn, "github"):
        return json.dumps({"error": "github.db not attached. Run: uv run scripts/fetch_github.py --org YOUR_ORG"})

    stats = conn.execute("SELECT * FROM github.v_gh_org_stats").fetchone()
    result = dict(stats)

    # Top languages
    langs = conn.execute(
        """SELECT language, SUM(bytes) AS total_bytes
           FROM github.gh_repo_language
           GROUP BY language ORDER BY total_bytes DESC LIMIT 20"""
    ).fetchall()
    result["languages"] = _rows_to_dicts(langs)

    # Top repos by commits
    top_repos = conn.execute(
        """SELECT name, commit_count, pr_count, merged_pr_count, issue_count,
                  contributor_count, total_language_bytes
           FROM github.v_gh_repo_summary
           ORDER BY commit_count DESC LIMIT 10"""
    ).fetchall()
    result["top_repos"] = _rows_to_dicts(top_repos)

    return json.dumps(result, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def search_github_repos(
    name: str | None = None,
    language: str | None = None,
    topic: str | None = None,
    limit: int = 50,
) -> str:
    """Search GitHub repos by name, language, or topic. All filters are partial match.

    Provide at least one parameter. Returns repo details with language breakdown.
    Returns up to `limit` repos (default 50, max 200).
    """
    if not any([name, language, topic]):
        return json.dumps({"error": "Provide at least one of: name, language, topic"})

    conn = _get_conn()
    if not _db_attached(conn, "github"):
        return json.dumps({"error": "github.db not attached. Run: uv run scripts/fetch_github.py --org YOUR_ORG"})

    limit = min(limit, MAX_QUERY_ROWS)
    conditions, params = [], []
    if name:
        conditions.append("r.name LIKE ?")
        params.append(f"%{name}%")
    if language:
        conditions.append(
            "EXISTS (SELECT 1 FROM github.gh_repo_language l WHERE l.repo_id = r.repo_id AND l.language LIKE ?)"
        )
        params.append(f"%{language}%")
    if topic:
        conditions.append(
            "EXISTS (SELECT 1 FROM github.gh_repo_topic t WHERE t.repo_id = r.repo_id AND t.topic LIKE ?)"
        )
        params.append(f"%{topic}%")

    where = " AND ".join(conditions)
    repos = conn.execute(
        f"""SELECT r.repo_id, r.name, r.description, r.html_url, r.default_branch,
                   r.is_fork, r.is_archived, r.stars, r.forks, r.open_issues,
                   r.size_kb, r.created_at, r.pushed_at,
                   r.commit_count, r.pr_count, r.merged_pr_count, r.issue_count,
                   r.contributor_count, r.topics, r.total_language_bytes
            FROM github.v_gh_repo_summary r
            WHERE {where}
            ORDER BY r.commit_count DESC
            LIMIT ?""",
        [*params, limit],
    ).fetchall()

    # Batch-load languages for all repos to avoid N+1 queries
    repo_ids = [repo["repo_id"] for repo in repos]
    langs_by_repo: dict[int, list[dict]] = {}
    if repo_ids:
        ph = ",".join("?" * len(repo_ids))
        lang_rows = conn.execute(
            f"SELECT repo_id, language, bytes FROM github.gh_repo_language WHERE repo_id IN ({ph}) ORDER BY bytes DESC",
            repo_ids,
        ).fetchall()
        for lr in lang_rows:
            langs_by_repo.setdefault(lr["repo_id"], []).append({"language": lr["language"], "bytes": lr["bytes"]})

    results = []
    for repo in repos:
        r = dict(repo)
        rid = r.pop("repo_id")
        r["languages"] = langs_by_repo.get(rid, [])
        results.append(r)

    return json.dumps({"repos": results, "count": len(results)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def search_github_commits(
    repo: str | None = None,
    author: str | None = None,
    keyword: str | None = None,
    since: str | None = None,
    until: str | None = None,
    limit: int = 50,
) -> str:
    """Search GitHub commits with optional filters. All are case-insensitive partial match.

    Provide at least one filter. since/until are ISO dates (e.g. 2026-03-01).
    """
    if not any([repo, author, keyword, since, until]):
        return json.dumps({"error": "Provide at least one filter"})

    conn = _get_conn()
    if not _db_attached(conn, "github"):
        return json.dumps({"error": "github.db not attached. Run: uv run scripts/fetch_github.py --org YOUR_ORG"})

    limit = min(limit, MAX_QUERY_ROWS)
    conditions, params = [], []

    if repo:
        conditions.append("r.name LIKE ?")
        params.append(f"%{repo}%")
    if author:
        conditions.append("c.author_login LIKE ?")
        params.append(f"%{author}%")
    if keyword:
        conditions.append("c.message LIKE ?")
        params.append(f"%{keyword}%")
    if since:
        conditions.append("c.author_date >= ?")
        params.append(since)
    if until:
        conditions.append("c.author_date <= ?")
        params.append(until)

    where = " AND ".join(conditions)
    rows = conn.execute(
        f"""SELECT r.name AS repo, c.sha, c.author_login, c.author_date, c.message
            FROM github.gh_commit c
            JOIN github.gh_repo r ON c.repo_id = r.repo_id
            WHERE {where}
            ORDER BY c.author_date DESC
            LIMIT ?""",
        [*params, limit],
    ).fetchall()
    return json.dumps({"commits": _rows_to_dicts(rows), "count": len(rows)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def search_github_prs(
    repo: str | None = None,
    state: str | None = None,
    author: str | None = None,
    keyword: str | None = None,
    limit: int = 50,
) -> str:
    """Search GitHub pull requests. state can be: open, closed, merged (merged_at IS NOT NULL).

    Provide at least one filter. All text filters are partial match.
    """
    if not any([repo, state, author, keyword]):
        return json.dumps({"error": "Provide at least one filter"})

    conn = _get_conn()
    if not _db_attached(conn, "github"):
        return json.dumps({"error": "github.db not attached. Run: uv run scripts/fetch_github.py --org YOUR_ORG"})

    limit = min(limit, MAX_QUERY_ROWS)
    conditions, params = [], []

    if repo:
        conditions.append("r.name LIKE ?")
        params.append(f"%{repo}%")
    if state:
        if state.lower() == "merged":
            conditions.append("p.merged_at IS NOT NULL")
        else:
            conditions.append("p.state LIKE ?")
            params.append(f"%{state}%")
    if author:
        conditions.append("p.author_login LIKE ?")
        params.append(f"%{author}%")
    if keyword:
        conditions.append("p.title LIKE ?")
        params.append(f"%{keyword}%")

    where = " AND ".join(conditions)
    rows = conn.execute(
        f"""SELECT r.name AS repo, p.number, p.title, p.state, p.author_login,
                   p.created_at, p.updated_at, p.merged_at,
                   p.additions, p.deletions, p.changed_files, p.html_url,
                   (SELECT GROUP_CONCAT(l.label, ', ')
                    FROM github.gh_pr_label l WHERE l.pr_id = p.pr_id) AS labels,
                   (SELECT COUNT(*) FROM github.gh_pr_review rv
                    WHERE rv.pr_id = p.pr_id AND rv.state = 'APPROVED') AS approvals
            FROM github.gh_pull_request p
            JOIN github.gh_repo r ON p.repo_id = r.repo_id
            WHERE {where}
            ORDER BY p.updated_at DESC
            LIMIT ?""",
        [*params, limit],
    ).fetchall()
    return json.dumps({"prs": _rows_to_dicts(rows), "count": len(rows)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def search_github_issues(
    repo: str | None = None,
    state: str | None = None,
    author: str | None = None,
    keyword: str | None = None,
    limit: int = 50,
) -> str:
    """Search GitHub issues (not PRs). All filters are case-insensitive partial match.

    Provide at least one filter.
    """
    if not any([repo, state, author, keyword]):
        return json.dumps({"error": "Provide at least one filter"})

    conn = _get_conn()
    if not _db_attached(conn, "github"):
        return json.dumps({"error": "github.db not attached. Run: uv run scripts/fetch_github.py --org YOUR_ORG"})

    limit = min(limit, MAX_QUERY_ROWS)
    conditions, params = [], []

    if repo:
        conditions.append("r.name LIKE ?")
        params.append(f"%{repo}%")
    if state:
        conditions.append("i.state LIKE ?")
        params.append(f"%{state}%")
    if author:
        conditions.append("i.author_login LIKE ?")
        params.append(f"%{author}%")
    if keyword:
        conditions.append("i.title LIKE ?")
        params.append(f"%{keyword}%")

    where = " AND ".join(conditions)
    rows = conn.execute(
        f"""SELECT r.name AS repo, i.number, i.title, i.state, i.author_login,
                   i.created_at, i.updated_at, i.closed_at, i.html_url,
                   (SELECT GROUP_CONCAT(l.label, ', ')
                    FROM github.gh_issue_label l WHERE l.issue_id = i.issue_id) AS labels
            FROM github.gh_issue i
            JOIN github.gh_repo r ON i.repo_id = r.repo_id
            WHERE {where}
            ORDER BY i.updated_at DESC
            LIMIT ?""",
        [*params, limit],
    ).fetchall()
    return json.dumps({"issues": _rows_to_dicts(rows), "count": len(rows)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def github_code_stats(repo: str | None = None) -> str:
    """Get LOC stats by language and weekly code frequency. Per-repo or org-wide.

    Without repo parameter, returns org-wide totals.
    """
    conn = _get_conn()
    if not _db_attached(conn, "github"):
        return json.dumps({"error": "github.db not attached. Run: uv run scripts/fetch_github.py --org YOUR_ORG"})

    result: dict = {}

    if repo:
        repo_row = conn.execute(
            "SELECT repo_id, name FROM github.gh_repo WHERE name LIKE ?",
            (f"%{repo}%",),
        ).fetchone()
        if not repo_row:
            return json.dumps({"error": f"No repo matching '{repo}'"})

        repo_id = repo_row["repo_id"]
        result["repo"] = repo_row["name"]

        langs = conn.execute(
            "SELECT language, bytes FROM github.gh_repo_language WHERE repo_id = ? ORDER BY bytes DESC",
            (repo_id,),
        ).fetchall()
        result["languages"] = _rows_to_dicts(langs)
        result["total_bytes"] = sum(r["bytes"] for r in langs)

        freq = conn.execute(
            """SELECT week_ts, additions, deletions FROM github.gh_code_frequency
               WHERE repo_id = ? ORDER BY week_ts DESC LIMIT 52""",
            (repo_id,),
        ).fetchall()
        result["weekly_frequency"] = _rows_to_dicts(freq)
        result["total_additions"] = sum(r["additions"] for r in freq)
        result["total_deletions"] = sum(abs(r["deletions"]) for r in freq)
    else:
        langs = conn.execute(
            """SELECT language, SUM(bytes) AS total_bytes, COUNT(DISTINCT repo_id) AS repo_count
               FROM github.gh_repo_language
               GROUP BY language ORDER BY total_bytes DESC"""
        ).fetchall()
        result["languages"] = _rows_to_dicts(langs)
        result["total_bytes"] = sum(r["total_bytes"] for r in langs)

        freq = conn.execute(
            """SELECT week_ts, SUM(additions) AS additions, SUM(deletions) AS deletions
               FROM github.gh_code_frequency
               GROUP BY week_ts ORDER BY week_ts DESC LIMIT 52"""
        ).fetchall()
        result["weekly_frequency"] = _rows_to_dicts(freq)
        result["total_additions"] = sum(r["additions"] for r in freq)
        result["total_deletions"] = sum(abs(r["deletions"]) for r in freq)

    return json.dumps(result, default=str)


# ---------------------------------------------------------------------------
# Governance tools
# ---------------------------------------------------------------------------


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def list_documents(doc_type: str | None = None) -> str:
    """List governance documents with metadata and table of contents (no full content).

    Optional filter by doc_type: constitution, policy, standard, reference.
    """
    conn = _get_conn()
    if doc_type:
        rows = conn.execute(
            """SELECT doc_id, doc_type, title, version, category, source_file, sections
               FROM governance_document WHERE doc_type = ? ORDER BY doc_id""",
            (doc_type,),
        ).fetchall()
    else:
        rows = conn.execute(
            """SELECT doc_id, doc_type, title, version, category, source_file, sections
               FROM governance_document ORDER BY doc_type, doc_id"""
        ).fetchall()

    results = []
    for r in rows:
        sections = json.loads(r["sections"]) if r["sections"] else []
        toc = [{"heading": s["heading"], "level": s["level"]} for s in sections]
        results.append(
            {
                "doc_id": r["doc_id"],
                "doc_type": r["doc_type"],
                "title": r["title"],
                "version": r["version"],
                "category": r["category"],
                "source_file": r["source_file"],
                "table_of_contents": toc,
            }
        )

    return json.dumps({"documents": results, "count": len(results)}, default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_document(doc_id: int) -> str:
    """Get a governance document's full content by ID.

    Use list_documents() first to find the doc_id.
    """
    conn = _get_conn()
    row = conn.execute("SELECT * FROM governance_document WHERE doc_id = ?", (doc_id,)).fetchone()
    if not row:
        return json.dumps({"error": f"No document with doc_id={doc_id}"})
    return json.dumps(dict(row), default=str)


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_document_section(doc_id: int, heading: str) -> str:
    """Get a single section from a governance document by fuzzy heading match.

    Use list_documents() to see available headings, then drill into a specific section.
    """
    conn = _get_conn()
    row = conn.execute("SELECT sections FROM governance_document WHERE doc_id = ?", (doc_id,)).fetchone()
    if not row:
        return json.dumps({"error": f"No document with doc_id={doc_id}"})

    sections = json.loads(row["sections"]) if row["sections"] else []
    headings = [s["heading"] for s in sections]
    matches = get_close_matches(heading, headings, n=1, cutoff=0.4)
    if not matches:
        return json.dumps({"error": f"No section matching '{heading}'", "available": headings})

    matched = matches[0]
    for s in sections:
        if s["heading"] == matched:
            return json.dumps(
                {
                    "heading": s["heading"],
                    "level": s["level"],
                    "content": s["content"],
                }
            )

    return json.dumps({"error": "Section not found"})


@mcp.tool(
    annotations={"readOnlyHint": True, "openWorldHint": False},
)
def get_gps_version() -> str:
    """Get the GPS version and build metadata."""
    result = {"version": VERSION}

    conn = _get_conn()
    try:
        meta = conn.execute("SELECT key, value FROM _meta").fetchall()
        result["build_info"] = {r["key"]: r["value"] for r in meta}
        return json.dumps(result, default=str)
    except sqlite3.Error:
        return json.dumps(result, default=str)


# ---------------------------------------------------------------------------
# Local extensions (not tracked upstream)
# ---------------------------------------------------------------------------

_ext_dir = Path(__file__).resolve().parent / "extensions"
if _ext_dir.is_dir():
    for _ext in sorted(_ext_dir.glob("*.py")):
        if _ext.name.startswith("_"):
            continue
        try:
            importlib.import_module(f"extensions.{_ext.stem}")
        except Exception as _e:
            logging.getLogger(__name__).warning("Failed to load extension %s: %s", _ext.name, _e)

# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------

ALLOWED_HTTP_HOSTS = [
    "127.0.0.1:*",
    "localhost:*",
    "[::1]:*",
    "host.docker.internal:*",
]

# Extra hosts from env (comma-separated), e.g. OpenShift route hostnames
_extra = os.environ.get("ALLOWED_HTTP_HOSTS", "")
if _extra:
    ALLOWED_HTTP_HOSTS.extend(h.strip() for h in _extra.split(",") if h.strip())


def _configure_http(port: int = 8000) -> None:
    """Configure MCP server for HTTP transport with DNS rebinding protection."""
    from mcp.server.fastmcp.server import TransportSecuritySettings

    mcp.settings.host = "0.0.0.0"
    mcp.settings.port = port
    mcp.settings.transport_security = TransportSecuritySettings(
        enable_dns_rebinding_protection=True,
        allowed_hosts=ALLOWED_HTTP_HOSTS,
        allowed_origins=[f"http://{h}" for h in ALLOWED_HTTP_HOSTS],
    )


def _wrap_basic_auth(app):
    """Wrap a Starlette app with basic auth if GPS_AUTH_USER/GPS_AUTH_PASS are set."""
    import base64
    import secrets

    from starlette.middleware.base import BaseHTTPMiddleware
    from starlette.responses import Response

    auth_user = os.environ.get("GPS_AUTH_USER", "")
    auth_pass = os.environ.get("GPS_AUTH_PASS", "")
    if not (auth_user and auth_pass):
        return app

    expected = base64.b64encode(f"{auth_user}:{auth_pass}".encode()).decode()

    class BasicAuthMiddleware(BaseHTTPMiddleware):
        async def dispatch(self, request, call_next):
            if request.url.path != "/mcp":
                return await call_next(request)
            # Check Authorization header
            auth_header = request.headers.get("Authorization", "")
            if auth_header.startswith("Basic "):
                if secrets.compare_digest(auth_header[6:], expected):
                    return await call_next(request)
            # Check query param fallback (for clients that don't support headers)
            token = request.query_params.get("token", "")
            if token and secrets.compare_digest(token, expected):
                return await call_next(request)
            return Response("Unauthorized", status_code=401, headers={"WWW-Authenticate": "Basic"})

    app.add_middleware(BasicAuthMiddleware)
    return app


if __name__ == "__main__":
    parser = argparse.ArgumentParser(description="GPS MCP Server")
    parser.add_argument("--http", action="store_true", help="Use HTTP transport (default: stdio)")
    parser.add_argument("--port", type=int, default=8000, help="HTTP port (default: 8000)")
    args = parser.parse_args()

    if args.http:
        _configure_http(args.port)
        app = mcp.streamable_http_app()
        app = _wrap_basic_auth(app)
        import uvicorn

        uvicorn.run(app, host="0.0.0.0", port=args.port)
    else:
        mcp.run(transport="stdio")