"""
Extended Hook Types for ToolUniverse
This module demonstrates how to extend the hook system with additional
hook types beyond summarization. It shows the pattern for creating
new hook types while maintaining compatibility with the existing system.
"""
import json
import re
from pathlib import Path
from typing import Dict, Any, List
from jsonschema import FormatChecker
from jsonschema.exceptions import SchemaError
from jsonschema.validators import validator_for
from .output_hook import HookProcessingError, OutputHook
[docs]
class FilteringHook(OutputHook):
"""
Hook for filtering sensitive or unwanted content from tool outputs.
This hook can be used to:
- Remove sensitive information (emails, phones, SSNs)
- Filter inappropriate content
- Sanitize data before display
Args:
config (Dict[str, Any]): Hook configuration containing filter settings
tooluniverse: Optional ToolUniverse instance (not used for filtering)
"""
[docs]
def __init__(self, config: Dict[str, Any], tooluniverse=None):
"""
Initialize the filtering hook with configuration.
Args:
config (Dict[str, Any]): Hook configuration
tooluniverse: ToolUniverse instance (optional, not used)
"""
super().__init__(config)
hook_config = config.get("hook_config", {})
# Filter configuration
self.filter_patterns = hook_config.get("filter_patterns", [])
self.replacement_text = hook_config.get("replacement_text", "[REDACTED]")
self.preserve_structure = hook_config.get("preserve_structure", True)
self.log_filtered_items = hook_config.get("log_filtered_items", False)
# Compile regex patterns for efficiency
self.compiled_patterns = []
for pattern in self.filter_patterns:
try:
compiled = re.compile(pattern, re.IGNORECASE)
self.compiled_patterns.append(compiled)
except re.error as e:
print(f"Warning: Invalid regex pattern '{pattern}': {e}")
[docs]
def process(
self,
result: Any,
tool_name: str,
arguments: Dict[str, Any],
context: Dict[str, Any],
) -> Any:
"""
Apply filtering to the tool output.
Args:
result (Any): The tool output to filter
tool_name (str): Name of the tool that produced the output
arguments (Dict[str, Any]): Arguments passed to the tool
context (Dict[str, Any]): Additional context information
Returns
Any: The filtered output, or original output if filtering fails
"""
try:
if not self.compiled_patterns:
return result
value = result if self.preserve_structure else str(result)
filtered_output, filtered_count = self._filter_value(value)
if filtered_count > 0:
print(
f"🔒 FilteringHook: Filtered {filtered_count} sensitive items from {tool_name} output"
)
return filtered_output
except Exception as e:
print(f"Error in filtering hook: {str(e)}")
return result
[docs]
def _filter_value(self, value: Any) -> tuple[Any, int]:
"""Filter string values recursively while preserving container types."""
if isinstance(value, str):
return self._filter_text(value)
if isinstance(value, dict):
filtered = {}
total = 0
for key, item in value.items():
filtered_item, count = self._filter_value(item)
filtered[key] = filtered_item
total += count
return filtered, total
if isinstance(value, list):
filtered_items = []
total = 0
for item in value:
filtered_item, count = self._filter_value(item)
filtered_items.append(filtered_item)
total += count
return filtered_items, total
if isinstance(value, tuple):
filtered_items = []
total = 0
for item in value:
filtered_item, count = self._filter_value(item)
filtered_items.append(filtered_item)
total += count
return tuple(filtered_items), total
return value, 0
[docs]
def _filter_text(self, text: str) -> tuple[str, int]:
"""Apply configured regular expressions to one string value."""
filtered = text
total = 0
for pattern in self.compiled_patterns:
filtered, count = pattern.subn(self.replacement_text, filtered)
total += count
if count and self.log_filtered_items:
print(
f"Filtered {count} items matching pattern: {pattern.pattern}"
)
return filtered, total
[docs]
class ValidationHook(OutputHook):
"""
Hook for validating tool outputs against schemas or rules.
This hook can be used to:
- Validate JSON against schemas
- Check required fields
- Ensure data quality
``strict_mode`` enables JSON Schema format checks. ``error_action`` may warn,
apply the supported missing-field fix, or fail the tool call.
Args:
config (Dict[str, Any]): Hook configuration containing validation settings
tooluniverse: Optional ToolUniverse instance (not used for validation)
"""
[docs]
def __init__(self, config: Dict[str, Any], tooluniverse=None):
"""
Initialize the validation hook with configuration.
Args:
config (Dict[str, Any]): Hook configuration
tooluniverse: ToolUniverse instance (optional, not used)
"""
super().__init__(config)
hook_config = config.get("hook_config", {})
self.validation_schema = hook_config.get("validation_schema")
self.strict_mode = hook_config.get("strict_mode", True)
self.error_action = hook_config.get("error_action", "warn")
self.required_fields = hook_config.get("required_fields", [])
if self.error_action not in {"warn", "fix", "fail"}:
raise ValueError(
"ValidationHook error_action must be one of: warn, fix, fail"
)
if not isinstance(self.strict_mode, bool):
raise ValueError("ValidationHook strict_mode must be a boolean")
if not isinstance(self.required_fields, list) or any(
not isinstance(field, str) or not field
for field in self.required_fields
):
raise ValueError(
"ValidationHook required_fields must be a list of non-empty strings"
)
self._schema_validator = None
if self.validation_schema is not None:
if not isinstance(self.validation_schema, dict):
raise ValueError("ValidationHook validation_schema must be an object")
validator_class = validator_for(self.validation_schema)
try:
validator_class.check_schema(self.validation_schema)
except SchemaError as exc:
raise ValueError(
f"Invalid ValidationHook validation_schema: {exc.message}"
) from exc
format_checker = FormatChecker() if self.strict_mode else None
self._schema_validator = validator_class(
self.validation_schema, format_checker=format_checker
)
[docs]
def process(
self,
result: Any,
tool_name: str,
arguments: Dict[str, Any],
context: Dict[str, Any],
) -> Any:
"""
Apply validation to the tool output.
Args:
result (Any): The tool output to validate
tool_name (str): Name of the tool that produced the output
arguments (Dict[str, Any]): Arguments passed to the tool
context (Dict[str, Any]): Additional context information
Returns
Any: The validated, fixed, or warning-preserved output
Raises:
HookProcessingError: If validation fails with ``error_action="fail"``
"""
try:
validation_result = self._validate_output(result)
if validation_result["valid"]:
if validation_result["warnings"]:
print(
f"✅ ValidationHook: {tool_name} output validated with warnings"
)
else:
print(
f"✅ ValidationHook: {tool_name} output validated successfully"
)
return result
else:
details = "; ".join(validation_result["errors"])
if self.error_action == "fail":
raise HookProcessingError(
f"ValidationHook rejected {tool_name} output: {details}"
)
elif self.error_action == "fix":
fixed_result = self._fix_output(result, validation_result["errors"])
remaining = self._validate_output(fixed_result)
if remaining["valid"]:
print(f"🔧 ValidationHook: Fixed {tool_name} output")
else:
print(
f"⚠️ ValidationHook: {tool_name} output remains invalid "
f"after supported fixes: {'; '.join(remaining['errors'])}"
)
return fixed_result
else: # warn
print(
f"⚠️ ValidationHook: {tool_name} output has validation "
f"issues: {details}"
)
return result
except HookProcessingError:
raise
except Exception as e:
print(f"Error in validation hook: {str(e)}")
return result
[docs]
def _validate_output(self, result: Any) -> Dict[str, Any]:
"""Validate the output against configured rules."""
validation_result = {"valid": True, "errors": [], "warnings": []}
if self.required_fields:
if not isinstance(result, dict):
validation_result["errors"].append(
"Required fields can only be checked on object outputs"
)
else:
for field in self.required_fields:
if field not in result:
validation_result["errors"].append(
f"Missing required field: {field}"
)
if self._schema_validator is not None:
schema_errors = sorted(
self._schema_validator.iter_errors(result),
key=lambda error: tuple(str(part) for part in error.absolute_path),
)
for error in schema_errors:
path = ".".join(str(part) for part in error.absolute_path) or "<root>"
validation_result["errors"].append(
f"Schema validation failed at {path}: {error.message}"
)
validation_result["valid"] = not validation_result["errors"]
return validation_result
[docs]
def _fix_output(self, result: Any, errors: List[str]) -> Any:
"""Attempt to fix validation errors."""
# Simple fixes for common issues
if isinstance(result, dict):
fixed_result = result.copy()
for error in errors:
if "Missing required field" in error:
field_name = error.split(": ")[1]
fixed_result[field_name] = None # Add missing field with None value
return fixed_result
return result
[docs]
class LoggingHook(OutputHook):
"""
Hook for logging tool outputs and execution details.
This hook can be used to:
- Log all tool outputs
- Track execution metrics
- Audit tool usage
Args:
config (Dict[str, Any]): Hook configuration containing logging settings
tooluniverse: Optional ToolUniverse instance (not used for logging)
"""
[docs]
def __init__(self, config: Dict[str, Any], tooluniverse=None):
"""
Initialize the logging hook with configuration.
Args:
config (Dict[str, Any]): Hook configuration
tooluniverse: ToolUniverse instance (optional, not used)
"""
super().__init__(config)
hook_config = config.get("hook_config", {})
# Logging configuration
self.log_level = hook_config.get("log_level", "INFO")
self.log_format = hook_config.get(
"log_format", "detailed"
) # simple, detailed, json
self.log_file = hook_config.get("log_file", None)
self.max_log_size = hook_config.get("max_log_size", 1000) # characters
[docs]
def process(
self,
result: Any,
tool_name: str,
arguments: Dict[str, Any],
context: Dict[str, Any],
) -> Any:
"""
Log the tool output and execution details.
Args:
result (Any): The tool output to log
tool_name (str): Name of the tool that produced the output
arguments (Dict[str, Any]): Arguments passed to the tool
context (Dict[str, Any]): Additional context information
Returns
Any: The original output (logging doesn't modify the output)
"""
try:
log_entry = self._create_log_entry(result, tool_name, arguments, context)
self._write_log(log_entry)
except Exception as e:
print(f"Error in logging hook: {str(e)}")
# Logging hook always returns the original result unchanged
return result
[docs]
def _create_log_entry(
self,
result: Any,
tool_name: str,
arguments: Dict[str, Any],
context: Dict[str, Any],
) -> str:
"""Create a log entry for the tool execution."""
if self.log_format == "simple":
return f"Tool: {tool_name} | Args: {arguments} | Output length: {len(str(result))}"
elif self.log_format == "json":
return json.dumps(
{
"tool_name": tool_name,
"arguments": arguments,
"output_length": len(str(result)),
"timestamp": context.get("execution_time", "unknown"),
"output_preview": str(result)[: self.max_log_size],
},
ensure_ascii=False,
default=str,
)
else: # detailed
return f"""
Tool Execution Log:
==================
Tool: {tool_name}
Arguments: {arguments}
Execution Time: {context.get("execution_time", "unknown")}
Output Length: {len(str(result))} characters
Output Preview: {str(result)[: self.max_log_size]}{"..." if len(str(result)) > self.max_log_size else ""}
==================
"""
[docs]
def _write_log(self, log_entry: str):
"""Write the log entry to the configured destination."""
if self.log_file:
log_path = Path(self.log_file).expanduser()
log_path.parent.mkdir(parents=True, exist_ok=True)
with log_path.open("a", encoding="utf-8") as f:
f.write(log_entry + "\n")
else:
print(f"📝 Log: {log_entry}")
# Hook type registry for easy extension
HOOK_TYPE_REGISTRY = {
"SummarizationHook": "SummarizationHook", # Import from parent module
"FilteringHook": FilteringHook,
"FormattingHook": FormattingHook,
"ValidationHook": ValidationHook,
"LoggingHook": LoggingHook,
}
