mcp_server.py

"""CrossBeam MCP server — exposes the municipal plan-review engine as agentic tools.

Works with any MCP client (Claude Desktop, ChatGPT, n8n, Gemini, HF UI, local, or via
a tunnel). Hostable on a Hugging Face Space or a local machine.

The tools are thin wrappers over engine.py, so the MCP surface stays in lock-step with
the Streamlit app — same jurisdictions, same skills, same grounded-citation guarantees.

Transports
----------
- STDIO (default): for Claude Desktop / local clients that launch the process.
    python mcp_server.py
- HTTP/SSE (for HF Spaces, n8n, tunnels): set MCP_HTTP=1 (and optionally MCP_PORT).
    MCP_HTTP=1 MCP_PORT=7860 python mcp_server.py

LLM calls
---------
The review tools need a provider + model + API key. Supply them per-call, or set
env vars (ANTHROPIC_API_KEY / OPENAI_API_KEY / GEMINI_API_KEY / GROQ_API_KEY) and pass
just the provider name. The MCP client's own model does NOT perform the review — these
tools call the configured provider so the deterministic engine logic (scope gate,
grounding, reconciliation) always runs.

DNS-rebinding note: FastMCP auto-enables Host-header validation when it is
constructed with the default host (127.0.0.1). Behind the HF Spaces proxy that
rejects every request to /mcp with 421 Misdirected Request. Setting
`mcp.settings.host` after construction is too late — the security settings are
frozen in __init__. Both host and transport_security are therefore passed to the
constructor below.
"""

from __future__ import annotations

import base64
import os
from typing import Optional

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

import engine as eng
from jurisdiction import list_jurisdictions, load_jurisdiction_file
from providers import PROVIDERS

BASE_DIR = os.path.dirname(os.path.abspath(__file__))

mcp = FastMCP(
    "crossbeam-plan-review",
    stateless_http=True,
    json_response=True,
    host="0.0.0.0",
    transport_security=TransportSecuritySettings(enable_dns_rebinding_protection=False),
)


# ── helpers ────────────────────────────────────────────────────────────────
def _real_slugs() -> list:
    """Configured municipality slugs, excluding templates and the root default."""
    choices = list_jurisdictions(BASE_DIR)
    return [s for s in choices if not s.startswith("_") and s != "default"]


def _slug(slug: str = "") -> str:
    """Resolve the active municipality slug.

    An unknown slug raises instead of silently falling back to the first configured
    municipality. Silently substituting Calgary for an unrecognised city would hand
    the agent a confidently-cited but wrong bylaw, which is worse than no answer.
    """
    choices = list_jurisdictions(BASE_DIR)
    real = _real_slugs()
    if slug:
        if slug in choices:
            return slug
        raise ValueError(
            f"Unknown municipality '{slug}'. Configured: {', '.join(real) or 'none'}. "
            "Call list_municipalities() for the valid slugs.")
    return real[0] if real else ""


def _jur(slug: str = ""):
    choices = list_jurisdictions(BASE_DIR)
    resolved = _slug(slug)          # raises on an unknown slug
    return load_jurisdiction_file(choices.get(resolved, ""))


def _resolve_key(provider: str, api_key: str) -> str:
    if api_key:
        return api_key
    env = PROVIDERS.get(provider, {}).get("env", "")
    return os.environ.get(env, "") if env else ""


def _load(track_id: str = "", municipality: str = ""):
    """Load knowledge scoped to a municipality so multiple cities can coexist."""
    skills, loose = eng.load_knowledge(BASE_DIR, jurisdiction=_slug(municipality))
    tracks = eng.discover_tracks(skills)
    if not tracks:
        # A configured municipality with no skills yet (e.g. a freshly added city).
        # Return an explicit empty track rather than crashing on next(iter({})).
        return skills, loose, {}, {"label": "unconfigured", "scope": "", "skills": []}
    if track_id and track_id in tracks:
        tr = tracks[track_id]
    else:
        tr = tracks.get("suites") or next(iter(tracks.values()))
    return skills, loose, tracks, tr


