EPA Envirofacts MCP Server

Instructions

Implementation Reference

• src/tools/compliance_history.py:157-177 (handler)

  The FastMCP tool handler function 'get_facility_compliance_history_tool' decorated with @mcp.tool(), which delegates to the core helper function.

  async def get_facility_compliance_history_tool(
      registry_id: str,
      program: Optional[str] = None,
      years: int = 5
  ) -> FacilityComplianceHistory:
      """Get compliance and enforcement history for an EPA-regulated facility.
      
      Retrieves compliance status, violations, and enforcement history for EPA-regulated
      facilities across RCRA and TRI programs. Supports both FRS registry IDs and 
      program-specific IDs with intelligent fallback logic.
      
      Args:
          registry_id: FRS Registry ID or program-specific ID (RCRA Handler ID, TRI Facility ID)
          program: Optional program filter ('TRI' or 'RCRA')
          years: Historical years to include (default: 5)
          
      Returns:
          Complete compliance history with facility information, compliance records,
          violations, and summary statistics
      """
      return await get_facility_compliance_history(registry_id, program, years)

• src/models/compliance.py:43-56 (schema)

  Pydantic BaseModel defining the output structure FacilityComplianceHistory, including compliance records, status, violations summary, used by the tool.

  class FacilityComplianceHistory(BaseModel):
      """Complete compliance history for a facility."""
      
      facility_info: FacilityInfo = Field(..., description="Facility information")
      compliance_records: List[ComplianceRecord] = Field(default_factory=list, description="Compliance records by program")
      overall_status: ComplianceStatus = Field(..., description="Overall facility compliance status")
      total_violations: int = Field(0, description="Total violations across all programs")
      total_penalties: Optional[float] = Field(None, description="Total penalties across all programs")
      years_analyzed: int = Field(..., description="Number of years analyzed")
      last_updated: Optional[date] = Field(None, description="Date of last compliance update")
      
      def __str__(self) -> str:
          return f"Compliance History for {self.facility_info.name} ({self.facility_info.registry_id})"

• server.py:95-98 (registration)

  Registration calls for all tools, including register_compliance_tool(mcp) which registers the get_facility_compliance_history_tool.

  register_tool(mcp)
  register_search_tool(mcp)
  register_compliance_tool(mcp)
  register_chemical_tool(mcp)

• src/tools/compliance_history.py:150-178 (registration)

  The register_tool function that defines the @mcp.tool() handler and registers it with the FastMCP instance.

  def register_tool(mcp: FastMCP):
      """Register the compliance history tool with FastMCP.
      
      Args:
          mcp: FastMCP instance
      """
      @mcp.tool()
      async def get_facility_compliance_history_tool(
          registry_id: str,
          program: Optional[str] = None,
          years: int = 5
      ) -> FacilityComplianceHistory:
          """Get compliance and enforcement history for an EPA-regulated facility.
          
          Retrieves compliance status, violations, and enforcement history for EPA-regulated
          facilities across RCRA and TRI programs. Supports both FRS registry IDs and 
          program-specific IDs with intelligent fallback logic.
          
          Args:
              registry_id: FRS Registry ID or program-specific ID (RCRA Handler ID, TRI Facility ID)
              program: Optional program filter ('TRI' or 'RCRA')
              years: Historical years to include (default: 5)
              
          Returns:
              Complete compliance history with facility information, compliance records,
              violations, and summary statistics
          """
          return await get_facility_compliance_history(registry_id, program, years)

