"""Zoning Firehose MCP server — municipal zoning data as tools for Claude and other
MCP clients.

A single-file stdio MCP server (JSON-RPC 2.0, newline-delimited) that bridges to the
Zoning Firehose public REST API (see /api-docs on your metro's site). No dependencies
beyond Python 3.11+ and httpx.

Zoning Firehose covers one metro per host: DFW at https://zoningfirehose.com, every
other metro at its subdomain (https://atlanta.zoningfirehose.com, ...). Point
ZONING_API_BASE at the metro you subscribe to; without it you get DFW.

Setup
-----
1. Get an API key. Mint one yourself at /profile on the Metro or Corporate
   plan; on any other plan, ask us at questions@zoningfirehose.com.
2. Add to your Claude Desktop / Claude Code MCP config:

    {
      "mcpServers": {
        "zoning-firehose": {
          "command": "python",
          "args": ["/path/to/mcp_server.py"],
          "env": {
            "ZONING_API_KEY": "zf_your_key_here",
            "ZONING_API_BASE": "https://atlanta.zoningfirehose.com"
          }
        }
      }
    }

Environment:
    ZONING_API_KEY   required — your API key
    ZONING_API_BASE  optional — your metro's host; defaults to
                     https://zoningfirehose.com (the DFW metro)
"""

import json
import os
import sys
from typing import Any
from urllib.parse import quote

import httpx

PROTOCOL_VERSION = "2025-06-18"
SERVER_INFO = {"name": "zoning-firehose", "version": "1.2.0"}

_FILTER_PROPS: dict[str, Any] = {
    "municipality": {
        "type": "string",
        "description": "City name within your metro, e.g. 'McKinney' or 'Sandy Springs'",
    },
    "county": {
        "type": "string",
        "description": (
            "County name within your metro (no ' County' suffix), e.g. Collin/Tarrant "
            "for DFW, Fulton/Cobb for Atlanta. Use list_municipalities to see them all."
        ),
    },
    "case_type": {
        "type": "string",
        # Must stay identical to worker.models.CaseType, which is what the REST
        # API validates against. It did not: site_plan, text_amendment and
        # annexation were added to the API and never here, so a schema-conforming
        # MCP client could not ask for three case types the product supports --
        # including site_plan, which is 11% of the corpus. This file is
        # downloaded and run on the customer's machine, so it cannot import the
        # definition; tests/test_mcp_schema.py compares the two instead.
        "enum": [
            "rezoning",
            "specific_use_permit",
            "plat",
            "pd_amendment",
            "comprehensive_plan_amendment",
            "variance",
            "site_plan",
            "text_amendment",
            "annexation",
            "other",
        ],
    },
    "status": {
        "type": "string",
        "enum": ["new", "updated", "continued", "approved", "denied", "withdrawn"],
        "description": (
            "Where the case is in OUR pipeline, not how a vote went. We read agendas, "
            "which are published before the meeting; minutes are not ingested, so "
            "outcomes are not observable. 'approved', 'denied' and 'withdrawn' are "
            "accepted by the schema but are never set by the pipeline and will always "
            "return nothing — an empty result means we do not track outcomes, NOT that "
            "no case was approved. Use 'staff_recommendation' for the planner's "
            "recommendation, which is a pre-hearing opinion and not a decision."
        ),
    },
    "hearing_after": {"type": "string", "description": "ISO date (YYYY-MM-DD), inclusive"},
    "hearing_before": {"type": "string", "description": "ISO date (YYYY-MM-DD), inclusive"},
    "petitioner": {
        "type": "string",
        "description": (
            "Exact applicant/petitioner name — the party that filed. Use this to pull a "
            "competitor's or partner's full filing history, e.g. 'Acme Land Partners LLC'."
        ),
    },
    "q": {"type": "string", "description": "Keyword: case number, LLC, street, agenda text"},
    "limit": {"type": "integer", "description": "Max results (default 50, max 200)"},
    "offset": {"type": "integer", "description": "Paging offset"},
}

