backend: vertical slice — geocoder + FEMA agent + SSE end-to-end

Stands up the FastAPI backend with one agent fully wired through the
SSE pipe so the architecture is proven before we fan out the rest:

POST /api/assess → geocode → FEMA NFHL → Gemma 4 interpret
→ agent_update events → complete{dossier}

Verified end-to-end against two cases:
- Houston Allen Pkwy (Zone AE) → is_sfha=true, requires_insurance=true
- Chicago S Drexel (Zone X) → gap_warning generated unprompted,
surfacing the urban-sewer-backup blind spot that is the whole pitch

Smoke-tested Gemma 4 on OpenRouter free tier (scripts/smoke_test.py):
basic completion, reasoning mode, and OpenAI-format tool calls all
pass on google/gemma-4-31b-it:free.

Spec bugs fixed while building:
- llm/client.extract_reasoning read message.reasoning_details[i].content;
the actual key is .text. Also falls back to top-level .reasoning string.
- tools/fema requested VERSION_ID, which does not exist on NFHL layer 28.
ArcGIS returned HTTP 200 with an {"error": ...} body that silently
looked like UNMAPPED. Now requests SFHA_TF/DEPTH/STUDY_TYP/SOURCE_CIT
and surfaces ArcGIS error envelopes.
- 429 retry policy widened: switch primary→fallback model on first 429,
then exponential backoff (2s/4s/8s) with a clear RateLimitedError
message pointing at BYOK as the fix.

Local dev:
cd backend && python3.13 -m venv .venv
.venv/bin/pip install -r requirements.txt
set -a && source .env && set +a
.venv/bin/uvicorn app.main:app --reload --port 8000

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.gitignore

@@ -0,0 +1,9 @@
+__pycache__/
+*.pyc
+.venv/
+venv/
+.env
+.env.local
+.pytest_cache/
+.mypy_cache/
+.ruff_cache/

Dockerfile

