EPA Envirofacts MCP Server

Instructions

Implementation Reference

• server.py:54-83 (handler)

  The MCP tool handler for 'health_check', decorated with @mcp.tool() for automatic registration. It checks server and EPA API health by calling FRSClient.health_check().

  @mcp.tool()
  async def health_check() -> Dict[str, Any]:
      """System health and API connectivity check.
      
      Returns:
          Health status information including EPA API connectivity
      """
      try:
          from src.client import FRSClient
          
          # Test EPA API connectivity
          async with FRSClient() as client:
              health_status = await client.health_check()
          
          return {
              "status": "healthy",
              "version": "1.0.0",
              "server_name": "epa-envirofacts",
              "epa_api": health_status,
              "timestamp": structlog.get_logger().info("Health check completed")
          }
          
      except Exception as e:
          return {
              "status": "unhealthy",
              "version": "1.0.0", 
              "server_name": "epa-envirofacts",
              "error": str(e),
              "timestamp": structlog.get_logger().error(f"Health check failed: {e}")
          }

• src/client/base.py:208-234 (helper)

  Helper method in BaseEPAClient (FRSClient base) that performs the actual EPA API health check via a test query, used by the tool handler.

  async def health_check(self) -> Dict[str, Any]:
      """Check EPA API connectivity.
      
      Returns:
          Health status information
      """
      try:
          # Simple test query to FRS facility_site table
          test_url = self._build_query_url('frs.frs_facility_site', limit=1)
          await self._execute_query(test_url)
          
          return {
              "status": "healthy",
              "api_reachable": True,
              "base_url": self.base_url,
              "timeout": self.timeout
          }
          
      except Exception as e:
          logger.error(f"EPA API health check failed: {e}")
          return {
              "status": "unhealthy",
              "api_reachable": False,
              "error": str(e),
              "base_url": self.base_url,
              "timeout": self.timeout
          }