get_311_complaints
Retrieve 311 service request complaints filed at or near a NYC property address to assess neighborhood quality and building distress, covering noise, rodents, heat, and 200+ complaint types.
Instructions
Get 311 service request complaints filed at or near a property address.
Queries the local 311 database (NYC Open Data). Covers noise, rodents,
illegal dumping, graffiti, heat/hot water, illegal parking, street
conditions, and ~200 other complaint types.
311 data is a leading-indicator for neighborhood quality and building
distress — complaints are filed *before* violations are issued. High
complaint volume at an address is a red flag for active tenant issues.
Provide either `address` OR `bbl` (not both).
Args:
address: Street address, e.g. "37-06 80th Street, Queens".
bbl: 10-digit NYC BBL. Resolved to street address via PAD table.
complaint_type: Filter by complaint type keyword, e.g. "NOISE",
"RODENT", "HEAT", "ILLEGAL PARKING". Case-insensitive.
since_year: Return only complaints from this year onward (2010–present).
status: Filter by status: "Open" or "Closed".
limit: Max complaints to return (1–100, default 30).Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| address | No | ||
| bbl | No | ||
| complaint_type | No | ||
| since_year | No | ||
| status | No | ||
| limit | No |
Output Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
No arguments | |||
Implementation Reference
- The main tool handler that queries NYC 311 complaints by address or BBL, with support for local DB (fast path via BBL index or LIKE scan) and Socrata API fallback.
@mcp.tool() async def get_311_complaints( address: str | None = None, bbl: str | None = None, complaint_type: str | None = None, since_year: int | None = None, status: str | None = None, limit: int = 30, ) -> dict[str, Any]: """Get 311 service request complaints filed at or near a property address. Queries the local 311 database (NYC Open Data). Covers noise, rodents, illegal dumping, graffiti, heat/hot water, illegal parking, street conditions, and ~200 other complaint types. 311 data is a leading-indicator for neighborhood quality and building distress — complaints are filed *before* violations are issued. High complaint volume at an address is a red flag for active tenant issues. Provide either `address` OR `bbl` (not both). Args: address: Street address, e.g. "37-06 80th Street, Queens". bbl: 10-digit NYC BBL. Resolved to street address via PAD table. complaint_type: Filter by complaint type keyword, e.g. "NOISE", "RODENT", "HEAT", "ILLEGAL PARKING". Case-insensitive. since_year: Return only complaints from this year onward (2010–present). status: Filter by status: "Open" or "Closed". limit: Max complaints to return (1–100, default 30). """ if not address and not bbl: raise ToolError("Provide either address or bbl.") if address and bbl: raise ToolError("Provide either address or bbl, not both.") if limit < 1 or limit > 100: raise ToolError("limit must be between 1 and 100.") if since_year is not None and (since_year < 2010 or since_year > 2030): raise ToolError("since_year must be between 2010 and 2030.") if status is not None and status.upper() not in ("OPEN", "CLOSED"): raise ToolError("status must be 'Open' or 'Closed'.") if complaint_type is not None and len(complaint_type) > 100: raise ToolError("complaint_type must be 100 characters or fewer.") house_number = "" street_name = "" resolved_address: str | None = None data_source = "NYC 311 Service Requests — local DB (NYC Open Data erm2-nwe9)" data_note = "Local bulk dataset. Address matching is approximate." # ── BBL path: direct lookup ─────────────────────────────────────── if bbl: from nyc_property_intel.utils import validate_bbl try: validate_bbl(bbl) except ValueError as exc: raise ToolError(str(exc)) from exc # Try local DB by BBL first (fastest, most accurate) try: complaints = await _query_local_by_bbl( bbl, complaint_type, since_year, status, limit ) # Resolve a human-readable address from PAD so address_queried # is meaningful rather than showing the raw BBL number. pad_row = await fetch_one( "SELECT lhnd AS house_number, stname AS street_name " "FROM pad_adr WHERE bbl = $1 LIMIT 1", bbl, ) bbl_address_label = ( f"{pad_row['house_number']} {pad_row['street_name']}" if pad_row else bbl ) return { "address_queried": bbl_address_label, "bbl": bbl, "total_returned": len(complaints), "summary": _summarize(complaints), "complaints": [dict(c) for c in complaints], "data_source": data_source, "data_note": data_note, } except asyncpg.UndefinedTableError: logger.info("nyc_311_complaints not found — falling back to Socrata") # BBL had no 311 hits or table missing: resolve address for Socrata fallback row = await fetch_one( "SELECT lhnd AS house_number, stname AS street_name " "FROM pad_adr WHERE bbl = $1 LIMIT 1", bbl, ) if row is None: raise ToolError( f"Could not find an address for BBL {bbl}. " "Try passing the street address directly." ) house_number = row["house_number"] street_name = row["street_name"] resolved_address = f"{house_number} {street_name}" # ── Address path ────────────────────────────────────────────────── else: from nyc_property_intel.geoclient import parse_address try: parsed = parse_address(address) # type: ignore[arg-type] house_number = parsed["house_number"] street_name = parsed["street"] resolved_address = f"{house_number} {street_name}" except ToolError: resolved_address = address street_name = address or "" # Attempt to resolve the address to a BBL so we can use the fast # indexed BBL path instead of the slow full-table LIKE scan. try: resolved_bbl_row = await fetch_one( "SELECT bbl FROM pad_adr " "WHERE upper(stname) = upper($1) AND lhnd = $2 LIMIT 1", street_name, house_number or "", ) if resolved_bbl_row and resolved_bbl_row.get("bbl"): bbl = resolved_bbl_row["bbl"] logger.debug( "get_311_complaints: resolved address %r to BBL %s, using fast path", resolved_address, bbl, ) complaints = await _query_local_by_bbl( bbl, complaint_type, since_year, status, limit ) return { "address_queried": resolved_address, "bbl": bbl, "total_returned": len(complaints), "summary": _summarize(complaints), "complaints": [dict(c) for c in complaints], "data_source": data_source, "data_note": data_note, } except asyncpg.UndefinedTableError: pass except Exception as exc: logger.debug("BBL resolution for address %r failed: %s", resolved_address, exc) # ── Local address LIKE scan (fallback for unresolvable addresses) ── # Warning: this is a full-table LIKE scan on a large dataset. # It may time out if pg_trgm indexes are not installed. try: complaints = await _query_local_by_address( street_name, house_number or None, complaint_type, since_year, status, limit, ) return { "address_queried": resolved_address, "bbl": bbl, "total_returned": len(complaints), "summary": _summarize(complaints), "complaints": [dict(c) for c in complaints], "data_source": data_source, "data_note": data_note, } except asyncpg.UndefinedTableError: logger.info("nyc_311_complaints not found — falling back to Socrata") except ToolError: # Re-raise ToolErrors (e.g. timeout) with a more actionable message. raise ToolError( f"311 complaint search for {resolved_address!r} timed out or failed. " "Try using a BBL instead of an address for faster results." ) # ── Socrata fallback ────────────────────────────────────────────── try: complaints_raw = await _query_socrata_fallback( street_name, house_number, complaint_type, since_year, status, limit ) except SocrataError as exc: raise ToolError(str(exc)) from exc return { "address_queried": resolved_address, "bbl": bbl, "total_returned": len(complaints_raw), "summary": _summarize(complaints_raw), "complaints": complaints_raw, "data_source": "NYC 311 Service Requests via Socrata API (erm2-nwe9)", "data_note": "Real-time via Socrata API (local table unavailable).", } - src/nyc_property_intel/tools/complaints_311.py:159-345 (registration)The tool is registered via the @mcp.tool() decorator applied to the async function get_311_complaints. Import of this module in server.py triggers the decoration.
@mcp.tool() async def get_311_complaints( address: str | None = None, bbl: str | None = None, complaint_type: str | None = None, since_year: int | None = None, status: str | None = None, limit: int = 30, ) -> dict[str, Any]: """Get 311 service request complaints filed at or near a property address. Queries the local 311 database (NYC Open Data). Covers noise, rodents, illegal dumping, graffiti, heat/hot water, illegal parking, street conditions, and ~200 other complaint types. 311 data is a leading-indicator for neighborhood quality and building distress — complaints are filed *before* violations are issued. High complaint volume at an address is a red flag for active tenant issues. Provide either `address` OR `bbl` (not both). Args: address: Street address, e.g. "37-06 80th Street, Queens". bbl: 10-digit NYC BBL. Resolved to street address via PAD table. complaint_type: Filter by complaint type keyword, e.g. "NOISE", "RODENT", "HEAT", "ILLEGAL PARKING". Case-insensitive. since_year: Return only complaints from this year onward (2010–present). status: Filter by status: "Open" or "Closed". limit: Max complaints to return (1–100, default 30). """ if not address and not bbl: raise ToolError("Provide either address or bbl.") if address and bbl: raise ToolError("Provide either address or bbl, not both.") if limit < 1 or limit > 100: raise ToolError("limit must be between 1 and 100.") if since_year is not None and (since_year < 2010 or since_year > 2030): raise ToolError("since_year must be between 2010 and 2030.") if status is not None and status.upper() not in ("OPEN", "CLOSED"): raise ToolError("status must be 'Open' or 'Closed'.") if complaint_type is not None and len(complaint_type) > 100: raise ToolError("complaint_type must be 100 characters or fewer.") house_number = "" street_name = "" resolved_address: str | None = None data_source = "NYC 311 Service Requests — local DB (NYC Open Data erm2-nwe9)" data_note = "Local bulk dataset. Address matching is approximate." # ── BBL path: direct lookup ─────────────────────────────────────── if bbl: from nyc_property_intel.utils import validate_bbl try: validate_bbl(bbl) except ValueError as exc: raise ToolError(str(exc)) from exc # Try local DB by BBL first (fastest, most accurate) try: complaints = await _query_local_by_bbl( bbl, complaint_type, since_year, status, limit ) # Resolve a human-readable address from PAD so address_queried # is meaningful rather than showing the raw BBL number. pad_row = await fetch_one( "SELECT lhnd AS house_number, stname AS street_name " "FROM pad_adr WHERE bbl = $1 LIMIT 1", bbl, ) bbl_address_label = ( f"{pad_row['house_number']} {pad_row['street_name']}" if pad_row else bbl ) return { "address_queried": bbl_address_label, "bbl": bbl, "total_returned": len(complaints), "summary": _summarize(complaints), "complaints": [dict(c) for c in complaints], "data_source": data_source, "data_note": data_note, } except asyncpg.UndefinedTableError: logger.info("nyc_311_complaints not found — falling back to Socrata") # BBL had no 311 hits or table missing: resolve address for Socrata fallback row = await fetch_one( "SELECT lhnd AS house_number, stname AS street_name " "FROM pad_adr WHERE bbl = $1 LIMIT 1", bbl, ) if row is None: raise ToolError( f"Could not find an address for BBL {bbl}. " "Try passing the street address directly." ) house_number = row["house_number"] street_name = row["street_name"] resolved_address = f"{house_number} {street_name}" # ── Address path ────────────────────────────────────────────────── else: from nyc_property_intel.geoclient import parse_address try: parsed = parse_address(address) # type: ignore[arg-type] house_number = parsed["house_number"] street_name = parsed["street"] resolved_address = f"{house_number} {street_name}" except ToolError: resolved_address = address street_name = address or "" # Attempt to resolve the address to a BBL so we can use the fast # indexed BBL path instead of the slow full-table LIKE scan. try: resolved_bbl_row = await fetch_one( "SELECT bbl FROM pad_adr " "WHERE upper(stname) = upper($1) AND lhnd = $2 LIMIT 1", street_name, house_number or "", ) if resolved_bbl_row and resolved_bbl_row.get("bbl"): bbl = resolved_bbl_row["bbl"] logger.debug( "get_311_complaints: resolved address %r to BBL %s, using fast path", resolved_address, bbl, ) complaints = await _query_local_by_bbl( bbl, complaint_type, since_year, status, limit ) return { "address_queried": resolved_address, "bbl": bbl, "total_returned": len(complaints), "summary": _summarize(complaints), "complaints": [dict(c) for c in complaints], "data_source": data_source, "data_note": data_note, } except asyncpg.UndefinedTableError: pass except Exception as exc: logger.debug("BBL resolution for address %r failed: %s", resolved_address, exc) # ── Local address LIKE scan (fallback for unresolvable addresses) ── # Warning: this is a full-table LIKE scan on a large dataset. # It may time out if pg_trgm indexes are not installed. try: complaints = await _query_local_by_address( street_name, house_number or None, complaint_type, since_year, status, limit, ) return { "address_queried": resolved_address, "bbl": bbl, "total_returned": len(complaints), "summary": _summarize(complaints), "complaints": [dict(c) for c in complaints], "data_source": data_source, "data_note": data_note, } except asyncpg.UndefinedTableError: logger.info("nyc_311_complaints not found — falling back to Socrata") except ToolError: # Re-raise ToolErrors (e.g. timeout) with a more actionable message. raise ToolError( f"311 complaint search for {resolved_address!r} timed out or failed. " "Try using a BBL instead of an address for faster results." ) # ── Socrata fallback ────────────────────────────────────────────── try: complaints_raw = await _query_socrata_fallback( street_name, house_number, complaint_type, since_year, status, limit ) except SocrataError as exc: raise ToolError(str(exc)) from exc return { "address_queried": resolved_address, "bbl": bbl, "total_returned": len(complaints_raw), "summary": _summarize(complaints_raw), "complaints": complaints_raw, "data_source": "NYC 311 Service Requests via Socrata API (erm2-nwe9)", "data_note": "Real-time via Socrata API (local table unavailable).", } - src/nyc_property_intel/server.py:325-329 (registration)The complaints_311 module is imported here, which triggers the @mcp.tool() decorator and registers get_311_complaints with the MCP server.
from nyc_property_intel.tools import ( # noqa: E402 analysis, # noqa: F401 comps, # noqa: F401 complaints_311, # noqa: F401 dob_complaints, # noqa: F401 - Helper that converts a year integer to an ISO date prefix for SQL filtering.
def _since_prefix(since_year: int | None) -> str | None: """Convert since_year to an ISO date prefix for text comparison.""" return f"{since_year}-01-01" if since_year else None - Helper that summarizes a list of complaints into open/closed counts and top 5 complaint types.
def _summarize(complaints: list[dict[str, Any]]) -> dict[str, Any]: open_count = sum(1 for c in complaints if (c.get("status") or "").upper() == "OPEN") type_counts: dict[str, int] = {} for c in complaints: ct = c.get("complaint_type") or "Unknown" type_counts[ct] = type_counts.get(ct, 0) + 1 top_types = sorted(type_counts.items(), key=lambda x: x[1], reverse=True)[:5] return { "open": open_count, "closed": len(complaints) - open_count, "top_complaint_types": [{"type": t, "count": c} for t, c in top_types], }