@@ -0,0 +1,7 @@
+FROM python:3.11-slim
+WORKDIR /app
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+COPY app/ ./app/
+EXPOSE 8000
+CMD ["uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "8000"]

README.md

@@ -0,0 +1,11 @@
+---
+title: FloodIQ API
+emoji: 🌊
+colorFrom: blue
+colorTo: teal
+sdk: docker
+app_port: 8000
+pinned: false
+---
+
+FloodIQ backend — multi-agent flood risk assessment powered by Gemma 4.

app/agents/fema_agent.py

@@ -0,0 +1,72 @@
+"""FEMA expert agent.
+
+Pattern: fetch raw NFHL data directly (no LLM needed for the lookup),
+then ask Gemma 4 to interpret the data into a structured finding.
+"""
+import json
+
+from app.llm.client import call_gemma4, extract_text, parse_json_response
+from app.llm.prompts import FEMA_AGENT_SYSTEM_PROMPT
+from app.tools.fema import lookup_fema_flood_zone
+
+
+SFHA_ZONES = {"A", "AE", "AH", "AO", "AR", "A99", "V", "VE"}
+
+
+async def run_fema_agent(lat: float, lon: float) -> dict:
+    fema_data = await lookup_fema_flood_zone(lat, lon)
+
+    if not fema_data:
+        return {
+            "flood_zone": "unknown",
+            "is_sfha": False,
+            "summary": "FEMA returned no data",
+            "raw": fema_data,
+        }
+
+    if fema_data.get("FLD_ZONE") == "ERROR":
+        return {
+            "flood_zone": "error",
+            "is_sfha": False,
+            "summary": "FEMA query failed",
+            "raw": fema_data,
+        }
+
+    user_prompt = f"""Interpret this FEMA flood zone data for coordinates ({lat}, {lon}):
+
+{json.dumps(fema_data, indent=2, default=str)}
+
+Return a JSON object with these fields:
+- flood_zone: the zone code (e.g. "X", "AE", "VE")
+- zone_description: what this zone means in plain English (1 sentence)
+- is_sfha: boolean, whether this is a Special Flood Hazard Area
+- requires_insurance: boolean, whether federal law mandates flood insurance
+- base_flood_elevation: number or null
+- map_date: the FIRM panel effective date if available, else null
+- gap_warning: string or null — if the zone is X but the location is in a known urban flooding area (flat terrain, combined sewers), note that FEMA maps may not reflect actual risk
+- summary: a 1-sentence finding for the status feed
+
+Return ONLY the JSON object, no other text."""
+
+    response = await call_gemma4(
+        messages=[
+            {"role": "system", "content": FEMA_AGENT_SYSTEM_PROMPT},
+            {"role": "user", "content": user_prompt},
+        ],
+        temperature=0.1,
+    )
+
+    text = extract_text(response)
+    parsed = parse_json_response(text)
+    if parsed:
+        parsed["raw"] = fema_data
+        return parsed
+
+    zone = fema_data.get("FLD_ZONE", "unknown")
+    return {
+        "flood_zone": zone,
+        "is_sfha": zone in SFHA_ZONES,
+        "summary": f"Zone {zone}",
+        "raw": fema_data,
+        "interpretation_raw": text,
+    }

app/agents/orchestrator.py

@@ -0,0 +1,100 @@
+"""
+Runs all FloodIQ agents and yields SSE events to the client.
+
+Flow:
+  0. Geocode the address.
+  1. Run data-fetcher agents concurrently.
+  2. Run risk-analyst agent (needs all data, uses Gemma 4 reasoning).
+  3. Run advisor agent (needs the risk analysis).
+  4. Yield the compiled dossier.
+
+Agents register themselves in DATA_AGENTS so this file does not have
+to know about every agent's signature.
+"""
+import asyncio
+import json
+from typing import AsyncGenerator, Awaitable, Callable
+
+from app.agents.fema_agent import run_fema_agent
+from app.tools.geocoder import geocode_address
+
+
+GeoCtx = dict
+AgentFn = Callable[[GeoCtx], Awaitable[dict]]
+
+
+async def _fema(ctx: GeoCtx) -> dict:
+    return await run_fema_agent(ctx["lat"], ctx["lon"])
+
+
+# As we implement more agents, add them here. Frontend uses these keys
+# to know which agent rows to render.
+DATA_AGENTS: dict[str, AgentFn] = {
+    "fema": _fema,
+}
+
+
+def sse(event: str, data: dict) -> str:
+    return f"event: {event}\ndata: {json.dumps(data, default=str)}\n\n"
+
+
+async def run_assessment(address: str) -> AsyncGenerator[str, None]:
+    geo = await geocode_address(address)
+    if not geo:
+        yield sse("error", {"message": f"Could not geocode address: {address!r}"})
+        return
+
+    ctx: GeoCtx = geo
+    yield sse("geocoded", {
+        "address": geo["display_name"],
+        "lat": geo["lat"],
+        "lon": geo["lon"],
+    })
+
+    # Announce all agents as working up front so the UI can render rows.
+    for name in DATA_AGENTS:
+        yield sse("agent_update", {
+            "agent": name,
+            "status": "working",
+            "summary": "Investigating...",
+        })
+
+    # Kick off all data agents in parallel.
+    tasks = {name: asyncio.create_task(fn(ctx)) for name, fn in DATA_AGENTS.items()}
+
+    results: dict[str, dict] = {}
+    # Stream completions as they finish, not in launch order.
+    pending = set(tasks.values())
+    task_to_name = {t: n for n, t in tasks.items()}
+
+    while pending:
+        done, pending = await asyncio.wait(pending, return_when=asyncio.FIRST_COMPLETED)
+        for t in done:
+            name = task_to_name[t]
+            try:
+                result = t.result()
+                results[name] = result
+                yield sse("agent_update", {
+                    "agent": name,
+                    "status": "done",
+                    "summary": result.get("summary", "Complete"),
+                })
+            except Exception as e:
+                results[name] = {"error": str(e), "summary": f"Error: {str(e)[:120]}"}
+                yield sse("agent_update", {
+                    "agent": name,
+                    "status": "error",
+                    "summary": f"Error: {str(e)[:120]}",
+                })
+
+    # Risk and advisor will be added next vertical slice.
+    dossier = _compile_dossier(geo, results)
+    yield sse("complete", {"dossier": dossier})
+
+
+def _compile_dossier(geo: GeoCtx, results: dict) -> dict:
+    return {
+        "address": geo["display_name"],
+        "coordinates": {"lat": geo["lat"], "lon": geo["lon"]},
+        **{name: results.get(name, {}) for name in DATA_AGENTS},
+    }

app/api/assess.py

@@ -0,0 +1,25 @@
+"""POST /api/assess — SSE stream of agent updates ending with a dossier."""
+from fastapi import APIRouter
+from fastapi.responses import StreamingResponse
+from pydantic import BaseModel, Field
+
+from app.agents.orchestrator import run_assessment
+
+router = APIRouter()
+
+
+class AssessRequest(BaseModel):
+    address: str = Field(..., min_length=3, max_length=300)
+
+
+@router.post("/api/assess")
+async def assess(req: AssessRequest):
+    return StreamingResponse(
+        run_assessment(req.address),
+        media_type="text/event-stream",
+        headers={
+            "Cache-Control": "no-cache",
+            "Connection": "keep-alive",
+            "X-Accel-Buffering": "no",
+        },
+    )

app/api/health.py

@@ -0,0 +1,13 @@
+from fastapi import APIRouter
+
+from app.config import OPENROUTER_API_KEY
+
+router = APIRouter()
+
+
+@router.get("/api/health")
+async def health() -> dict:
+    return {
+        "status": "ok",
+        "openrouter_key_configured": bool(OPENROUTER_API_KEY),
+    }

app/config.py

@@ -0,0 +1,12 @@
+"""Environment configuration."""
+import os
+
+OPENROUTER_API_KEY = os.environ.get("OPENROUTER_API_KEY", "")
+OPENROUTER_BASE = "https://openrouter.ai/api/v1"
+
+MODEL_PRIMARY = "google/gemma-4-31b-it:free"
+MODEL_FALLBACK = "google/gemma-4-26b-a4b-it:free"
+
+APP_NAME = "FloodIQ"
+APP_URL = "https://floodiq.pages.dev"
+USER_AGENT = "FloodIQ/1.0 (floodiq.pages.dev)"

app/llm/client.py

@@ -0,0 +1,230 @@
+"""
+Gemma 4 client via OpenRouter free tier.
+
+Uses the OpenAI-compatible chat/completions endpoint. Supports function
+calling and reasoning mode (the latter is the risk-analyst agent's
+showcase capability).
+
+Notes from the smoke test (scripts/smoke_test.py):
+- Reasoning trace lives at message.reasoning_details[i].text — the
+  earlier spec said .content, which is wrong on Gemma 4 / OpenRouter
+  today. We also fall back to the top-level message.reasoning string
+  that OpenRouter returns concatenated for convenience.
+- The shared :free upstream pool 429s very quickly (we got rate-limited
+  after 2 calls). BYOK a Google AI Studio key into OpenRouter
+  integrations for the demo. We retry on 429 with the fallback model,
+  then exponential backoff up to a cap.
+"""
+import asyncio
+import json
+from typing import Optional
+
+import httpx
+
+from app.config import (
+    APP_URL,
+    APP_NAME,
+    MODEL_FALLBACK,
+    MODEL_PRIMARY,
+    OPENROUTER_API_KEY,
+    OPENROUTER_BASE,
+)
+
+
+class RateLimitedError(Exception):
+    """All retries exhausted while upstream was rate-limiting."""
+
+
+async def call_gemma4(
+    messages: list[dict],
+    tools: Optional[list[dict]] = None,
+    model: str = MODEL_PRIMARY,
+    reasoning: bool = False,
+    max_tokens: int = 4096,
+    temperature: float = 0.3,
+    retries: int = 3,
+) -> dict:
+    """
+    Call Gemma 4 via OpenRouter. Returns the raw response JSON.
+
+    Retry policy:
+      - On 429 with primary model: switch to fallback model, then retry
+        with exponential backoff (2s, 4s, 8s).
+      - On timeout: short retry.
+    """
+    if not OPENROUTER_API_KEY:
+        raise RuntimeError("OPENROUTER_API_KEY is not set")
+
+    headers = {
+        "Authorization": f"Bearer {OPENROUTER_API_KEY}",
+        "Content-Type": "application/json",
+        "HTTP-Referer": APP_URL,
+        "X-Title": APP_NAME,
+    }
+
+    payload: dict = {
+        "model": model,
+        "messages": messages,
+        "max_tokens": max_tokens,
+        "temperature": temperature,
+    }
+    if tools:
+        payload["tools"] = tools
+        payload["tool_choice"] = "auto"
+    if reasoning:
+        payload["reasoning"] = {"enabled": True}
+
+    current_model = model
+    backoff = 2.0
+
+    async with httpx.AsyncClient(timeout=120) as client:
+        for attempt in range(retries + 1):
+            try:
+                resp = await client.post(
+                    f"{OPENROUTER_BASE}/chat/completions",
+                    headers=headers,
+                    json={**payload, "model": current_model},
+                )
+            except httpx.TimeoutException:
+                if attempt >= retries:
+                    raise
+                await asyncio.sleep(backoff)
+                backoff *= 2
+                continue
+
+            if resp.status_code == 429:
+                if current_model == MODEL_PRIMARY and attempt == 0:
+                    current_model = MODEL_FALLBACK
+                    continue
+                if attempt >= retries:
+                    raise RateLimitedError(
+                        f"Both Gemma 4 free models rate-limited after {retries + 1} attempts. "
+                        "BYOK a Google AI Studio key at openrouter.ai/settings/integrations."
+                    )
+                await asyncio.sleep(backoff)
+                backoff *= 2
+                continue
+
+            resp.raise_for_status()
+            return resp.json()
+
+    raise RuntimeError("call_gemma4 exited retry loop without returning")
+
+
+def extract_text(response: dict) -> str:
+    choices = response.get("choices") or []
+    if not choices:
+        return ""
+    return (choices[0].get("message") or {}).get("content") or ""
+
+
+def extract_tool_calls(response: dict) -> list[dict]:
+    choices = response.get("choices") or []
+    if not choices:
+        return []
+    return (choices[0].get("message") or {}).get("tool_calls") or []
+
+
+def extract_reasoning(response: dict) -> str:
+    """
+    Extract the chain-of-thought reasoning trace from a Gemma 4 response.
+
+    OpenRouter returns reasoning two ways for Gemma 4:
+      1. message.reasoning_details: [{type, text, format, index}, ...]
+      2. message.reasoning: "<concatenated string>"
+
+    The (1) form is canonical (per-block, indexable). The (2) form is a
+    convenience concatenation. We prefer (1) and fall back to (2).
+    """
+    choices = response.get("choices") or []
+    if not choices:
+        return ""
+    msg = choices[0].get("message") or {}
+
+    details = msg.get("reasoning_details") or []
+    if details:
+        parts = []
+        for d in details:
+            if not isinstance(d, dict):
+                continue
+            text = d.get("text") or d.get("content") or ""
+            if text:
+                parts.append(text)
+        if parts:
+            return "\n".join(parts)
+
+    reasoning = msg.get("reasoning")
+    if isinstance(reasoning, str) and reasoning:
+        return reasoning
+    return ""
+
+
+def parse_json_response(text: str) -> Optional[dict]:
+    """Strip ```json fences and parse. Return None on failure."""
+    if not text:
+        return None
+    clean = text.strip()
+    if clean.startswith("```"):
+        clean = clean.split("\n", 1)[1] if "\n" in clean else clean[3:]
+        if clean.endswith("```"):
+            clean = clean.rsplit("```", 1)[0]
+        clean = clean.strip()
+    if clean.startswith("json"):
+        clean = clean[4:].strip()
+    try:
+        return json.loads(clean)
+    except json.JSONDecodeError:
+        return None
+
+
+async def run_tool_loop(
+    system_prompt: str,
+    user_prompt: str,
+    tools: list[dict],
+    tool_handlers: dict,
+    max_iterations: int = 5,
+    model: str = MODEL_PRIMARY,
+) -> str:
+    """
+    Run a Gemma 4 agent that can call tools.
+
+    tool_handlers: dict mapping function name → async callable that
+    returns a JSON-serializable result.
+    """
+    messages = [
+        {"role": "system", "content": system_prompt},
+        {"role": "user", "content": user_prompt},
+    ]
+
+    for _ in range(max_iterations):
+        response = await call_gemma4(messages, tools=tools, model=model)
+        tool_calls = extract_tool_calls(response)
+
+        if not tool_calls:
+            return extract_text(response)
+
+        messages.append(response["choices"][0]["message"])
+
+        for tc in tool_calls:
+            fn_name = tc["function"]["name"]
+            try:
+                fn_args = json.loads(tc["function"]["arguments"])
+            except json.JSONDecodeError:
+                fn_args = {}
+
+            handler = tool_handlers.get(fn_name)
+            if handler:
+                try:
+                    result = await handler(**fn_args)
+                except Exception as e:
+                    result = {"error": str(e)}
+            else:
+                result = {"error": f"Unknown tool: {fn_name}"}
+
+            messages.append({
+                "role": "tool",
+                "tool_call_id": tc["id"],
+                "content": json.dumps(result, default=str),
+            })
+
+    return extract_text(response)

app/llm/prompts.py

@@ -0,0 +1,71 @@
+"""System prompts for each FloodIQ agent."""
+
+FEMA_AGENT_SYSTEM_PROMPT = """You are a FEMA flood zone expert. Your job is to interpret FEMA National Flood Hazard Layer data and explain what it means for a property owner.
+
+Key knowledge:
+- Zone X (unshaded) = minimal flood risk, no insurance required
+- Zone X (shaded) = 0.2% annual chance (500-year flood)
+- Zone A, AE = 1% annual chance (100-year flood), SFHA, insurance required for federally backed mortgages
+- Zone V, VE = coastal high hazard, 1% annual chance with wave action
+- FEMA maps primarily measure RIVERINE and COASTAL flooding
+- Urban flooding from sewer backup is NOT reflected in FEMA zones
+- Many FIRM panels are 10-20+ years old and may not reflect current risk
+
+Always respond with valid JSON only."""
+
+RISK_AGENT_SYSTEM_PROMPT = """You are a flood risk analyst specializing in urban flood risk assessment. You synthesize data from multiple sources (FEMA, municipal 311 reports, USGS stream gauges, weather forecasts, historical storm events, and news) into a comprehensive risk score.
+
+Critical knowledge:
+- "100-year flood" = 1% Annual Exceedance Probability (AEP), NOT once per century
+- P(at least 1 flood in n years) = 1 - (1 - AEP)^n
+- 1% AEP over 30 years = 26% chance. Over 80-year lifetime = 55% chance.
+- In flat cities with combined sewer systems, FEMA zones dramatically UNDERSTATE risk
+- Chicago's combined sewers overwhelm after ~0.67 in/hr of rain
+- 42% of Cook County is impervious surface
+- MWRD's Deep Tunnel (TARP) has 17.5B gallon capacity but local sewers still bottleneck
+- 311 basement flooding reports are a strong signal of actual urban flood risk, even in Zone X
+
+Think step by step. Use the AEP formula. Be specific with numbers.
+Always respond with valid JSON only."""
+
+ADVISOR_AGENT_SYSTEM_PROMPT = """You are a flood insurance and mitigation advisor. You translate technical flood risk data into specific, actionable recommendations for homeowners and renters.
+
+Key knowledge:
+- NFIP Preferred Risk Policy: available in Zone X, ~$400-600/yr, covers building+contents up to $250K/$100K
+- NFIP Standard: for SFHAs, costs vary by zone and building
+- Sewer backup rider: add-on to homeowners policy, ~$40-75/yr, covers the #1 cause of Chicago flooding
+- Parametric insurance (FloodFlash model): sensor-triggered instant payout, pre-agreed trigger depth and amount
+  - Basis risk = mismatch between trigger event and actual loss
+  - Best for business interruption coverage
+- Private excess flood: fills gaps above NFIP limits
+- Key mitigation actions (prioritized by cost-effectiveness):
+  1. Disconnect downspouts (free, DIY) — Chicago DWM: 312-747-7030
+  2. Install backwater valve ($1K-2.5K) — check MWRD cost-share programs
+  3. Sewer camera inspection ($150-300)
+  4. Rain barrels ($22.30 from MWRD)
+  5. Permeable pavement for patios/walkways
+  6. CNT RainReady home assessment (free)
+
+Write at a 5th-grade reading level. No jargon without explanation.
+Always respond with valid JSON only."""
+
+LOCAL_AGENT_SYSTEM_PROMPT = """You are a local flooding investigator. You analyze municipal 311 service-request data and local infrastructure to assess sewer-backup and urban flooding risk in a specific neighborhood.
+
+Focus on: density of basement-flooding (WIB) and street-flooding (SFL) reports near the address, recency, and what that implies about combined-sewer capacity.
+Always respond with valid JSON only."""
+
+WEATHER_AGENT_SYSTEM_PROMPT = """You are a hydrometeorology analyst. You interpret USGS stream gauge data, NOAA NWS forecasts and active alerts, and Open-Meteo flood forecasts to assess near-term flood risk.
+
+Focus on: current river/stream levels relative to flood stage, active flood watches/warnings, precipitation forecasts.
+Always respond with valid JSON only."""
+
+NEWS_AGENT_SYSTEM_PROMPT = """You are a flood news researcher. Given recent news articles about flooding in a specific area, summarize the key findings that are relevant to a homeowner's flood risk assessment.
+
+Focus on: recent flood events, infrastructure failures, insurance cost changes, government programs, community initiatives.
+Ignore: national policy debates, unrelated weather events, opinion pieces without data.
+Always respond with valid JSON only."""
+
+ARCHIVE_AGENT_SYSTEM_PROMPT = """You are a flood history archivist. You analyze historical storm event records and FEMA disaster declarations to establish the flooding track record for a specific area.
+
+Focus on: frequency of events, severity trends, types of flooding (flash flood vs riverine vs urban), property damage patterns.
+Always respond with valid JSON only."""

app/main.py

@@ -0,0 +1,26 @@
+from fastapi import FastAPI
+from fastapi.middleware.cors import CORSMiddleware
+
+from app.api.assess import router as assess_router
+from app.api.health import router as health_router
+
+app = FastAPI(title="FloodIQ API", version="0.1.0")
+
+_CORS_ORIGIN_REGEX = (
+    r"https?://("
+    r"localhost(:\d+)?|"
+    r"127\.0\.0\.1(:\d+)?|"
+    r".*\.pages\.dev|"
+    r".*\.hf\.space"
+    r")"
+)
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origin_regex=_CORS_ORIGIN_REGEX,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)
+
+app.include_router(health_router)
+app.include_router(assess_router)