# ── discovery tools (no API key needed) ─────────────────────────────────────
@mcp.tool()
def list_municipalities() -> dict:
    """List the jurisdictions this server is configured for (Calgary, etc.).
    Returns slugs to pass as `municipality` to other tools."""
    choices = list_jurisdictions(BASE_DIR)
    out = []
    for slug in choices:
        if slug.startswith("_") or slug == "default":
            continue
        j = load_jurisdiction_file(choices[slug])
        out.append({"slug": slug, "name": j.place,
                    "safety_framework": j.safety_short, "landuse_framework": j.landuse_short,
                    "transition": j.transition_label or None})
    return {"municipalities": out}


@mcp.tool()
def list_review_tracks(municipality: str = "") -> dict:
    """List available review tracks/types (e.g. suites, multi-residential, institutional)
    and the models available for the LLM-backed tools."""
    _, _, tracks, _ = _load("", municipality)
    return {
        "tracks": [{"id": t, "label": v["label"], "scope": v["scope"]}
                   for t, v in tracks.items()],
        "providers": {p: info["models"] for p, info in PROVIDERS.items()},
    }


@mcp.tool()
def list_knowledge(municipality: str = "") -> dict:
    """Show the knowledge base (skills + reference files) the reviews are grounded in."""
    skills, loose = eng.load_knowledge(BASE_DIR, jurisdiction=_slug(municipality))
    return {"manifest": eng.knowledge_manifest(skills, loose)}


# ── Flow 3: plan review ─────────────────────────────────────────────────────
# ═══════════════════════════════════════════════════════════════════════════
#  CLIENT-SIDE TOOLS — no API key required.
#
#  When the caller is itself an LLM (ChatGPT, Claude, Gemini, HuggingChat), it
#  makes no sense for this server to call ANOTHER model: you would pay twice and
#  wait twice. These tools instead hand the agent everything it needs — the
#  grounded knowledge, the review rules, the output schema, and deterministic
#  document extraction — and the agent's own model does the reasoning.
#
#  Use the *_with_llm tools further below only for non-LLM callers (n8n, scripts,
#  cron) that have no model of their own.
# ═══════════════════════════════════════════════════════════════════════════

@mcp.tool()
def extract_document(pdf_base64: str = "", max_pages: int = 15,
                     include_images: bool = False) -> dict:
    """Extract text (and optionally page images) from a PDF or DXF. NO API KEY NEEDED.

    Deterministic parsing via PyMuPDF/ezdxf — no LLM involved. Returns per-page text
    so an agent can read a plan set or letter it cannot otherwise open. Set
    include_images=true to also get base64 JPEGs of each page for vision models
    (omitted by default because they are large)."""
    if not pdf_base64:
        return {"error": "Provide pdf_base64 (base64-encoded PDF or DXF bytes)."}
    try:
        data = base64.b64decode(pdf_base64)
    except Exception as exc:  # noqa: BLE001
        return {"error": f"pdf_base64 is not valid base64: {exc}"}
    try:
        b64s, texts, total, dims = eng.file_to_pages("upload.pdf", data, max_pages=max_pages)
    except Exception as exc:  # noqa: BLE001
        return {"error": str(exc)}
    out = {"total_pages": total, "pages_extracted": len(texts),
           "pages": [{"page": i + 1, "sheet_id": eng.guess_sheet_id(t) or "", "text": t}
                     for i, t in enumerate(texts)]}
    if include_images:
        out["page_images_base64_jpeg"] = b64s
    return out


@mcp.tool()
def identify_site(pdf_base64: str = "", address: str = "", municipality: str = "") -> dict:
    """Step 1 of a municipal review: get the SITE ADDRESS and its LAND USE DISTRICT.
    NO API KEY NEEDED.

    Municipal reviewers work address → land use map → district → that district's
    standards. This tool does the first two steps: it extracts the site address from a
    plan set (deterministic, offline) and attempts a district lookup.

    The district lookup is best-effort. When it is unavailable or unconfigured, the
    result carries `manual_url` and an empty `district` — it never guesses, because a
    wrong district silently invalidates every downstream conclusion. Pass the district
    you determined into `get_review_kit(land_use_district=...)` or `review_plan_set`.
    """
    jur = _jur(municipality)
    detected = address
    pages = 0
    if not detected and pdf_base64:
        try:
            data = base64.b64decode(pdf_base64)
            _, texts, total, _ = eng.file_to_pages("upload.pdf", data, max_pages=15)
            detected = eng.guess_site_address(texts)
            pages = total
        except Exception as exc:  # noqa: BLE001
            return {"error": f"Could not read the document: {exc}"}
    lookup = eng.lookup_land_use_district(detected, jur=jur)
    return {
        "municipality": jur.place,
        "site_address": detected or "(not found — supply `address`)",
        "pages_scanned": pages,
        "land_use_district": lookup.get("district", ""),
        "lookup_note": lookup.get("note", ""),
        "manual_lookup_url": lookup.get("manual_url", ""),
        "next_step": ("Determine the district (manually if the lookup is empty), then call "
                      "get_review_kit with land_use_district set so the review applies that "
                      "district's standards."),
    }


