get_property_issues
Get HPD housing, DOB building code, and ECB/OATH violations for a property. Assess regulatory risk with summary counts and detailed violation history.
Instructions
Get HPD housing violations, DOB building code violations, and ECB/OATH violations for a property.
HPD Class C violations are immediately hazardous. ECB violations include
penalties and balances due. Returns both summary counts and violation
details. Use this to assess a building's regulatory risk profile.
Note on historical depth: our local DB retains all historical HPD
violations and complaints, while NYC's live Socrata API rolls older
records out of its public feed. As a result, the totals reported here
may exceed what data.cityofnewyork.us shows for the same BBL — the
extra rows are real, just no longer surfaced by NYC Open Data.Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| bbl | Yes | ||
| source | No | ALL | |
| status | No | ||
| severity | No | ||
| since_date | No | ||
| limit | No | ||
| include_summary | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The get_property_issues async function is the actual tool handler, decorated with @mcp.tool(). It fetches HPD housing violations, DOB building code violations, and ECB/OATH violations for a property by BBL, with optional filtering by source, status, severity, and date range. Returns summary counts (from materialized view) plus violation details, with a data freshness note.
@mcp.tool() async def get_property_issues( bbl: str, source: str = "ALL", status: str | None = None, severity: str | None = None, since_date: str | None = None, limit: int = 25, include_summary: bool = True, ) -> dict[str, Any]: """Get HPD housing violations, DOB building code violations, and ECB/OATH violations for a property. HPD Class C violations are immediately hazardous. ECB violations include penalties and balances due. Returns both summary counts and violation details. Use this to assess a building's regulatory risk profile. Note on historical depth: our local DB retains all historical HPD violations and complaints, while NYC's live Socrata API rolls older records out of its public feed. As a result, the totals reported here may exceed what data.cityofnewyork.us shows for the same BBL — the extra rows are real, just no longer surfaced by NYC Open Data. """ # ── Validate inputs ────────────────────────────────────────────── try: validate_bbl(bbl) except ValueError as exc: raise ToolError(str(exc)) from exc if limit < 1 or limit > 200: raise ToolError("limit must be between 1 and 200.") source_upper = source.upper() if source_upper not in _VALID_SOURCES: raise ToolError( f"Invalid source: {source!r}. Must be one of: HPD, DOB, ECB, ALL." ) # Normalize optional filters — DB stores uppercase values ('OPEN', 'CLOSE', 'A', 'B', 'C'). # Accepting mixed-case from callers prevents silent empty-result bugs. normalized_status = normalize_filter(status) normalized_severity = normalize_filter(severity) # Parse since_date if provided. since: datetime.date | None = None if since_date is not None: try: since = datetime.date.fromisoformat(since_date) except ValueError as exc: raise ToolError( f"Invalid date format: {since_date!r}. " "Please use ISO 8601 format: YYYY-MM-DD." ) from exc # ── Summary from materialized view ─────────────────────────────── # Always return a fully-populated summary dict so callers can rely on # numeric fields without null-checks. Clean buildings (no row in the # materialized view) get zeroed counts. summary: dict[str, Any] | None = None if include_summary: try: summary = await fetch_one(_SQL_SUMMARY, bbl) except asyncpg.UndefinedTableError: logger.info("mv_violation_summary not available, skipping summary") summary = None if summary is None: summary = { "bbl": bbl, "hpd_total": 0, "hpd_class_a": 0, "hpd_class_b": 0, "hpd_class_c": 0, "hpd_open": 0, "hpd_most_recent": None, "dob_total": 0, "dob_no_disposition": 0, "dob_has_disposition": 0, "dob_most_recent": None, } # Augment with ECB stats — not in the materialized view. if source_upper in ("ECB", "ALL"): try: ecb_stats = await fetch_one(_SQL_ECB_SUMMARY, bbl) bal = (ecb_stats or {}).get("ecb_balance_due_total") summary["ecb_total"] = int((ecb_stats or {}).get("ecb_total") or 0) summary["ecb_active"] = int((ecb_stats or {}).get("ecb_active") or 0) summary["ecb_balance_due_total"] = float(bal) if bal is not None else 0.0 summary["ecb_most_recent"] = (ecb_stats or {}).get("ecb_most_recent") except asyncpg.UndefinedTableError: logger.info("ecb_violations table not loaded, skipping ECB summary") # ── HPD violations ─────────────────────────────────────────────── hpd_violations: list[dict[str, Any]] = [] if source_upper in ("HPD", "ALL"): try: hpd_violations = await fetch_all( _SQL_HPD, bbl, normalized_severity, # $2 — class filter (A/B/C), uppercased normalized_status, # $3 — currentstatus filter, uppercased since, # $4 — date filter limit, # $5 — row limit ) except asyncpg.UndefinedTableError: logger.info("hpd_violations table not loaded, skipping HPD section") # ── DOB violations ─────────────────────────────────────────────── dob_violations: list[dict[str, Any]] = [] if source_upper in ("DOB", "ALL"): try: dob_violations = await fetch_all( _SQL_DOB, bbl, since, # $2 — date filter limit, # $3 — row limit ) except asyncpg.UndefinedTableError: logger.info("dob_violations table not loaded, skipping DOB section") # ── ECB violations ────────────────────────────────────────────── ecb_violations: list[dict[str, Any]] = [] if source_upper in ("ECB", "ALL"): try: ecb_violations = await fetch_all( _SQL_ECB, bbl, since, # $2 — date filter limit, # $3 — row limit ) except asyncpg.UndefinedTableError: logger.info("ecb_violations table not loaded, skipping ECB section") # ── Cross-validation guard ─────────────────────────────────────── # Warn if summary says open violations exist but filter returned nothing — # catches future regressions where filter normalization silently drops data. if ( summary and normalized_status == "OPEN" and source_upper in ("HPD", "ALL") and (summary.get("hpd_open") or 0) > 0 and len(hpd_violations) == 0 ): logger.warning( "get_property_issues: summary reports %d open HPD violations for BBL %s " "but query returned 0 rows (status=%r, severity=%r, since=%r) — " "possible filter mismatch or stale materialized view", summary["hpd_open"], bbl, normalized_status, normalized_severity, since, ) # ── Build response ─────────────────────────────────────────────── total_returned = len(hpd_violations) + len(dob_violations) + len(ecb_violations) freshness_parts: list[str] = [] if source_upper in ("HPD", "ALL"): freshness_parts.append(data_freshness_note("hpd_violations")) if source_upper in ("DOB", "ALL"): freshness_parts.append(data_freshness_note("dob_violations")) if source_upper in ("ECB", "ALL"): freshness_parts.append(data_freshness_note("ecb_violations")) return { "bbl": bbl, "summary": summary, "hpd_violations": hpd_violations, "dob_violations": dob_violations, "ecb_violations": ecb_violations, "total_returned": total_returned, "data_as_of": " | ".join(freshness_parts), } - Function signature defines the input schema: bbl (str, required), source (str, default 'ALL'), status (str|None), severity (str|None), since_date (str|None), limit (int, default 25), include_summary (bool, default True). Output is dict[str, Any] with keys: bbl, summary, hpd_violations, dob_violations, ecb_violations, total_returned, data_as_of.
async def get_property_issues( bbl: str, source: str = "ALL", status: str | None = None, severity: str | None = None, since_date: str | None = None, limit: int = 25, include_summary: bool = True, ) -> dict[str, Any]: - src/nyc_property_intel/tools/issues.py:75-75 (registration)Registration via @mcp.tool() decorator on line 75. The mcp instance is imported from nyc_property_intel.app (line 17). The tool module is imported in server.py line 337, which triggers registration at module load.
@mcp.tool() - data_freshness_note helper used in get_property_issues to annotate returned data with source/cadence info for hpd_violations, dob_violations, ecb_violations tables.
def data_freshness_note(table_name: str) -> str: """Return a human-readable note about the freshness of a data table. Args: table_name: Internal table/dataset name (e.g., "pluto", "rpad"). Returns: A string like "Source: NYC DCP PLUTO, updated quarterly." """ info = _DATA_SOURCES.get(table_name.lower()) if info is None: return f"Source: NYC Open Data ({table_name})." return f"Source: {info['source']}, {info['cadence']}." - normalize_filter helper used by get_property_issues to uppercase user-supplied status/severity filters so they match DB values.
def normalize_filter(value: str | None) -> str | None: """Normalize an optional filter string to uppercase, treating empty string as None. Used by tools that accept status/class/type filter params so that 'Open', 'open', and 'OPEN' all match the uppercase DB values. Args: value: Raw user-supplied filter string, or None. Returns: Uppercased string, or None if value is None or empty. """ if not value or not value.strip(): return None return value.strip().upper()