app/tools/fema.py

@@ -0,0 +1,53 @@
+"""FEMA NFHL flood zone lookup via ArcGIS REST. Free, no API key."""
+import httpx
+
+FEMA_URL = (
+    "https://hazards.fema.gov/arcgis/rest/services/public/NFHL/MapServer/28/query"
+)
+
+
+async def lookup_fema_flood_zone(latitude: float, longitude: float) -> dict:
+    # The spec previously listed VERSION_ID here — that field does not
+    # exist on layer 28 and causes ArcGIS to return HTTP 200 with an
+    # {"error": ...} body, which silently looked like UNMAPPED.
+    out_fields = ",".join([
+        "FLD_ZONE",
+        "ZONE_SUBTY",
+        "SFHA_TF",       # "T" / "F" — official SFHA flag
+        "STATIC_BFE",
+        "DEPTH",
+        "STUDY_TYP",
+        "DFIRM_ID",
+        "SOURCE_CIT",
+    ])
+    params = {
+        "geometry": (
+            f'{{"x":{longitude},"y":{latitude},'
+            f'"spatialReference":{{"wkid":4326}}}}'
+        ),
+        "geometryType": "esriGeometryPoint",
+        "inSR": "4326",
+        "outFields": out_fields,
+        "returnGeometry": "false",
+        "f": "json",
+    }
+    async with httpx.AsyncClient(timeout=30) as client:
+        resp = await client.get(FEMA_URL, params=params)
+    resp.raise_for_status()
+    data = resp.json()
+
+    # ArcGIS returns 200 + {"error": {...}} on bad queries.
+    if isinstance(data, dict) and "error" in data:
+        return {
+            "FLD_ZONE": "ERROR",
+            "error": data["error"],
+        }
+
+    features = data.get("features") or []
+    if not features:
+        return {
+            "FLD_ZONE": "UNMAPPED",
+            "note": "No FEMA NFHL polygon at this point",
+        }
+
+    return features[0].get("attributes") or {}