@mcp.tool()
def get_review_kit(municipality: str = "", review_track: str = "suites",
                   application_date: str = "", land_use_district: str = "",
                   site_address: str = "") -> dict:
    """Everything needed to REVIEW a plan set yourself. NO API KEY NEEDED.

    Returns the jurisdiction framing, the grounded knowledge base for the chosen
    track, the critical review rules (scope gate, grounded-citation rule, category
    discipline), and the JSON output schema.

    Recommended agent workflow:
      1. extract_document(pdf_base64=...)   → the plan text/images
      2. get_review_kit(municipality=..., review_track=...)  → rules + knowledge
      3. YOUR model produces the findings, obeying `critical_rules` and citing ONLY
         clause numbers that appear verbatim in `knowledge`.
    """
    jur = _jur(municipality)
    _, _, tracks, tr = _load(review_track, municipality)
    if not tracks:
        return {"error": f"No review skills are configured for '{jur.place}' yet.",
                "municipality": jur.place,
                "how_to_fix": ("Add skill folders under skills/ whose SKILL.md frontmatter "
                               f"declares `jurisdiction: {_slug(municipality)}` (plus track, "
                               "track_label, track_scope). Until then this municipality has "
                               "framing but no grounded rules, so no review can be performed."),
                "frameworks": {"safety": jur.safety_framework,
                               "land_use": jur.landuse_framework}}
    knowledge = "\n\n".join(
        [f"=== {sk.name}/SKILL.md ===\n{sk.skill_md}" for sk in tr["skills"]] +
        [f"=== {s.key} ===\n{s.content}" for sk in tr["skills"] for s in sk.sources])
    rules = eng.build_critical_rules(tr["label"], tr["scope"],
                                     ", ".join(v["label"] for v in tracks.values()), jur)
    trans = f" (assess against {jur.transition_label})" if jur.transition_label else ""
    return {
        "municipality": jur.place,
        "review_track": {"id": review_track, "label": tr["label"], "scope": tr["scope"]},
        "frameworks": {"safety": jur.safety_framework, "land_use": jur.landuse_framework},
        "project_framing": (
            f"Municipality: {jur.place}\nReview type: {tr['label']}\n"
            f"Site address: {site_address or 'not stated'}\n"
            + (f"Land use district: {land_use_district} (apply this district's standards)\n"
               if land_use_district else
               "Land use district: NOT SUPPLIED — treat every district-dependent conclusion "
               "as unconfirmed and require it as a prior-to-decision item\n")
            + f"Intended application date: {application_date or 'not stated'}{trans}"),
        "critical_rules": rules,
        "knowledge": knowledge,
        "output_schema": eng.REVIEW_SYSTEM_TMPL.split("OUTPUT:", 1)[-1].strip(),
        "letter_format_hint": ("For a municipal-style Detailed Review, give each land-use "
                               "finding `regulation`, `standard` and `provided` fields and "
                               "render: header block (Application Number, Description, Land "
                               "Use District, Use Type, Site Address, Applicant), General "
                               "Comments, a Bylaw Discrepancies table (Regulation | Standard "
                               "| Provided with numeric deltas), Prior to Decision "
                               "Requirements, then Advisory Comments."),
        "note": ("Cite ONLY clause/section numbers that appear verbatim in `knowledge`. "
                 "If a rule is real but its number is not present, name the reference file "
                 "instead or record an information gap — never invent a number."),
    }


