""" Web application: FastAPI REST endpoints + FastMCP mount + React static. Layout: / → React SPA (index.html + assets) /api/... → FastAPI REST endpoints used by the React UI /api/docs → Swagger UI for the REST API /mcp → FastMCP streamable-http endpoint (for LLM clients) /health → liveness probe (also mounted under /api/health) """ from __future__ import annotations import logging from pathlib import Path from typing import Any from fastapi import FastAPI, HTTPException, Query from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import StreamingResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from starlette.responses import FileResponse, JSONResponse from nadlan_mcp.fastmcp_server import ( client as govmap_client, ) from nadlan_mcp.fastmcp_server import ( decisive_appraiser_client, mcp, ) from nadlan_mcp.govil.client import ALLOWED_PDF_HOSTS logger = logging.getLogger(__name__) # ────────────────────────────────────────────────────────────────────── # Request / response schemas # ────────────────────────────────────────────────────────────────────── class DealsSearchRequest(BaseModel): address: str = Field(..., min_length=2, max_length=200) years_back: int = Field(2, ge=1, le=10) radius_meters: int = Field(100, ge=10, le=2000) max_deals: int = Field(50, ge=1, le=200) deal_type: int = Field(2, ge=1, le=2) class AppraisalsSearchRequest(BaseModel): block: str | None = Field(None, max_length=20) plot: str | None = Field(None, max_length=20) decisive_appraiser: str | None = Field(None, max_length=100) committee: str | None = Field(None, max_length=100) decision_date_from: str | None = Field(None, max_length=20) decision_date_to: str | None = Field(None, max_length=20) publicity_date_from: str | None = Field(None, max_length=20) publicity_date_to: str | None = Field(None, max_length=20) search_text: str | None = Field(None, max_length=200) appraisal_header: str | None = Field(None, max_length=200) max_results: int = Field(30, ge=1, le=100) # When true and `block` is supplied, also fetch decisions from other # plots in the same block and from neighbouring blocks. Returned in # separate buckets in the response. include_nearby: bool = Field(False) nearby_block_radius: int = Field(2, ge=1, le=10) # ────────────────────────────────────────────────────────────────────── # App factory # ────────────────────────────────────────────────────────────────────── def create_app() -> FastAPI: """ Build the combined FastAPI + FastMCP application. The static-file mount at "/" must be the LAST mount so that more specific paths (/api, /mcp, /health) take precedence. """ # FastMCP needs its lifespan to run, so we forward it to FastAPI. if hasattr(mcp, "streamable_http_app"): mcp_app = mcp.streamable_http_app() elif hasattr(mcp, "http_app"): mcp_app = mcp.http_app() else: raise RuntimeError("FastMCP version does not expose an HTTP app") app = FastAPI( title="Nadlan-MCP Web", description=( "REST API for Israeli real-estate (Govmap) and decisive-appraiser " "(Ministry of Justice) data. Powers the React UI; also exposes the " "raw FastMCP endpoint at /mcp for LLM clients." ), version="2.1.0", docs_url="/api/docs", redoc_url=None, openapi_url="/api/openapi.json", lifespan=mcp_app.router.lifespan_context, ) # CORS — only relevant during local dev when Vite serves on :5173. app.add_middleware( CORSMiddleware, allow_origins=["http://localhost:5173", "http://127.0.0.1:5173"], allow_methods=["GET", "POST", "OPTIONS"], allow_headers=["*"], ) # ── Health ──────────────────────────────────────────────────────── @app.get("/api/health") @app.get("/health") async def health() -> dict[str, str]: return {"status": "ok", "service": "nadlan-mcp"} # ── /api/search/address ─────────────────────────────────────────── @app.get("/api/search/address") async def search_address(q: str = Query(..., min_length=2, max_length=120)): """Autocomplete an Israeli address. Also returns coordinates and (when available) the gush/חלקה for the top match.""" try: response = govmap_client.autocomplete_address(q) except Exception as e: logger.exception("autocomplete_address failed") raise HTTPException(status_code=502, detail=str(e)) results = [] for r in response.results: item: dict[str, Any] = { "text": r.text, "type": r.type, "score": r.score, "id": r.id, } if r.coordinates: item["coordinates"] = { "longitude": r.coordinates.longitude, "latitude": r.coordinates.latitude, } results.append(item) # Best-effort gush/חלקה for the top result. Use PARCEL_ALL layer # explicitly — the default layer 16 in get_gush_helka is `nadlan` # (deals) which has null gush/parcel for many addresses. gush_helka = None if response.results and response.results[0].coordinates: try: coords = response.results[0].coordinates gh = _query_parcel_layer(govmap_client, coords.longitude, coords.latitude) gush_helka = _extract_gush_helka(gh) except Exception as e: logger.warning(f"parcel lookup failed for '{q}': {e}") return {"query": q, "results": results, "gush_helka": gush_helka} # ── /api/search/deals ───────────────────────────────────────────── @app.post("/api/search/deals") async def search_deals(req: DealsSearchRequest): """Find recent real-estate deals around an address (Govmap).""" try: deals = govmap_client.find_recent_deals_for_address( req.address, req.years_back, req.radius_meters, req.max_deals, req.deal_type, ) except Exception as e: logger.exception("find_recent_deals_for_address failed") raise HTTPException(status_code=502, detail=str(e)) # Search-center coords + resolved address text for the UI (best-effort). search_coords = None resolved_address = None try: ar = govmap_client.autocomplete_address(req.address) if ar.results and ar.results[0].coordinates: c = ar.results[0].coordinates search_coords = {"longitude": c.longitude, "latitude": c.latitude} resolved_address = ar.results[0].text except Exception: pass # Strip bloat to keep payload small for the UI. bloat = { "shape", "sourceorder", "objectid", "priority", "source_polygon_id", "settlementId", "streetCode", "dealId", "polygonId", } deal_dicts = [] for idx, deal in enumerate(deals, start=1): d = deal.model_dump(mode="json", exclude_none=True) d = {k: v for k, v in d.items() if k not in bloat} d["id"] = idx d["deal_source"] = getattr(deal, "deal_source", None) deal_dicts.append(d) return { "search": { "address": req.address, "resolved_address": resolved_address, "coordinates": search_coords, "years_back": req.years_back, "radius_meters": req.radius_meters, "deal_type": req.deal_type, }, "total": len(deal_dicts), "deals": deal_dicts, } # ── /api/search/appraisals ──────────────────────────────────────── @app.post("/api/search/appraisals") async def search_appraisals(req: AppraisalsSearchRequest): """Search decisive-appraiser decisions (gov.il / Ministry of Justice). If `include_nearby=True` and a block is provided, also runs follow-up searches for the same block (different plots) and for adjacent blocks, returned in separate buckets. The buckets are mutually exclusive (deduped by upstream decision id). """ def _run(block: str | None, plot: str | None) -> Any: return decisive_appraiser_client.search_decisions_paged( max_results=req.max_results, block=block, plot=plot, decisive_appraiser=req.decisive_appraiser, committee=req.committee, decision_date_from=req.decision_date_from, decision_date_to=req.decision_date_to, publicity_date_from=req.publicity_date_from, publicity_date_to=req.publicity_date_to, search_text=req.search_text, appraisal_header=req.appraisal_header, ) try: exact_response = _run(req.block, req.plot) except Exception as e: logger.exception("search_decisions_paged (exact) failed") raise HTTPException(status_code=502, detail=str(e)) # Dedupe across buckets by a composite key (gov.il doesn't expose a # single stable id we can rely on). seen_keys: set[tuple] = set() def _serialize(decisions_iter: Any, start_idx: int = 1) -> list[dict]: out: list[dict] = [] running = start_idx for decision in decisions_iter: key = ( decision.appraisal_header, decision.decision_date, decision.decisive_appraiser, decision.block, decision.plot, ) if key in seen_keys: continue seen_keys.add(key) d = decision.model_dump(mode="json", exclude_none=True) documents = d.pop("documents", []) or [] d["id"] = running d["pdf_url"] = documents[0]["file_url"] if documents else None d["all_documents"] = documents out.append(d) running += 1 return out exact_decisions = _serialize(exact_response.results) same_block_decisions: list[dict] = [] nearby_block_decisions: list[dict] = [] if req.include_nearby and req.block: # Same block, different plots — only meaningful if a plot was given; # otherwise the exact search already covers the whole block. if req.plot: try: sb = _run(req.block, None) same_block_decisions = _serialize( sb.results, start_idx=len(exact_decisions) + 1 ) except Exception: logger.exception("same-block appraisals lookup failed") # Neighbouring blocks (numeric ±radius). Israeli cadastral blocks # are usually sequential within a city, so this is a reasonable # heuristic for "nearby parcels" without needing a spatial query. try: block_int = int(req.block) except ValueError: block_int = None if block_int is not None: base_idx = len(exact_decisions) + len(same_block_decisions) + 1 deltas = list(range(1, req.nearby_block_radius + 1)) ordered_neighbours = [] for delta in deltas: ordered_neighbours.extend([block_int - delta, block_int + delta]) for nb in ordered_neighbours: if nb <= 0: continue try: nbr = _run(str(nb), None) chunk = _serialize(nbr.results, start_idx=base_idx) nearby_block_decisions.extend(chunk) base_idx += len(chunk) except Exception: logger.exception(f"nearby block {nb} lookup failed") return { "total_in_db": exact_response.total_results, "returned": len(exact_decisions), "filters": req.model_dump(exclude_none=True), "decisions": exact_decisions, "same_block": { "count": len(same_block_decisions), "decisions": same_block_decisions, }, "nearby_blocks": { "count": len(nearby_block_decisions), "decisions": nearby_block_decisions, }, } # ── /api/appraisals/pdf ─────────────────────────────────────────── @app.get("/api/appraisals/pdf") async def proxy_appraisal_pdf(url: str = Query(..., min_length=20)): """ Stream a decisive-appraiser PDF through this server. Required because the upstream PDF host demands the same x-client-id header as the search API; a normal browser fetch would fail. """ from urllib.parse import urlparse parsed = urlparse(url) if parsed.scheme != "https" or parsed.hostname not in ALLOWED_PDF_HOSTS: raise HTTPException(status_code=400, detail="URL host not allowed") try: # Reuse the curl_cffi session that already carries x-client-id. response = decisive_appraiser_client._session.get( url, timeout=( decisive_appraiser_client.config.connect_timeout, decisive_appraiser_client.config.read_timeout * 4, ), headers={"Accept": "application/pdf,*/*"}, ) except Exception as e: logger.exception("PDF proxy upstream call failed") raise HTTPException(status_code=502, detail=str(e)) if response.status_code != 200: raise HTTPException( status_code=response.status_code, detail=f"Upstream returned {response.status_code}", ) if not response.content.startswith(b"%PDF"): raise HTTPException(status_code=502, detail="Upstream did not return a PDF") # Suggest a reasonable filename for the browser. filename = parsed.path.rstrip("/").split("/")[-1] or "appraisal" if not filename.lower().endswith(".pdf"): filename = f"{filename}.pdf" def _iter() -> Any: yield response.content return StreamingResponse( _iter(), media_type="application/pdf", headers={ "Content-Disposition": f'inline; filename="{filename}"', "Cache-Control": "private, max-age=3600", }, ) # ── /mcp/ → FastMCP ─────────────────────────────────────────────── app.mount("/mcp", mcp_app) # ── / → React static (must be LAST) ─────────────────────────────── web_dist = Path(__file__).resolve().parent.parent / "web" / "dist" if web_dist.is_dir(): app.mount( "/assets", StaticFiles(directory=web_dist / "assets"), name="assets", ) @app.get("/{full_path:path}", include_in_schema=False) async def spa_fallback(full_path: str): # Unknown paths fall back to index.html so React Router can take over. index = web_dist / "index.html" if index.is_file(): return FileResponse(index) return JSONResponse({"detail": "UI not built"}, status_code=404) else: logger.warning( f"Web build directory not found at {web_dist}; UI will not be served. " f"Run `cd web && pnpm build` to populate it." ) @app.get("/", include_in_schema=False) async def no_ui(): return JSONResponse( { "service": "nadlan-mcp", "ui": "not built", "api_docs": "/api/docs", "mcp": "/mcp", } ) return app # ────────────────────────────────────────────────────────────────────── # Helpers # ────────────────────────────────────────────────────────────────────── def _query_parcel_layer(client: Any, longitude: float, latitude: float) -> dict[str, Any]: """ Query Govmap's PARCEL_ALL layer at a point. Returns the raw JSON. The default `client.get_gush_helka` queries layer 16 (`nadlan`), which has null gush/parcel for many address points. PARCEL_ALL is the cadastral layer that reliably exposes them. """ url = f"{client.base_url}/layers-catalog/entitiesByPoint" payload = { "point": [longitude, latitude], "layers": [{"layerId": "PARCEL_ALL"}], "tolerance": 5, } timeout = (client.config.connect_timeout, client.config.read_timeout) response = client.session.post(url, json=payload, timeout=timeout) response.raise_for_status() return response.json() def _extract_gush_helka(govmap_response: Any) -> dict[str, Any] | None: """ Pull the (gush, helka) pair out of the layered Govmap response. Real shape (layerId 16 / nadlan layer): {"data": [{"name": "...", "entities": [ {"fields": [{"fieldName": "גוש", "fieldValue": "6212"}, {"fieldName": "חלקה", "fieldValue": "894"}, ...]} ]}]} Values can be null when the layer has no data at that point — we skip those and keep looking through every entity in every layer. """ if not isinstance(govmap_response, dict): return None block: str | None = None plot: str | None = None for layer in govmap_response.get("data") or []: if not isinstance(layer, dict): continue for entity in layer.get("entities") or []: if not isinstance(entity, dict): continue for f in entity.get("fields") or []: name = (f.get("fieldName") or "").strip() value = f.get("fieldValue") if value is None: continue value = str(value).strip() if not value: continue if ( name in ("גוש", "מספר גוש", "gush", "block", "GUSH_NUM", "gush_num") and not block ): block = value elif ( name in ("חלקה", "מספר חלקה", "helka", "parcel", "PARCEL", "PARCEL_NUM") and not plot ): plot = value if block and plot: return {"block": block, "plot": plot} if block or plot: return {"block": block, "plot": plot} return None