52 lines
1.8 KiB
Python
52 lines
1.8 KiB
Python
"""Formal validators: JSON schema, policy constraints, lint/type check on outputs."""
|
|
|
|
import json
|
|
from typing import Any
|
|
|
|
from fusionagi._logger import logger
|
|
|
|
|
|
class FormalValidators:
|
|
"""
|
|
Validates outputs against JSON schema, policy constraints.
|
|
Extend with lint/type check for code outputs.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._schemas: dict[str, dict[str, Any]] = {} # name -> JSON schema
|
|
|
|
def register_schema(self, name: str, schema: dict[str, Any]) -> None:
|
|
"""Register a JSON schema for output validation."""
|
|
self._schemas[name] = schema
|
|
|
|
def validate_json(self, data: str | dict[str, Any], schema_name: str | None = None) -> tuple[bool, str]:
|
|
"""
|
|
Validate JSON structure. If schema_name given, validate against that schema
|
|
(requires jsonschema lib for full validation; else structure-only).
|
|
Returns (valid, error_message).
|
|
"""
|
|
if isinstance(data, dict):
|
|
obj = data
|
|
else:
|
|
try:
|
|
obj = json.loads(data)
|
|
except json.JSONDecodeError as e:
|
|
return False, str(e)
|
|
if schema_name and schema_name in self._schemas:
|
|
try:
|
|
import jsonschema
|
|
jsonschema.validate(instance=obj, schema=self._schemas[schema_name])
|
|
except ImportError:
|
|
logger.warning("jsonschema not installed; skipping schema validation")
|
|
except Exception as e:
|
|
return False, str(e)
|
|
return True, ""
|
|
|
|
def validate_policy(self, action: str, context: dict[str, Any]) -> tuple[bool, str]:
|
|
"""
|
|
Check action against policy (context has tool_name, domain, etc.).
|
|
Returns (allowed, reason).
|
|
"""
|
|
# Placeholder: no policy rules by default
|
|
return True, ""
|