@mcp.tool()
def get_corrections_kit(municipality: str = "") -> dict:
    """Everything needed to INTERPRET a corrections letter yourself. NO API KEY NEEDED.

    Returns the grounded knowledge, the honesty rules, and the output schema for
    turning a municipal corrections/Detailed-Review letter into an item-by-item
    analysis and a draft response.

    The honesty rule is the important part: a draft response must NEVER claim a
    correction has been resolved. Every resolution belongs to the applicant and is
    represented by an [APPLICANT: ...] placeholder."""
    jur = _jur(municipality)
    # Scope to the requested municipality. Loading unscoped pulled EVERY city's
    # skills, so a Toronto request was framed as Toronto but grounded in Calgary
    # clauses — grounded, provenance-tagged, and wrong.
    skills, _ = eng.load_knowledge(BASE_DIR, jurisdiction=_slug(municipality))
    sk = [x for t in eng.discover_tracks(skills).values() for x in t["skills"]]
    if not sk:
        return {"error": f"No review knowledge is configured for '{jur.place}' yet.",
                "municipality": jur.place,
                "how_to_fix": ("Add skill folders under skills/ whose SKILL.md frontmatter "
                               f"declares `jurisdiction: {_slug(municipality)}`.")}
    knowledge = "\n\n".join(
        [f"=== {x.name}/SKILL.md ===\n{x.skill_md}" for x in sk] +
        [f"=== {s.key} ===\n{s.content}" for x in sk for s in x.sources])
    system = eng.CORRECTIONS_SYSTEM_TMPL.format(knowledge="<knowledge supplied separately>",
                                                **eng._jur_fields(jur))
    return {
        "municipality": jur.place,
        "frameworks": {"safety": jur.safety_framework, "land_use": jur.landuse_framework},
        "rules_and_schema": system,
        "knowledge": knowledge,
        "honesty_rule": ("NEVER state or imply a correction has been resolved. Every "
                         "resolution is an [APPLICANT: ...] placeholder the applicant fills in."),
    }


# ═══════════════════════════════════════════════════════════════════════════
#  SERVER-SIDE TOOLS — these DO call an LLM provider (need an API key).
#  Use them from non-LLM callers: n8n, scripts, schedulers.
# ═══════════════════════════════════════════════════════════════════════════

@mcp.tool()
def review_plan_set(
    pdf_base64: str,
    provider: str = "ChatGPT (OpenAI)",
    model: str = "gpt-4.1",
    api_key: str = "",
    municipality: str = "",
    review_track: str = "suites",
    application_date: str = "",
    project_description: str = "",
    max_pages: int = 15,
    base_url: str = "",
) -> dict:
    """Review a plan set (PDF, base64-encoded) against a municipality's rules and produce
    grounded correction findings. This is the city-side pre-screen. Returns the structured
    result (submission_check, findings, summary) plus a Markdown report.

    The scope gate, grounded-citation rule, and verdict reconciliation all run in the
    engine — the findings never invent clause numbers not in the loaded knowledge."""
    key = _resolve_key(provider, api_key)
    if not key:
        return {"error": f"No API key for {provider}. Pass api_key or set the env var."}
    jur = _jur(municipality)
    skills_all, loose, tracks, tr = _load(review_track, municipality)
    skills = tr["skills"]
    data = base64.b64decode(pdf_base64)
    b64s, texts, total, dims = eng.file_to_pages("upload.pdf", data, max_pages=max_pages)
    trans = f" (assess against {jur.transition_label})" if jur.transition_label else ""
    desc = (f"Municipality: {jur.place}\nReview type: {tr['label']}\n"
            f"Intended application date: {application_date or 'not stated'}{trans}\n"
            f"Description: {project_description or 'not provided'}")
    selected, routed = eng.route(provider, model, key, skills, loose, desc,
                                 base_url=base_url, jur=jur)
    result = eng.run_review(provider, model, key, skills, selected, desc, b64s, texts,
                            track={"label": tr["label"], "scope": tr["scope"]},
                            available_tracks=[v["label"] for v in tracks.values()],
                            base_url=base_url, jur=jur)
    report = eng.render_report(result, desc, routed, selected, len(b64s), total,
                               provider, model, jur=jur)
    date_warning = eng.check_application_date(application_date, texts)
    city_letter = eng.render_city_letter(result, desc, len(b64s), total, provider, model,
                                         jur=jur, date_warning=date_warning)
    return {"result": result, "report_markdown": report,
            "city_review_letter_markdown": city_letter,
            "date_warning": date_warning,
            "pages_reviewed": len(b64s), "total_pages": total, "routing": routed}


