EPA Envirofacts MCP Server

Instructions

Implementation Reference

• src/tools/search_facilities.py:131-163 (handler)

  The MCP tool handler function `search_facilities_tool` that wraps the core search logic.
  async def search_facilities_tool( facility_name: Optional[str] = None, naics_code: Optional[str] = None, state: Optional[str] = None, zip_code: Optional[str] = None, city: Optional[str] = None, limit: int = 100 ) -> List[FacilityInfo]: """Search for EPA-regulated facilities using various filters. Search for facilities in the EPA's Facility Registry Service using facility name, NAICS code, state, ZIP code, or city. At least one search parameter must be provided. Args: facility_name: Partial or full facility name (uses contains matching) naics_code: NAICS industry code state: Two-letter state code (e.g., 'NY', 'CA') zip_code: 5-digit ZIP code city: City name limit: Maximum results to return (default: 100) Returns: List of facilities with registry ID, name, address, coordinates, active programs, industry codes, and status information """ return await search_facilities( facility_name=facility_name, naics_code=naics_code, state=state, zip_code=zip_code, city=city, limit=limit )
• src/models/facility.py:23-40 (schema)

  Pydantic `BaseModel` defining the structure of `FacilityInfo` returned by the tool.
  class FacilityInfo(BaseModel): """Information about an EPA-regulated facility.""" registry_id: str = Field(..., description="FRS Registry ID") name: str = Field(..., description="Facility name") address: Optional[str] = Field(None, description="Facility address") city: Optional[str] = Field(None, description="City") state: Optional[str] = Field(None, description="State abbreviation") zip_code: Optional[str] = Field(None, description="ZIP code") coordinates: Optional[Coordinates] = Field(None, description="Facility coordinates") programs: List[FacilityType] = Field(default_factory=list, description="Active EPA programs") naics_code: Optional[str] = Field(None, description="NAICS industry code") naics_description: Optional[str] = Field(None, description="NAICS industry description") distance_miles: Optional[float] = Field(None, ge=0, description="Distance from search center in miles") status: Optional[str] = Field(None, description="Facility status") def __str__(self) -> str: return f"{self.name} ({self.registry_id})"
• src/tools/search_facilities.py:124-164 (registration)

  The `register_tool` function that defines and registers the `search_facilities_tool` handler with the FastMCP instance.
  def register_tool(mcp: FastMCP): """Register the search facilities tool with FastMCP. Args: mcp: FastMCP instance """ @mcp.tool() async def search_facilities_tool( facility_name: Optional[str] = None, naics_code: Optional[str] = None, state: Optional[str] = None, zip_code: Optional[str] = None, city: Optional[str] = None, limit: int = 100 ) -> List[FacilityInfo]: """Search for EPA-regulated facilities using various filters. Search for facilities in the EPA's Facility Registry Service using facility name, NAICS code, state, ZIP code, or city. At least one search parameter must be provided. Args: facility_name: Partial or full facility name (uses contains matching) naics_code: NAICS industry code state: Two-letter state code (e.g., 'NY', 'CA') zip_code: 5-digit ZIP code city: City name limit: Maximum results to return (default: 100) Returns: List of facilities with registry ID, name, address, coordinates, active programs, industry codes, and status information """ return await search_facilities( facility_name=facility_name, naics_code=naics_code, state=state, zip_code=zip_code, city=city, limit=limit )
• server.py:95-98 (registration)

  Invocation of tool registration functions, including `register_search_tool(mcp)` which registers the search_facilities_tool.
  register_tool(mcp) register_search_tool(mcp) register_compliance_tool(mcp) register_chemical_tool(mcp)
• src/tools/search_facilities.py:15-123 (helper)

  Core helper function `search_facilities` that performs input validation and queries the FRSClient API.
  async def search_facilities( facility_name: Optional[str] = None, naics_code: Optional[str] = None, state: Optional[str] = None, zip_code: Optional[str] = None, city: Optional[str] = None, limit: int = 100 ) -> List[FacilityInfo]: """Search for EPA-regulated facilities using various filters. This tool allows searching for facilities in the EPA's Facility Registry Service (FRS) using facility name, NAICS code, state, ZIP code, or city. At least one search parameter must be provided. Args: facility_name: Partial or full facility name (uses contains matching) naics_code: NAICS industry code state: Two-letter state code (e.g., 'NY', 'CA') zip_code: 5-digit ZIP code city: City name limit: Maximum results to return (default: 100, max: 1000) Returns: List of FacilityInfo objects containing: - Registry ID and facility name - Address and location information - Active EPA programs (TRI, RCRA, etc.) - Industry codes and descriptions - Facility status Raises: ValueError: If no search parameters provided or invalid parameter format Exception: If EPA API query fails Example: >>> facilities = await search_facilities(facility_name="Chemical", state="NY") >>> print(f"Found {len(facilities)} facilities") >>> for facility in facilities[:3]: ... print(f"{facility.name} - {facility.city}, {facility.state}") """ # Validate input parameters if not any([facility_name, naics_code, state, zip_code, city]): raise ValueError("At least one search parameter must be provided") if not (1 <= limit <= 1000): raise ValueError("Limit must be between 1 and 1000") # Clean and validate individual parameters if facility_name: facility_name = facility_name.strip() if not facility_name: raise ValueError("Facility name cannot be empty") if naics_code: naics_code = naics_code.strip() if not naics_code: raise ValueError("NAICS code cannot be empty") if state: state = state.strip().upper() if len(state) != 2: raise ValueError(f"State code must be 2 letters, got: {state}") if zip_code: # Convert to string and strip whitespace zip_code_str = str(zip_code).strip() # Remove any non-digit characters zip_code_clean = ''.join(c for c in zip_code_str if c.isdigit()) # Validate length if len(zip_code_clean) > 5: raise ValueError(f"ZIP code must be 5 digits or less, got: {zip_code_clean}") # Zero-pad to 5 digits zip_code = zip_code_clean.zfill(5) if city: city = city.strip() if not city: raise ValueError("City cannot be empty") try: logger.info(f"Searching facilities with filters: facility_name={facility_name}, " f"naics_code={naics_code}, state={state}, zip_code={zip_code}, city={city}, limit={limit}") # Use FRS client to search facilities async with FRSClient() as client: facilities = await client.search_facilities( facility_name=facility_name, naics_code=naics_code, state=state, zip_code=zip_code, city=city, limit=limit ) logger.info(f"Found {len(facilities)} facilities matching search criteria") return facilities except ValueError: # Re-raise validation errors raise except EPAAPIError as e: logger.error(f"EPA API error during facility search: {e}") raise Exception(f"Failed to search facilities: {e}") except Exception as e: logger.error(f"Unexpected error during facility search: {e}") raise Exception(f"Failed to search facilities: {e}") # Register the tool with FastMCP