• src/tools/compliance_history.py:17-147 (helper)

  Core helper function implementing the business logic: validates inputs, fetches facility info and compliance records from clients, computes overall status and totals, constructs and returns FacilityComplianceHistory.

  async def get_facility_compliance_history(
      registry_id: str,
      program: Optional[str] = None,
      years: int = 5
  ) -> FacilityComplianceHistory:
      """Get compliance and enforcement history for an EPA-regulated facility.
      
      This tool retrieves compliance status, violations, and enforcement history
      for EPA-regulated facilities across RCRA and TRI programs. It supports both
      FRS registry IDs and program-specific IDs with intelligent fallback logic.
      
      Args:
          registry_id: FRS Registry ID or program-specific ID (RCRA Handler ID, TRI Facility ID)
          program: Optional program filter ('TRI' or 'RCRA')
          years: Historical years to include (default: 5, max: 20)
      
      Returns:
          FacilityComplianceHistory containing:
          - Facility information
          - Compliance records by program
          - Violations with dates and status
          - Overall compliance status
          - Summary statistics
      
      Raises:
          ValueError: If parameters are invalid
          Exception: If EPA API queries fail
      
      Example:
          >>> # By FRS registry ID
          >>> compliance = await get_facility_compliance_history("110000012345")
          >>> 
          >>> # By program-specific ID with filter
          >>> compliance = await get_facility_compliance_history("VAD000012345", program="RCRA")
          >>> 
          >>> # With custom timeframe
          >>> compliance = await get_facility_compliance_history("110000012345", years=10)
      """
      # Validate input parameters
      if not registry_id or not registry_id.strip():
          raise ValueError("Registry ID cannot be empty")
      
      if not (1 <= years <= 20):
          raise ValueError("Years must be between 1 and 20")
      
      if program and program.upper() not in ['TRI', 'RCRA']:
          raise ValueError("Program must be 'TRI' or 'RCRA'")
      
      registry_id = registry_id.strip()
      if program:
          program = program.strip().upper()
      
      try:
          logger.info(f"Getting compliance history for facility {registry_id} "
                     f"(program: {program or 'all'}, years: {years})")
          
          # Step 1: Get facility information from FRS
          facility_info = None
          try:
              async with FRSClient() as frs_client:
                  facility_info = await frs_client.get_facility_by_id(registry_id)
          except Exception as e:
              logger.warning(f"Failed to get facility info from FRS for {registry_id}: {e}")
          
          # If FRS lookup failed, create a basic facility info
          if not facility_info:
              facility_info = FacilityInfo(
                  registry_id=registry_id,
                  name=f"Facility {registry_id}",
                  programs=[FacilityType.FRS]
              )
          
          # Step 2: Get compliance records
          compliance_records = []
          try:
              async with ComplianceClient() as compliance_client:
                  compliance_records = await compliance_client.get_compliance_by_registry_id(
                      registry_id=registry_id,
                      program=program,
                      years=years
                  )
          except Exception as e:
              logger.error(f"Failed to get compliance records for {registry_id}: {e}")
              raise Exception(f"Failed to retrieve compliance data: {e}")
          
          # Step 3: Calculate overall status and statistics
          overall_status = ComplianceStatus.UNKNOWN
          total_violations = 0
          total_penalties = None
          
          if compliance_records:
              # Determine overall status based on individual program statuses
              statuses = [record.status for record in compliance_records]
              if ComplianceStatus.VIOLATION in statuses:
                  overall_status = ComplianceStatus.VIOLATION
              elif ComplianceStatus.COMPLIANT in statuses:
                  overall_status = ComplianceStatus.COMPLIANT
              
              # Calculate totals
              total_violations = sum(record.violation_count for record in compliance_records)
              
              # Sum penalties (if available)
              penalties = [record.total_penalties for record in compliance_records 
                          if record.total_penalties is not None]
              if penalties:
                  total_penalties = sum(penalties)
          
          # Step 4: Build compliance history
          compliance_history = FacilityComplianceHistory(
              facility_info=facility_info,
              compliance_records=compliance_records,
              overall_status=overall_status,
              total_violations=total_violations,
              total_penalties=total_penalties,
              years_analyzed=years,
              last_updated=date.today()
          )
          
          logger.info(f"Compliance history complete for {registry_id}: "
                     f"{overall_status}, {total_violations} violations, "
                     f"{len(compliance_records)} programs")
          
          return compliance_history
          
      except ValueError:
          # Re-raise validation errors
          raise
      except Exception as e:
          logger.error(f"Failed to get compliance history for {registry_id}: {e}")
          raise Exception(f"Failed to retrieve compliance history: {e}")