EPA Envirofacts MCP Server

Instructions

Implementation Reference

• src/tools/chemical_releases.py:211-243 (handler)

  The registered MCP tool handler function `get_chemical_release_data` that defines the tool interface and delegates to the core implementation.
  async def get_chemical_release_data( chemical_name: Optional[str] = None, cas_number: Optional[str] = None, state: Optional[str] = None, county: Optional[str] = None, year: Optional[int] = None, limit: int = 100 ) -> ChemicalReleaseData: """Query TRI chemical releases with flexible search parameters. Provides comprehensive chemical release data from the Toxics Release Inventory (TRI), allowing searches by chemical name, CAS number, state, county, and year. Args: chemical_name: Chemical name (partial match) cas_number: CAS Registry Number (exact match) state: Two-letter state code county: County name (filtered client-side) year: Reporting year (None for most recent available) limit: Maximum results to return (default: 100) Returns: Comprehensive chemical release data with aggregations """ # Call the actual implementation function return await _get_chemical_release_data_impl( chemical_name=chemical_name, cas_number=cas_number, state=state, county=county, year=year, limit=limit )
• src/tools/chemical_releases.py:21-207 (helper)

  Core helper function implementing the tool logic: input validation, querying TRIClient, aggregating data by facility and chemical, calculating summaries and top lists.
  async def _get_chemical_release_data_impl( chemical_name: Optional[str] = None, cas_number: Optional[str] = None, state: Optional[str] = None, county: Optional[str] = None, year: Optional[int] = None, limit: int = 100, ) -> ChemicalReleaseData: """Query TRI chemical releases with flexible search parameters. This tool provides comprehensive chemical release data from the Toxics Release Inventory (TRI), allowing searches by chemical name, CAS number, state, county, and year. Results include both facility-centric and chemical-centric aggregations with optional year-over-year trends. Args: chemical_name: Chemical name (partial match, e.g., 'benzene') cas_number: CAS Registry Number (exact match, e.g., '71-43-2') state: Two-letter state code (e.g., 'NY', 'CA') county: County name (filtered client-side) year: Reporting year (None for most recent available) limit: Maximum results to return (default: 100, max: 1000) Returns: ChemicalReleaseData containing: - Search parameters used - Summary statistics (total facilities, chemicals, releases) - Releases by medium (air, water, land, underground injection) - Facilities grouped by facility with all their chemical releases - Chemicals grouped by chemical with all facilities releasing them - Top facilities and chemicals by total releases Raises: ValueError: If no search parameters provided or parameters invalid Exception: If EPA API queries fail Example: >>> # Search by chemical name >>> data = await get_chemical_release_data(chemical_name="benzene", state="CA") >>> print(f"Found {data.total_facilities} facilities releasing benzene") >>> # Search by CAS number >>> data = await get_chemical_release_data(cas_number="71-43-2", year=2022) >>> print(f"Total releases: {data.total_releases} pounds") """ # Validate input parameters if not any([chemical_name, cas_number, state, county]): raise ValueError("At least one search parameter must be provided (chemical_name, cas_number, state, or county)") if limit <= 0 or limit > 1000: raise ValueError("Limit must be between 1 and 1000") if state and len(state) != 2: raise ValueError("State must be a two-letter code (e.g., 'NY', 'CA')") # Clean and validate parameters search_params = {} if chemical_name: chemical_name = chemical_name.strip() search_params['chemical_name'] = chemical_name if cas_number: cas_number = cas_number.strip() search_params['cas_number'] = cas_number if state: state = state.strip().upper() search_params['state'] = state if county: county = county.strip() search_params['county'] = county if year: search_params['year'] = year try: logger.info(f"Getting chemical release data with params: {search_params}") # Initialize TRI client async with TRIClient() as tri_client: # Get chemical releases releases = await tri_client.get_chemical_releases( chemical_name=chemical_name, cas_number=cas_number, state=state, county=county, year=year, limit=limit ) if not releases: logger.info("No chemical releases found") return ChemicalReleaseData( search_params=search_params, total_facilities=0, total_chemicals=0, total_releases=0.0, query_timestamp=datetime.utcnow().isoformat() + "Z" ) # Apply county filter client-side if specified if county: # Note: This would require additional facility data for county matching # For now, we'll include all releases logger.info(f"County filtering not yet implemented - returning all releases") # Aggregate data by facility facilities_dict = defaultdict(list) for release in releases: facilities_dict[release.facility_id].append(release) facilities = [] for facility_id, facility_releases in facilities_dict.items(): facility_info = FacilityReleaseInfo( facility_id=facility_id, facility_name=facility_releases[0].facility_name, chemical_releases=facility_releases ) facilities.append(facility_info) # Aggregate data by chemical chemicals_dict = defaultdict(list) for release in releases: chemical_key = f"{release.chemical_name}_{release.cas_number or 'unknown'}" chemicals_dict[chemical_key].append(release) chemicals = [] for chemical_key, chemical_releases in chemicals_dict.items(): chemical_agg = ChemicalAggregation( chemical_name=chemical_releases[0].chemical_name, cas_number=chemical_releases[0].cas_number, facilities_releasing=chemical_releases ) chemicals.append(chemical_agg) # Calculate totals by medium air_releases = sum(r.air_release or 0.0 for r in releases) water_releases = sum(r.water_release or 0.0 for r in releases) land_releases = sum(r.land_release or 0.0 for r in releases) underground_injections = sum(r.underground_injection or 0.0 for r in releases) total_releases = sum(r.total_release for r in releases) # Sort facilities and chemicals by total releases facilities.sort(key=lambda f: f.total_releases, reverse=True) chemicals.sort(key=lambda c: c.total_releases, reverse=True) # Get top facilities and chemicals (top 20 each) top_facilities = facilities[:20] top_chemicals = chemicals[:20] # Determine reporting year reporting_year = None if year: reporting_year = year elif releases: # Use the most common year in the results years = [r.reporting_year for r in releases] reporting_year = max(set(years), key=years.count) # Build response result = ChemicalReleaseData( search_params=search_params, total_facilities=len(facilities), total_chemicals=len(chemicals), total_releases=total_releases, air_releases=air_releases, water_releases=water_releases, land_releases=land_releases, underground_injections=underground_injections, facilities=facilities, chemicals=chemicals, top_facilities=top_facilities, top_chemicals=top_chemicals, reporting_year=reporting_year, query_timestamp=datetime.utcnow().isoformat() + "Z" ) logger.info(f"Chemical release data complete: {result.total_facilities} facilities, " f"{result.total_chemicals} chemicals, {result.total_releases:.1f} pounds total") return result except ValueError: # Re-raise validation errors raise except Exception as e: logger.error(f"Failed to get chemical release data: {e}") raise Exception(f"Failed to retrieve chemical release data: {e}")