TOOLS: list[dict[str, Any]] = [
    {
        "name": "search_cases",
        "description": (
            "Search validated municipal zoning cases (rezonings, SUPs, plats, PD "
            "amendments) extracted from P&Z agendas across your subscribed metro "
            "(DFW, Houston, Atlanta, Phoenix, Nashville, or Miami — set by "
            "ZONING_API_BASE), newest first. All filters optional."
        ),
        "inputSchema": {"type": "object", "properties": _FILTER_PROPS},
    },
    {
        "name": "get_petitioner_history",
        "description": (
            "Every tracked case filed by one applicant in your metro — the fastest way to "
            "answer 'what is <developer> working on?' or 'where is <competitor> "
            "assembling land?'. Returns their cases with cities, hearing dates and "
            "proposed uses. Exact name match; use search_cases with q= to find the "
            "spelling first if unsure."
        ),
        "inputSchema": {
            "type": "object",
            "properties": {
                "petitioner": {
                    "type": "string",
                    "description": "Exact applicant name, e.g. 'Acme Land Partners LLC'",
                },
                "limit": {"type": "integer", "description": "Max results (default 100)"},
            },
            "required": ["petitioner"],
        },
    },
    {
        "name": "get_case",
        "description": "Fetch one zoning case by municipality and case number.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "municipality": {
                    "type": "string",
                    "description": "City name within your metro, e.g. 'McKinney'",
                },
                "case_number": {"type": "string", "description": "e.g. 'Z-2026-014'"},
            },
            "required": ["municipality", "case_number"],
        },
    },
    {
        "name": "list_municipalities",
        "description": (
            "List every municipality Zoning Firehose covers in your subscribed metro: "
            "county, agenda platform, P&Z meeting cadence, live status, and tracked "
            "case count."
        ),
        "inputSchema": {"type": "object", "properties": {}},
    },
]


def _api_get(path: str, params: dict[str, Any] | None = None) -> Any:
    """GET a public REST API endpoint with the configured key; returns parsed JSON."""
    key = os.environ.get("ZONING_API_KEY", "")
    if not key:
        raise RuntimeError("ZONING_API_KEY environment variable is not set")
    base = os.environ.get("ZONING_API_BASE", "https://zoningfirehose.com").rstrip("/")
    resp = httpx.get(
        f"{base}{path}",
        params={k: v for k, v in (params or {}).items() if v not in (None, "")},
        headers={"X-API-Key": key},
        timeout=30.0,
        follow_redirects=True,
    )
    resp.raise_for_status()
    return resp.json()


def _call_tool(name: str, args: dict[str, Any]) -> Any:
    if name == "search_cases":
        return _api_get("/api/v1/cases", args)
    if name == "get_petitioner_history":
        return _api_get(
            "/api/v1/cases",
            {
                "petitioner": args.get("petitioner", ""),
                "limit": args.get("limit", 100),
            },
        )
    if name == "get_case":
        muni = quote(str(args.get("municipality", "")).lower())
        case_number = quote(str(args.get("case_number", "")), safe="")
        return _api_get(f"/api/v1/cases/{muni}/{case_number}")
    if name == "list_municipalities":
        return _api_get("/api/v1/municipalities")
    raise ValueError(f"unknown tool: {name}")


def _result(req_id: Any, result: dict[str, Any]) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": req_id, "result": result}


def _error(req_id: Any, code: int, message: str) -> dict[str, Any]:
    return {"jsonrpc": "2.0", "id": req_id, "error": {"code": code, "message": message}}


def handle_request(req: dict[str, Any]) -> dict[str, Any] | None:
    """One JSON-RPC request -> response dict, or None for notifications."""
    method = req.get("method", "")
    req_id = req.get("id")
    if method.startswith("notifications/"):
        return None  # notifications get no response
    if method == "initialize":
        return _result(
            req_id,
            {
                "protocolVersion": PROTOCOL_VERSION,
                "capabilities": {"tools": {}},
                "serverInfo": SERVER_INFO,
            },
        )
    if method == "ping":
        return _result(req_id, {})
    if method == "tools/list":
        return _result(req_id, {"tools": TOOLS})
    if method == "tools/call":
        params = req.get("params") or {}
        name = params.get("name", "")
        args = params.get("arguments") or {}
        try:
            data = _call_tool(name, args)
        except Exception as exc:  # tool errors are results, not protocol errors
            return _result(
                req_id,
                {"content": [{"type": "text", "text": f"Error: {exc}"}], "isError": True},
            )
        return _result(
            req_id,
            {
                "content": [{"type": "text", "text": json.dumps(data, indent=2)}],
                "isError": False,
            },
        )
    return _error(req_id, -32601, f"method not found: {method}")


def main() -> None:
    """Serve newline-delimited JSON-RPC over stdio until EOF."""
    for line in sys.stdin:
        line = line.strip()
        if not line:
            continue
        try:
            req = json.loads(line)
        except json.JSONDecodeError:
            resp: dict[str, Any] | None = _error(None, -32700, "parse error")
        else:
            resp = handle_request(req)
        if resp is not None:
            sys.stdout.write(json.dumps(resp) + "\n")
            sys.stdout.flush()


if __name__ == "__main__":
    main()