# ── Flow 1: corrections response ────────────────────────────────────────────
@mcp.tool()
def analyze_corrections_letter(
    letter_base64: str,
    provider: str = "ChatGPT (OpenAI)",
    model: str = "gpt-4.1",
    api_key: str = "",
    municipality: str = "",
    plan_context: str = "",
    applicant_name: str = "",
    base_url: str = "",
) -> dict:
    """Interpret a municipal corrections / Detailed-Review letter (PDF, base64) into an
    item-by-item analysis grounded in the knowledge base, plus a DRAFT response letter.

    The draft NEVER claims a correction is resolved — every resolution is an
    [APPLICANT: ...] placeholder for the applicant to fill in. Returns the structured
    analysis, an analysis Markdown, and the draft response letter Markdown."""
    key = _resolve_key(provider, api_key)
    if not key:
        return {"error": f"No API key for {provider}. Pass api_key or set the env var."}
    jur = _jur(municipality)
    skills, loose = eng.load_knowledge(BASE_DIR, jurisdiction=_slug(municipality))
    sk = [x for t in eng.discover_tracks(skills).values() for x in t["skills"]]
    sources = [s for x in sk for s in x.sources]
    data = base64.b64decode(letter_base64)
    lb64, ltext, ltot, _ = eng.file_to_pages("letter.pdf", data, max_pages=15)
    result = eng.run_corrections(provider, model, key, sk, sources, "\n\n".join(ltext),
                                 plan_context=plan_context, images_b64=lb64,
                                 base_url=base_url, jur=jur)
    return {
        "result": result,
        "analysis_markdown": eng.render_corrections_analysis(result, jur=jur),
        "draft_response_letter_markdown": eng.render_response_letter(
            result, jur=jur, applicant_name=applicant_name),
    }


# ── Flow 2: pre-submission checklist ────────────────────────────────────────
@mcp.tool()
def generate_checklist(
    provider: str = "ChatGPT (OpenAI)",
    model: str = "gpt-4.1",
    api_key: str = "",
    municipality: str = "",
    review_track: str = "suites",
    project_description: str = "",
    base_url: str = "",
) -> dict:
    """Generate a pre-submission checklist (required drawings, data/calcs, common pitfalls)
    for a project type in a municipality, drawn from the loaded knowledge. Returns the
    structured checklist and a Markdown version."""
    key = _resolve_key(provider, api_key)
    if not key:
        return {"error": f"No API key for {provider}. Pass api_key or set the env var."}
    jur = _jur(municipality)
    _, _, tracks, tr = _load(review_track, municipality)
    result = eng.run_checklist(provider, model, key, tr["skills"],
                               {"label": tr["label"], "scope": tr["scope"]},
                               project_desc=project_description, base_url=base_url, jur=jur)
    return {"result": result,
            "checklist_markdown": eng.render_checklist(result, tr["label"], jur=jur)}


@mcp.custom_route("/health", methods=["GET"])
async def _health(request):
    """Plain-text liveness check — open this in a browser to tell a sleeping/failed
    Space apart from an MCP-protocol problem.

    - Page loads with "ok" → the Space is up; any client error is protocol/config.
    - Page does not load at all → the Space is asleep, building, or crashed.
    """
    from starlette.responses import PlainTextResponse
    try:
        tools = await mcp.list_tools()
        n = len(tools)
    except Exception as exc:  # noqa: BLE001
        return PlainTextResponse(f"degraded: tool registry error: {exc}", status_code=500)
    return PlainTextResponse(f"ok\ntools={n}\nendpoint=/mcp\n")