• src/models/releases.py:93-122 (schema)

  Pydantic BaseModel defining the output schema `ChemicalReleaseData` with fields for search params, statistics, medium breakdowns, facility/chemical aggregations, top lists, and metadata.
  class ChemicalReleaseData(BaseModel): """Comprehensive chemical release data response.""" # Search parameters search_params: Dict[str, Any] = Field(..., description="Search parameters used") # Summary statistics total_facilities: int = Field(..., ge=0, description="Total facilities with releases") total_chemicals: int = Field(..., ge=0, description="Total unique chemicals released") total_releases: float = Field(..., ge=0, description="Total releases across all media (pounds)") # Releases by medium air_releases: float = Field(default=0.0, ge=0, description="Total air releases (pounds)") water_releases: float = Field(default=0.0, ge=0, description="Total water releases (pounds)") land_releases: float = Field(default=0.0, ge=0, description="Total land releases (pounds)") underground_injections: float = Field(default=0.0, ge=0, description="Total underground injections (pounds)") # Aggregated data facilities: List[FacilityReleaseInfo] = Field(default_factory=list, description="Facilities grouped by facility") chemicals: List[ChemicalAggregation] = Field(default_factory=list, description="Chemicals grouped by chemical") # Top facilities and chemicals top_facilities: List[FacilityReleaseInfo] = Field(default_factory=list, description="Top facilities by total releases") top_chemicals: List[ChemicalAggregation] = Field(default_factory=list, description="Top chemicals by total releases") # Metadata reporting_year: Optional[int] = Field(None, description="Year of data (if single year)") query_timestamp: str = Field(..., description="Timestamp of query") data_sources: List[str] = Field(default=["TRI"], description="Data sources used")
• src/tools/chemical_releases.py:247-254 (registration)

  Local registration function that applies the `@mcp.tool()` decorator to the handler function.
  def register_tool(mcp: FastMCP): """Register the chemical release data tool with FastMCP. Args: mcp: FastMCP instance """ mcp.tool()(get_chemical_release_data)
• server.py:95-100 (registration)

  Server main() function calls the module's register_tool to register the tool with the FastMCP instance and logs the registered tools list including 'get_chemical_release_data'.
  register_tool(mcp) register_search_tool(mcp) register_compliance_tool(mcp) register_chemical_tool(mcp) logger.info("Registered tools", tools=["environmental_summary_by_location", "search_facilities", "get_facility_compliance_history", "get_chemical_release_data", "health_check"])