fromfastagenticimportReadinessCheckerchecker=ReadinessChecker()# Run all checksreport=awaitchecker.run_checks()# Run with configurationconfig={"auth":{"enabled":True},"timeout":30,"telemetry":True,}report=awaitchecker.run_checks(config)
# Get failures onlyfailures=report.get_failures()# Get warnings onlywarnings=report.get_warnings()# Get by categorysecurity_results=report.get_by_category(CheckCategory.SECURITY)
asyncdefcheck_external_service(config):"""Check external service connectivity."""try:response=awaithttp_client.get(config.get("service_url"))ifresponse.status_code==200:returnCheckResult(name="external_service",status=CheckStatus.PASS,message="External service reachable",category=CheckCategory.RELIABILITY,)exceptExceptionase:passreturnCheckResult(name="external_service",status=CheckStatus.WARN,message="External service unreachable",category=CheckCategory.RELIABILITY,recommendation="Verify service URL and network connectivity",)checker.add_check(ReadinessCheck(name="external_service",description="Check external service connectivity",category=CheckCategory.RELIABILITY,check_fn=check_external_service,is_async=True,))
Checks that raise exceptions are marked as failed:
defrisky_check(config):raiseValueError("Something went wrong")checker.add_check(ReadinessCheck(name="risky",description="A risky check",category=CheckCategory.CONFIGURATION,check_fn=risky_check,))report=awaitchecker.run_checks()# The check will show as FAIL with the exception message
fromfastagenticimportApp,ReadinessCheckerapp=App()@app.on_event("startup")asyncdefvalidate_readiness():checker=ReadinessChecker()report=awaitchecker.run_checks(app.config)ifnotreport.is_ready:failures=[r.nameforrinreport.get_failures()]raiseRuntimeError(f"App not ready for production: {failures}")ifreport.warnings:forwarninginreport.get_warnings():logger.warning(f"Readiness warning: {warning.name}")
# Comprehensive configuration for all checksconfig={# Security"auth":{"enabled":True},"oidc_issuer":"https://auth.example.com","https":{"enabled":True},# Reliability"timeout":30,"retry_policy":{"max_attempts":3,"backoff":"exponential"},"rate_limit":{"rpm":60,"rpd":10000},# Observability"logging":{"level":"INFO","format":"json"},"telemetry":True,"health_endpoint":"/health",# Compliance"pii_detection":{"enabled":True},"audit":{"enabled":True},}report=awaitchecker.run_checks(config)print(f"Score: {report.score}/100")# Should be 100 with full config