Utilities
Helper functions and utility classes provided by the Claif framework.
Common Utilities
Message Formatting
format_response(message, format_type="text")
Format a message response for display.
from claif.common.utils import format_response
from claif import ClaifClient
client = ClaifClient()
response = client.query("Hello")
# Format as text (default)
text = format_response(response)
print(text)
# Format as markdown
markdown = format_response(response, format_type="markdown")
# Format as JSON
json_str = format_response(response, format_type="json")
Parameters:
message
(Message) - The message to formatformat_type
(str) - Output format (“text”, “markdown”, “json”, “html”)
Returns: str
- Formatted message content
format_metrics(metrics)
Format performance metrics for display.
from claif.common.utils import format_metrics
# Metrics from a query
metrics = {
'response_time': 2.34,
'tokens_used': 150,
'provider': 'claude',
'model': 'claude-3-sonnet'
}
formatted = format_metrics(metrics)
print(formatted)
# Output:
# Provider: claude (claude-3-sonnet)
# Response time: 2.34s
# Tokens used: 150
Parameters:
metrics
(Dict[str, Any]) - Metrics dictionary
Returns: str
- Formatted metrics display
Configuration Utilities
load_config(path=None)
Load configuration from file with fallback defaults.
from claif.common.utils import load_config
# Load from default location
config = load_config()
# Load from specific path
config = load_config("./custom-config.toml")
Parameters:
path
(Optional[str]) - Path to config file
Returns: Config
- Configuration object
validate_config(config)
Validate configuration object.
from claif.common.utils import validate_config, load_config
from claif.common.errors import ConfigurationError
config = load_config()
try:
validate_config(config)
print("Configuration is valid")
except ConfigurationError as e:
print(f"Invalid configuration: {e}")
Parameters:
config
(Config) - Configuration to validate
Raises: ConfigurationError
- If configuration is invalid
Provider Utilities
discover_providers()
Discover and return available providers.
from claif.common.utils import discover_providers
providers = discover_providers()
print(f"Found providers: {list(providers.keys())}")
# Check specific provider
if 'claude' in providers:
claude_provider = providers['claude']
print(f"Claude version: {claude_provider.version}")
Returns: Dict[str, Provider]
- Dictionary of provider name to provider instance
get_provider_info(provider_name)
Get detailed information about a provider.
from claif.common.utils import get_provider_info
info = get_provider_info("claude")
print(f"Name: {info['name']}")
print(f"Version: {info['version']}")
print(f"Available: {info['available']}")
print(f"Description: {info['description']}")
print(f"Models: {info['models']}")
Parameters:
provider_name
(str) - Name of the provider
Returns: Dict[str, Any]
- Provider information
check_provider_health(provider_name, timeout=10.0)
Check if a provider is healthy and responsive.
from claif.common.utils import check_provider_health
# Quick health check
is_healthy = check_provider_health("claude")
print(f"Claude is healthy: {is_healthy}")
# With custom timeout
is_healthy = check_provider_health("gemini", timeout=5.0)
Parameters:
provider_name
(str) - Name of the providertimeout
(float) - Timeout for health check
Returns: bool
- True if provider is healthy
Logging Utilities
setup_logging(level="INFO", file=None)
Set up logging configuration.
from claif.common.utils import setup_logging
# Basic setup
setup_logging()
# With custom level
setup_logging(level="DEBUG")
# With log file
setup_logging(level="INFO", file="claif.log")
Parameters:
level
(str) - Logging level (“DEBUG”, “INFO”, “WARNING”, “ERROR”)file
(Optional[str]) - Log file path
get_logger(name)
Get a configured logger instance.
from claif.common.utils import get_logger
logger = get_logger(__name__)
logger.info("Starting operation")
logger.debug("Debug information")
logger.error("Error occurred")
Parameters:
name
(str) - Logger name (usually__name__
)
Returns: Logger
- Configured logger instance
File and Path Utilities
get_config_dir()
Get the Claif configuration directory.
from claif.common.utils import get_config_dir
import os
config_dir = get_config_dir()
print(f"Config directory: {config_dir}")
# Ensure directory exists
os.makedirs(config_dir, exist_ok=True)
Returns: Path
- Path to configuration directory
get_cache_dir()
Get the Claif cache directory.
from claif.common.utils import get_cache_dir
cache_dir = get_cache_dir()
print(f"Cache directory: {cache_dir}")
Returns: Path
- Path to cache directory
get_data_dir()
Get the Claif data directory.
from claif.common.utils import get_data_dir
data_dir = get_data_dir()
print(f"Data directory: {data_dir}")
Returns: Path
- Path to data directory
Validation Utilities
validate_message(message)
Validate a message object.
from claif.common.utils import validate_message
from claif.common.types import Message, TextBlock
from claif.common.errors import MessageValidationError
message = Message(content=[TextBlock(text="Hello")])
try:
validate_message(message)
print("Message is valid")
except MessageValidationError as e:
print(f"Invalid message: {e}")
Parameters:
message
(Message) - Message to validate
Raises: MessageValidationError
- If message is invalid
validate_options(options, provider_name=None)
Validate query options.
from claif.common.utils import validate_options
from claif.common.types import ClaifOptions
from claif.common.errors import OptionsValidationError
options = ClaifOptions(temperature=0.7, max_tokens=1000)
try:
validate_options(options, provider_name="claude")
print("Options are valid")
except OptionsValidationError as e:
print(f"Invalid options: {e}")
Parameters:
options
(ClaifOptions) - Options to validateprovider_name
(Optional[str]) - Provider to validate against
Raises: OptionsValidationError
- If options are invalid
Terminal Utilities
Rich Formatting
print_response(message, style="default")
Print a formatted response to the terminal.
from claif.common.utils import print_response
from claif import ClaifClient
client = ClaifClient()
response = client.query("Hello")
# Default formatting
print_response(response)
# With different styles
print_response(response, style="markdown")
print_response(response, style="code")
Parameters:
message
(Message) - Message to printstyle
(str) - Formatting style (“default”, “markdown”, “code”, “json”)
print_table(data, headers=None)
Print data as a formatted table.
from claif.common.utils import print_table
# Provider information table
data = [
["claude", "1.0.0", "Available"],
["gemini", "1.2.0", "Available"],
["codex", "0.9.0", "Unavailable"]
]
print_table(data, headers=["Provider", "Version", "Status"])
Parameters:
data
(List[List[str]]) - Table dataheaders
(Optional[List[str]]) - Column headers
print_error(error, show_traceback=False)
Print an error with appropriate formatting.
from claif.common.utils import print_error
from claif.common.errors import ProviderError
try:
# Some operation that fails
pass
except ProviderError as e:
print_error(e)
# With traceback for debugging
print_error(e, show_traceback=True)
Parameters:
error
(Exception) - Error to printshow_traceback
(bool) - Whether to show full traceback
Progress Indicators
progress_spinner(message="Working...")
Context manager for showing a progress spinner.
from claif.common.utils import progress_spinner
import time
with progress_spinner("Processing query..."):
time.sleep(3) # Simulate work
print("Done!")
Parameters:
message
(str) - Message to display
progress_bar(total, description="Progress")
Create a progress bar for long operations.
from claif.common.utils import progress_bar
import time
with progress_bar(100, "Downloading...") as bar:
for i in range(100):
time.sleep(0.01)
bar.update(1)
Parameters:
total
(int) - Total number of stepsdescription
(str) - Progress description
Data Utilities
Serialization
serialize_message(message)
Serialize a message to JSON-compatible format.
from claif.common.utils import serialize_message, deserialize_message
from claif.common.types import Message, TextBlock
message = Message(content=[TextBlock(text="Hello")])
# Serialize
data = serialize_message(message)
print(data) # Dict that can be JSON-serialized
# Deserialize
restored = deserialize_message(data)
print(restored.text) # "Hello"
Parameters:
message
(Message) - Message to serialize
Returns: Dict[str, Any]
- Serialized message data
deserialize_message(data)
Deserialize message from JSON-compatible format.
Parameters:
data
(Dict[str, Any]) - Serialized message data
Returns: Message
- Deserialized message
Caching
cache_response(key, response, ttl=3600)
Cache a response with TTL (time-to-live).
from claif.common.utils import cache_response, get_cached_response
# Cache a response
response = client.query("What is AI?")
cache_response("ai_explanation", response, ttl=3600)
# Retrieve cached response
cached = get_cached_response("ai_explanation")
if cached:
print("Using cached response")
print(cached.text)
Parameters:
key
(str) - Cache keyresponse
(Message) - Response to cachettl
(int) - Time-to-live in seconds
get_cached_response(key)
Get a cached response if available and not expired.
Parameters:
key
(str) - Cache key
Returns: Optional[Message]
- Cached response or None
clear_cache(key=None)
Clear cache entries.
from claif.common.utils import clear_cache
# Clear specific key
clear_cache("ai_explanation")
# Clear all cache
clear_cache()
Parameters:
key
(Optional[str]) - Specific key to clear, or None for all
Network Utilities
HTTP Helpers
make_request(url, method="GET", **kwargs)
Make an HTTP request with proper error handling.
from claif.common.utils import make_request
from claif.common.errors import NetworkError
try:
response = make_request(
"https://api.example.com/status",
method="GET",
headers={"Authorization": "Bearer token"},
timeout=30
)
print(response.json())
except NetworkError as e:
print(f"Network error: {e}")
Parameters:
url
(str) - URL to requestmethod
(str) - HTTP method**kwargs
- Additional arguments for requests
Returns: requests.Response
- HTTP response
Raises: NetworkError
- On network-related errors
check_connectivity(host="8.8.8.8", port=53, timeout=3)
Check internet connectivity.
from claif.common.utils import check_connectivity
if check_connectivity():
print("Internet connection available")
else:
print("No internet connection")
# Check specific host
if check_connectivity("api.anthropic.com", 443):
print("Can reach Anthropic API")
Parameters:
host
(str) - Host to checkport
(int) - Port to checktimeout
(float) - Connection timeout
Returns: bool
- True if connectivity is available
System Utilities
Environment Detection
get_platform_info()
Get platform and system information.
from claif.common.utils import get_platform_info
info = get_platform_info()
print(f"OS: {info['system']}")
print(f"Version: {info['version']}")
print(f"Architecture: {info['architecture']}")
print(f"Python: {info['python_version']}")
Returns: Dict[str, str]
- Platform information
is_docker()
Check if running inside Docker container.
from claif.common.utils import is_docker
if is_docker():
print("Running in Docker")
else:
print("Running on host system")
Returns: bool
- True if running in Docker
get_terminal_size()
Get terminal dimensions.
from claif.common.utils import get_terminal_size
width, height = get_terminal_size()
print(f"Terminal: {width}x{height}")
Returns: Tuple[int, int]
- Terminal width and height
Process Management
run_command(command, cwd=None, timeout=None)
Run a shell command with proper error handling.
from claif.common.utils import run_command
from claif.common.errors import CommandError
try:
result = run_command(["ls", "-la"], cwd="/tmp", timeout=10)
print(f"Exit code: {result.returncode}")
print(f"Output: {result.stdout}")
except CommandError as e:
print(f"Command failed: {e}")
Parameters:
command
(List[str]) - Command and argumentscwd
(Optional[str]) - Working directorytimeout
(Optional[float]) - Command timeout
Returns: subprocess.CompletedProcess
- Command result
Raises: CommandError
- On command execution errors
Usage Examples
Complete Error Handling Setup
from claif import ClaifClient
from claif.common.utils import (
setup_logging,
get_logger,
print_error,
check_connectivity
)
from claif.common.errors import ClaifError
# Setup logging
setup_logging(level="INFO", file="claif.log")
logger = get_logger(__name__)
# Check connectivity first
if not check_connectivity():
print("No internet connection available")
exit(1)
client = ClaifClient()
try:
response = client.query("Hello, AI!")
print(response.text)
except ClaifError as e:
logger.error(f"Claif error: {e}")
print_error(e)
exit(1)
Configuration and Provider Management
from claif.common.utils import (
load_config,
validate_config,
discover_providers,
check_provider_health,
print_table
)
# Load and validate config
config = load_config()
validate_config(config)
# Discover providers
providers = discover_providers()
# Check health and display status
provider_data = []
for name, provider in providers.items():
is_healthy = check_provider_health(name)
status = "Healthy" if is_healthy else "Unavailable"
provider_data.append([name, provider.version, status])
print_table(
provider_data,
headers=["Provider", "Version", "Status"]
)
Response Caching and Formatting
from claif import ClaifClient
from claif.common.utils import (
cache_response,
get_cached_response,
format_response,
serialize_message
)
client = ClaifClient()
cache_key = "common_question"
# Try to get cached response first
cached = get_cached_response(cache_key)
if cached:
print("Using cached response:")
response = cached
else:
print("Fetching new response:")
response = client.query("What is machine learning?")
cache_response(cache_key, response, ttl=3600)
# Format and display
markdown = format_response(response, format_type="markdown")
print(markdown)
# Save to file
with open("response.json", "w") as f:
json.dump(serialize_message(response), f, indent=2)