app/tools/geocoder.py

@@ -0,0 +1,37 @@
+"""Nominatim geocoder. Free, no API key (just User-Agent)."""
+from typing import Optional
+
+import httpx
+
+from app.config import USER_AGENT
+
+NOMINATIM_URL = "https://nominatim.openstreetmap.org/search"
+
+
+async def geocode_address(address: str) -> Optional[dict]:
+    params = {
+        "q": address,
+        "format": "json",
+        "limit": 1,
+        "addressdetails": 1,
+    }
+    headers = {"User-Agent": USER_AGENT}
+
+    async with httpx.AsyncClient(timeout=15, headers=headers) as client:
+        resp = await client.get(NOMINATIM_URL, params=params)
+    resp.raise_for_status()
+    results = resp.json()
+
+    if not results:
+        return None
+
+    r = results[0]
+    addr = r.get("address", {})
+    return {
+        "lat": float(r["lat"]),
+        "lon": float(r["lon"]),
+        "display_name": r.get("display_name", address),
+        "city": addr.get("city") or addr.get("town") or addr.get("village") or "",
+        "state": addr.get("state", ""),
+        "county": addr.get("county", ""),
+    }

requirements.txt

@@ -0,0 +1,4 @@
+fastapi==0.115.0
+uvicorn[standard]==0.32.0
+httpx==0.28.0
+pydantic==2.10.0