@mcp.custom_route("/", methods=["GET"])
async def _landing(request):
    """Status page at `/`.

    An MCP-only server has no route at `/`, so the Hugging Face App tab would
    otherwise show a bare "Not Found" and you couldn't tell a healthy server from a
    broken one. This renders the endpoint URL and the tool list instead.
    """
    from starlette.responses import HTMLResponse

    choices = list_jurisdictions(BASE_DIR)
    muns = ", ".join(s for s in choices if not s.startswith("_") and s != "default") or "—"
    # HF Spaces terminates TLS at its proxy, so request.base_url reports http://
    # inside the container. Emitting that URL makes clients (e.g. HuggingChat)
    # reject it as insecure — honour the forwarded proto and default to https.
    base = str(request.base_url).rstrip("/")
    proto = request.headers.get("x-forwarded-proto", "")
    host = request.headers.get("host", "")
    if proto:
        base = f"{proto}://{host}"
    elif base.startswith("http://") and not host.startswith(("localhost", "127.0.0.1")):
        base = "https://" + base[len("http://"):]
    # Dynamic tool table — generated from the live registry so it can never go
    # stale again, with a badge showing which tools need no API key.
    KEYLESS = {"list_municipalities", "list_review_tracks", "list_knowledge",
               "extract_document", "get_review_kit", "get_corrections_kit"}
    tools = await mcp.list_tools()

    def _row(t):
        desc = (t.description or "").strip().splitlines()[0]
        badge = ('<span style="background:#0a7b34;color:#fff;border-radius:99px;'
                 'padding:1px 8px;font-size:11px;margin-left:6px">no API key</span>'
                 if t.name in KEYLESS else
                 '<span style="background:#8a5a00;color:#fff;border-radius:99px;'
                 'padding:1px 8px;font-size:11px;margin-left:6px">needs provider key</span>')
        return f"<tr><td style='white-space:nowrap'><code>{t.name}</code>{badge}</td><td>{desc}</td></tr>"

    rows = ("".join(_row(t) for t in tools if t.name in KEYLESS)
            + "".join(_row(t) for t in tools if t.name not in KEYLESS))
    return HTMLResponse(f"""<!doctype html><meta charset="utf-8">
<title>CrossBeam MCP Server</title>
<style>
 body{{font:15px/1.6 system-ui,sans-serif;max-width:760px;margin:40px auto;padding:0 20px;color:#111}}
 .ok{{display:inline-block;background:#0a7b34;color:#fff;padding:3px 10px;border-radius:99px;font-size:12px}}
 code{{background:#f3f4f6;padding:2px 6px;border-radius:4px}}
 table{{border-collapse:collapse;width:100%;margin-top:10px}}
 td{{border-top:1px solid #e5e7eb;padding:7px 6px;vertical-align:top}}
 .u{{background:#111;color:#fff;padding:10px 14px;border-radius:6px;display:block;margin:10px 0}}
</style>
<h1>🔌 CrossBeam MCP Server <span class="ok">running</span></h1>
<p>Municipal plan review as MCP tools. This server has <b>no web UI by design</b> —
point an MCP client at the endpoint below.</p>
<span class="u">{base}/mcp</span>
<p>Municipalities configured: <b>{muns}</b></p>
<p style="font-size:13px;color:#555">Liveness check: <a href="{base}/health"><code>{base}/health</code></a>
— if that page loads, the server is up and any client error is a protocol/config issue on the
client side. If it does not load, the Space is asleep, building, or crashed (open the Space and
check <b>Logs</b>).</p>
<h3>Tools</h3><table>{rows}</table>
<p style="color:#666;font-size:13px;margin-top:22px">
<b>Agent clients (ChatGPT, Claude, Gemini, HuggingChat) need no API key</b> — use
<code>extract_document</code> + <code>get_review_kit</code> /
<code>get_corrections_kit</code> and let your own model do the review, citing only
clauses present in the returned knowledge. The "needs provider key" tools are for
non-LLM callers (n8n, scripts): pass <code>api_key</code> per call or set a Space
secret (<code>OPENAI_API_KEY</code>, <code>ANTHROPIC_API_KEY</code>,
<code>GEMINI_API_KEY</code>, <code>GROQ_API_KEY</code>).</p>""")


if __name__ == "__main__":
    if os.environ.get("MCP_HTTP") == "1":
        import uvicorn
        from starlette.middleware.cors import CORSMiddleware

        app = mcp.streamable_http_app()
        app.add_middleware(
            CORSMiddleware,
            allow_origins=["*"],
            allow_methods=["GET", "POST", "DELETE", "OPTIONS"],
            allow_headers=["*"],
            expose_headers=["Mcp-Session-Id"],
        )
        uvicorn.run(app, host="0.0.0.0",
                    port=int(os.environ.get("MCP_PORT", "7860")))
    else:
        mcp.run()