"""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="", **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 = ('no API key' if t.name in KEYLESS else 'needs provider key') return f"{t.name}{badge}{desc}" 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""" CrossBeam MCP Server

🔌 CrossBeam MCP Server running

Municipal plan review as MCP tools. This server has no web UI by design — point an MCP client at the endpoint below.

{base}/mcp

Municipalities configured: {muns}

Liveness check: {base}/health — 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 Logs).

Tools

{rows}

Agent clients (ChatGPT, Claude, Gemini, HuggingChat) need no API key — use extract_document + get_review_kit / get_corrections_kit 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 api_key per call or set a Space secret (OPENAI_API_KEY, ANTHROPIC_API_KEY, GEMINI_API_KEY, GROQ_API_KEY).

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