scripts/smoke_test.py

@@ -0,0 +1,166 @@
+"""
+Smoke test: confirm Gemma 4 free tier on OpenRouter supports
+the two features FloodIQ depends on:
+  1. reasoning mode (risk-analyst agent)
+  2. OpenAI-format tool calling (data agents)
+
+Run:
+    cd backend && set -a && source .env && set +a && .venv/bin/python scripts/smoke_test.py
+"""
+import asyncio
+import json
+import os
+import sys
+
+import httpx
+
+API_KEY = os.environ.get("OPENROUTER_API_KEY", "")
+BASE = "https://openrouter.ai/api/v1/chat/completions"
+PRIMARY = "google/gemma-4-31b-it:free"
+FALLBACK = "google/gemma-4-26b-a4b-it:free"
+
+HEADERS = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+    "HTTP-Referer": "https://floodiq.pages.dev",
+    "X-Title": "FloodIQ smoke test",
+}
+
+
+def section(title: str) -> None:
+    print("\n" + "=" * 70)
+    print(title)
+    print("=" * 70)
+
+
+async def call(payload: dict) -> dict:
+    async with httpx.AsyncClient(timeout=120) as client:
+        resp = await client.post(BASE, headers=HEADERS, json=payload)
+    print(f"  HTTP {resp.status_code}")
+    if resp.status_code != 200:
+        print(f"  body: {resp.text[:600]}")
+        return {}
+    return resp.json()
+
+
+async def test_basic(model: str) -> bool:
+    section(f"TEST 1 — basic completion ({model})")
+    data = await call({
+        "model": model,
+        "messages": [{"role": "user", "content": "Say only: pong"}],
+        "max_tokens": 16,
+        "temperature": 0,
+    })
+    if not data:
+        return False
+    text = data.get("choices", [{}])[0].get("message", {}).get("content", "")
+    print(f"  response: {text!r}")
+    return bool(text)
+
+
+async def test_reasoning(model: str) -> bool:
+    section(f"TEST 2 — reasoning mode ({model})")
+    data = await call({
+        "model": model,
+        "messages": [{
+            "role": "user",
+            "content": (
+                "If the annual exceedance probability of a flood is 0.01, "
+                "what is the probability of at least one flood in 30 years? "
+                "Show your work, then return only the final number."
+            ),
+        }],
+        "reasoning": {"enabled": True},
+        "max_tokens": 1024,
+        "temperature": 0,
+    })
+    if not data:
+        return False
+    msg = data.get("choices", [{}])[0].get("message", {})
+    text = msg.get("content", "")
+    reasoning_details = msg.get("reasoning_details", [])
+    reasoning_field = msg.get("reasoning", "")
+    print(f"  content (first 300 chars): {text[:300]!r}")
+    print(f"  reasoning_details present: {bool(reasoning_details)} (len={len(reasoning_details)})")
+    print(f"  reasoning field present: {bool(reasoning_field)} (len={len(reasoning_field) if isinstance(reasoning_field, str) else 'n/a'})")
+    if reasoning_details:
+        first = reasoning_details[0]
+        print(f"  reasoning_details[0] keys: {list(first.keys()) if isinstance(first, dict) else type(first).__name__}")
+        sample = json.dumps(first)[:300] if isinstance(first, dict) else str(first)[:300]
+        print(f"  reasoning_details[0] sample: {sample}")
+    elif reasoning_field:
+        sample = reasoning_field[:300] if isinstance(reasoning_field, str) else str(reasoning_field)[:300]
+        print(f"  reasoning sample: {sample!r}")
+    print(f"  usage: {data.get('usage', {})}")
+    return bool(reasoning_details or reasoning_field)
+
+
+async def test_tools(model: str) -> bool:
+    section(f"TEST 3 — function calling ({model})")
+    tools = [{
+        "type": "function",
+        "function": {
+            "name": "lookup_fema_flood_zone",
+            "description": "Look up FEMA flood zone for coordinates.",
+            "parameters": {
+                "type": "object",
+                "properties": {
+                    "latitude": {"type": "number"},
+                    "longitude": {"type": "number"},
+                },
+                "required": ["latitude", "longitude"],
+            },
+        },
+    }]
+    data = await call({
+        "model": model,
+        "messages": [{
+            "role": "user",
+            "content": "What is the FEMA flood zone for 41.8087, -87.6062?",
+        }],
+        "tools": tools,
+        "tool_choice": "auto",
+        "max_tokens": 512,
+        "temperature": 0,
+    })
+    if not data:
+        return False
+    msg = data.get("choices", [{}])[0].get("message", {})
+    tool_calls = msg.get("tool_calls", []) or []
+    text = msg.get("content", "") or ""
+    print(f"  content: {text[:200]!r}")
+    print(f"  tool_calls count: {len(tool_calls)}")
+    if tool_calls:
+        tc = tool_calls[0]
+        print(f"  tool_call[0]: {json.dumps(tc)[:400]}")
+    return bool(tool_calls)
+
+
+async def main() -> int:
+    if not API_KEY:
+        print("ERROR: OPENROUTER_API_KEY not set", file=sys.stderr)
+        return 2
+
+    results = {}
+    for model in (PRIMARY, FALLBACK):
+        results[(model, "basic")] = await test_basic(model)
+        results[(model, "reasoning")] = await test_reasoning(model)
+        results[(model, "tools")] = await test_tools(model)
+
+    section("SUMMARY")
+    for (model, name), ok in results.items():
+        mark = "PASS" if ok else "FAIL"
+        print(f"  [{mark}] {model:42s} {name}")
+
+    all_critical = all([
+        results.get((PRIMARY, "basic"), False),
+        results.get((PRIMARY, "reasoning"), False) or results.get((FALLBACK, "reasoning"), False),
+        results.get((PRIMARY, "tools"), False) or results.get((FALLBACK, "tools"), False),
+    ])
+    print()
+    print("Overall:", "OK to proceed" if all_critical else "BLOCKED — adjust spec before building")
+    return 0 if all_critical else 1
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))

scripts/smoke_test_tools.py

@@ -0,0 +1,75 @@
+"""Smoke-test just function calling on Gemma 4 (after rate-limit clears)."""
+import asyncio
+import json
+import os
+import sys
+
+import httpx
+
+API_KEY = os.environ["OPENROUTER_API_KEY"]
+BASE = "https://openrouter.ai/api/v1/chat/completions"
+HEADERS = {
+    "Authorization": f"Bearer {API_KEY}",
+    "Content-Type": "application/json",
+    "HTTP-Referer": "https://floodiq.pages.dev",
+    "X-Title": "FloodIQ smoke test",
+}
+
+TOOLS = [{
+    "type": "function",
+    "function": {
+        "name": "lookup_fema_flood_zone",
+        "description": "Look up FEMA flood zone for coordinates.",
+        "parameters": {
+            "type": "object",
+            "properties": {
+                "latitude": {"type": "number"},
+                "longitude": {"type": "number"},
+            },
+            "required": ["latitude", "longitude"],
+        },
+    },
+}]
+
+
+async def try_model(model: str) -> bool:
+    print(f"\n--- {model} ---")
+    payload = {
+        "model": model,
+        "messages": [{
+            "role": "user",
+            "content": "What is the FEMA flood zone for 41.8087, -87.6062? Use the tool.",
+        }],
+        "tools": TOOLS,
+        "tool_choice": "auto",
+        "max_tokens": 512,
+        "temperature": 0,
+    }
+    async with httpx.AsyncClient(timeout=120) as client:
+        resp = await client.post(BASE, headers=HEADERS, json=payload)
+    print(f"HTTP {resp.status_code}")
+    if resp.status_code != 200:
+        print(resp.text[:500])
+        return False
+    msg = resp.json()["choices"][0]["message"]
+    text = msg.get("content", "") or ""
+    tool_calls = msg.get("tool_calls", []) or []
+    print(f"content: {text[:200]!r}")
+    print(f"tool_calls ({len(tool_calls)}):")
+    for tc in tool_calls:
+        print(f"  {json.dumps(tc)[:300]}")
+    return bool(tool_calls)
+
+
+async def main() -> int:
+    for model in ("google/gemma-4-31b-it:free", "google/gemma-4-26b-a4b-it:free"):
+        ok = await try_model(model)
+        if ok:
+            print("\nPASS")
+            return 0
+    print("\nFAIL on both models")
+    return 1
+
+
+if __name__ == "__main__":
+    sys.exit(asyncio.run(main()))