agent-validator

🚀 Agent Validator

A simple drop-in tool to validate LLM/agent outputs against schemas with automatic retries, logging, and optional cloud monitoring.

---

✨ Features

• 🔍 Schema Validation: Validate JSON outputs against Python dict schemas
• 🔄 Automatic Retries: Retry failed validations with exponential backoff
• 🔧 Type Coercion: Optional safe type coercion (e.g., "42" → 42)
• 📝 Local Logging: JSON Lines logging with automatic redaction
• ☁️ Cloud Monitoring: Optional cloud logging with secure authentication
• 🛠️ CLI Tools: Command-line interface for testing and log management
• 🛡️ Size Limits: Configurable limits to prevent abuse

---

🚀 Quick Start

📦 Installation

pip install agent-validator

💻 Basic Usage

from agent_validator import validate, Schema, ValidationMode

# Define your schema
schema = Schema({
    "name": str,
    "age": int,
    "email": str,
    "tags": [str]
})

# Validate agent output
try:
    result = validate(
        agent_output,  # Your agent's output (string or dict)
        schema,
        retry_fn=call_agent,  # Function to retry if validation fails
        retries=2,
        mode=ValidationMode.COERCE,  # Allow type coercion
        context={"correlation_id": "abc123"}  # Optional correlation ID for tracking
    )
    print("✅ Validation successful!")
    print(result)
except ValidationError as e:
    print(f"❌ Validation failed: {e}")
    print(f"Correlation ID: {e.correlation_id}")  # For debugging

🖥️ CLI Usage

# Test validation with files
agent-validator test schema.json input.json --mode COERCE

# View recent logs
agent-validator logs -n 20

# View cloud logs
agent-validator cloud-logs -n 20

# Open web dashboard (secure proxy)
agent-validator dashboard

# Generate correlation ID
agent-validator id

# Configure cloud logging
agent-validator config --set-license-key YOUR_LICENSE_KEY
agent-validator config --set-log-to-cloud true

---

📋 Schema Definition

Schemas are defined using Python dictionaries with type annotations:

schema = Schema({
    "name": str,           # Required string field
    "age": int,            # Required integer field
    "email": str,          # Required string field
    "is_active": bool,     # Required boolean field
    "score": float,        # Required float field
    "tags": [str],         # List of strings
    "metadata": None,      # Optional field (can be omitted)
    "address": {           # Nested object
        "street": str,
        "city": str,
        "zip": str
    },
    "scores": [int]        # List of integers
})

🎯 Supported Types

• Primitives: str, int, float, bool
• Lists: [type] for lists of that type
• Objects: dict for nested objects
• Optional: None for optional fields

📄 Schema File Formats

When using the CLI with JSON files, you can use either format:

Direct Schema Format (Recommended)

{
  "name": "string",
  "age": "integer",
  "email": "string",
  "is_active": "boolean",
  "tags": ["string"],
  "metadata": {
    "source": "string",
    "version": "string"
  }
}

Wrapped Schema Format

{
  "schema": {
    "name": "string",
    "age": "integer",
    "email": "string",
    "is_active": "boolean"
  }
}

Supported string types: string, integer, int, float, number, boolean, bool, list, array, dict, object

---

🔄 Validation Modes

🚫 Strict Mode (Default)

No type coercion allowed. Input must match schema exactly.

schema = Schema({"age": int})
data = {"age": "30"}  # String instead of int

# This will fail
validate(data, schema, mode=ValidationMode.STRICT)

🔧 Coerce Mode

Safe type coercion is performed:

schema = Schema({"age": int, "is_active": bool})
data = {"age": "30", "is_active": "true"}

# This will succeed and coerce types
result = validate(data, schema, mode=ValidationMode.COERCE)
# result = {"age": 30, "is_active": True}

🔄 Coercion Rules

---

🔄 Retry Logic

When validation fails and a retry_fn is provided, the system will automatically retry:

def call_agent(prompt: str, context: dict) -> str:
    """Your agent function that returns JSON string."""
    # Call your LLM/agent here
    return json.dumps({"name": "John", "age": 30})

result = validate(
    malformed_output,
    schema,
    retry_fn=call_agent,
    retries=2,  # Retry up to 2 times
    timeout_s=20  # 20 second timeout per attempt
)

⚡ Retry Behavior

• Exponential Backoff: Delays increase with each retry (0.5s, 1s, 2s)
• Jitter: Random variation to prevent thundering herd
• Timeout: Per-attempt timeout to prevent hanging
• Context Preservation: Original context is passed to retry function

---

📝 Logging

💾 Local Logging

All validation attempts are logged to ~/.agent_validator/logs/YYYY-MM-DD.jsonl:

{
  "ts": "2023-12-01T10:30:00Z",
  "correlation_id": "abc123-def456",
  "valid": true,
  "errors": [],
  "attempts": 1,
  "duration_ms": 150,
  "mode": "coerce",
  "limits": {
    "max_output_bytes": 131072,
    "max_str_len": 8192,
    "max_list_len": 2048,
    "max_dict_keys": 512
  },
  "context": { "correlation_id": "abc123" },
  "output_sample": "{\"name\": \"John\", \"age\": 30}"
}

Correlation IDs are automatically generated for each validation attempt and help you:

• 🔍 Track specific validations across logs and error messages
• 🐛 Debug issues by correlating errors with specific validation attempts
• 📊 Monitor performance by tracking validation duration and retry attempts
• 🔗 Link related operations when using retry functions

When viewing logs with agent-validator logs, logs are displayed in a clear table format:

┌─────────────────────────────────────┬────────┬─────────────┬─────────┬──────────┬─────────────┬─────────┬─────────┐
│ Timestamp                           │ Status │ Correlation │ Mode    │ Attempts │ Duration    │ Errors  │ Size    │
├─────────────────────────────────────┼────────┼─────────────┼─────────┼──────────┼─────────────┼─────────┼─────────┤
│ 2025-08-30 18:40:00                 │      ✗ │        none │  strict │        1 │         0ms │       2 │    45B  │
│ 2025-08-30 18:40:00                 │      ✓ │        none │  coerce │        1 │         0ms │       0 │    45B  │
└─────────────────────────────────────┴────────┴─────────────┴─────────┴──────────┴─────────────┴─────────┴─────────┘

Correlation IDs are truncated for readability and show [none] when not set.

☁️ Cloud Logging

Enable cloud logging for monitoring and recovery:

from agent_validator import Config

config = Config(
    log_to_cloud=True,
    license_key="your-license-key",
    cloud_endpoint="https://api.agentvalidator.dev"
)

result = validate(
    agent_output,
    schema,
    log_to_cloud=True,
    config=config
)

🔐 Webhook Secret Management

For enhanced security, generate a webhook secret for HMAC signature validation:

# Generate a new webhook secret
agent-validator webhook --generate

# Check webhook status
agent-validator webhook --status

# Show current webhook secret
agent-validator webhook --show

# Revoke webhook secret
agent-validator webhook --revoke

Security Features:

• ✅ One-time display: Webhook secrets are only shown once when generated
• ✅ Secure storage: Secrets are stored encrypted in the database
• ✅ User isolation: Each license key has its own webhook secret
• ✅ Revocation: Users can revoke and regenerate secrets as needed

Using HMAC Signatures

Once you have a webhook secret, requests can include an HMAC signature:

import hmac
import hashlib

# Create signature
signature = hmac.new(
    webhook_secret.encode(),
    payload.encode(),
    hashlib.sha256
).hexdigest()

# Include in request headers
headers = {
    "license-key": "your-license-key",
    "x-signature": signature
}

Security Notes

• One-time display: Webhook secrets are only shown once when generated
• Secure storage: Secrets are stored encrypted in the database
• User isolation: Each license key has its own webhook secret
• Revocation: Users can revoke and regenerate secrets as needed

🌐 Web Dashboard

Access your validation logs through a secure web dashboard:

# Configure your license key
agent-validator config --set-license-key your-license-key

# Open dashboard with secure local proxy
agent-validator dashboard

This opens a local proxy server at http://localhost:8080 that securely forwards requests to the cloud API with proper authentication headers. The dashboard shows:

• Recent validation attempts
• Success/failure rates
• Performance metrics
• Error details
• Correlation IDs for debugging

Security Features:

• ✅ No credentials in URLs
• ✅ Secure header authentication
• ✅ User data isolation
• ✅ Local proxy prevents credential exposure

---

⚙️ Configuration

🌍 Environment Variables

export AGENT_VALIDATOR_LICENSE_KEY="your-license-key"
export AGENT_VALIDATOR_WEBHOOK_SECRET="your-webhook-secret"
export AGENT_VALIDATOR_LOG_TO_CLOUD="1"
export AGENT_VALIDATOR_ENDPOINT="https://api.agentvalidator.dev"
export AGENT_VALIDATOR_MAX_OUTPUT_BYTES="131072"
export AGENT_VALIDATOR_MAX_STR_LEN="8192"
export AGENT_VALIDATOR_MAX_LIST_LEN="2048"
export AGENT_VALIDATOR_MAX_DICT_KEYS="512"
export AGENT_VALIDATOR_TIMEOUT_S="20"
export AGENT_VALIDATOR_RETRIES="2"

📄 Configuration File

Configuration is stored in ~/.agent_validator/config.toml:

max_output_bytes = 131072
max_str_len = 8192
max_list_len = 2048
max_dict_keys = 512
log_to_cloud = false
cloud_endpoint = "https://api.agentvalidator.dev"
timeout_s = 20
retries = 2
license_key = "your-license-key"
webhook_secret = "your-webhook-secret"

🔄 Configuration Precedence

Configuration values are loaded in the following order (highest to lowest priority):

1. CLI Arguments (highest priority)

   agent-validator test schema.json input.json --mode COERCE --timeout-s 30
   

2. Environment Variables

   export AGENT_VALIDATOR_MODE="COERCE"
   export AGENT_VALIDATOR_TIMEOUT_S="30"
   

3. Configuration File (lowest priority)

   # ~/.agent_validator/config.toml
   mode = "STRICT"
   timeout_s = 20
   

Example: If you have timeout_s = 20 in your config file, but set export AGENT_VALIDATOR_TIMEOUT_S="30", the environment variable (30 seconds) will take precedence.

Note: CLI arguments always override environment variables and config file settings.

---

🔒 Security & Privacy

🚫 Redaction

Sensitive data is automatically redacted before logging:

• API Keys: sk-1234567890abcdef → [REDACTED]
• JWT Tokens: Bearer eyJ... → [REDACTED]
• Emails: john@example.com → j***n@example.com
• Phone Numbers: +1-555-123-4567 → ***-***-4567
• SSNs: 123-45-6789 → ***-**-6789
• Credit Cards: 1234-5678-9012-3456 → ************3456
• Passwords: secret123 → [REDACTED]

🔧 Custom Redaction Patterns

from agent_validator.redact import add_redaction_pattern

# Add custom pattern
add_redaction_pattern("custom_token", r"custom-[a-zA-Z0-9]{20,}")

---

📏 Size Limits

Default limits to prevent abuse:

---

⚠️ Error Handling

🚨 ValidationError

Raised when validation fails after all retries:

from agent_validator import ValidationError

try:
    result = validate(data, schema)
except ValidationError as e:
    print(f"Path: {e.path}")
    print(f"Reason: {e.reason}")
    print(f"Attempt: {e.attempt}")
    print(f"Correlation ID: {e.correlation_id}")

📋 SchemaError

Raised when schema definition is invalid:

from agent_validator import SchemaError

try:
    schema = Schema({"name": bytes})  # Unsupported type
except SchemaError as e:
    print(f"Schema error: {e}")

☁️ CloudLogError

Raised when cloud logging fails (non-fatal):

from agent_validator import CloudLogError

try:
    result = validate(data, schema, log_to_cloud=True)
except CloudLogError as e:
    print(f"Cloud logging failed: {e}")
    # Validation still succeeds

---

💡 Examples

🎯 Basic Example

from agent_validator import validate, Schema, ValidationMode

def call_agent(prompt: str, context: dict) -> str:
    """Mock agent function."""
    import random
    if random.random() < 0.3:
        return "This is not valid JSON"
    return '{"name": "John", "age": 30, "email": "john@example.com"}'

schema = Schema({
    "name": str,
    "age": int,
    "email": str
})

result = validate(
    call_agent("", {}),
    schema,
    retry_fn=call_agent,
    retries=2,
    mode=ValidationMode.COERCE
)

☁️ With Cloud Logging

from agent_validator import validate, Schema, Config

config = Config(
    log_to_cloud=True,
    license_key=os.getenv("AGENT_VALIDATOR_LICENSE_KEY")
)

schema = Schema({
    "user": {
        "name": str,
        "age": int,
        "preferences": {
            "theme": str,
            "notifications": bool
        }
    }
})

result = validate(
    agent_output,
    schema,
    log_to_cloud=True,
    config=config,
    context={"user_id": "123", "environment": "production"}
)

---

🖥️ CLI Reference

🛠️ Commands

# Test validation
agent-validator test <schema.json> <input.json> [--mode STRICT|COERCE]

# View local logs
agent-validator logs [-n <number>] [--clear]

# View cloud logs
agent-validator cloud-logs [-n <number>]

# Open web dashboard
agent-validator dashboard [--port <port>] [--url] [--open]

# Generate correlation ID
agent-validator id

# Manage configuration
agent-validator config [--show] [--show-secrets] [--set-license-key <key>] [--set-endpoint <url>] [--set-webhook-secret <secret>] [--set-log-to-cloud <true|false>]

# Manage webhook secrets
agent-validator webhook [--generate] [--status] [--show] [--revoke] [--force]

📊 Exit Codes

• 0: Success
• 1: General error
• 2: Validation failed

🔒 Security Notes

• Configuration Display: By default, sensitive values (license key, webhook secret) are masked as *** when showing configuration

• Show Secrets: Use --show-secrets flag to display actual values for debugging/verification

• Example:

  # Default (masked)
  agent-validator config --show
  # Output: license_key: ***
  
  # With secrets visible
  agent-validator config --show --show-secrets
  # Output: license_key: my-actual-key
  

---

🛠️ Development

📦 Installation

git clone https://github.com/agent-validator/agent-validator.git
cd agent-validator
pip install -e ".[dev]"

Note: The [dev] extras include build tools (build, twine) needed for creating releases.

🧪 Running Tests

# Run all tests
python -m pytest tests/ -v

# Run with coverage
pytest --cov=agent_validator

# Run property-based tests
python -m pytest tests/property/ -v

# Run type checking
mypy agent_validator cli

# Run linting
ruff check .
black --check .
isort --check-only .

# Run smoke tests (isolated environment)
python smoke_tests/smoke_tests.py

🔧 Pre-commit Hooks

pre-commit install
pre-commit run --all-files

🚀 Smoke Tests

Smoke tests verify the complete user experience in an isolated environment:

# Run comprehensive smoke tests
python smoke_tests/smoke_tests.py

# With backend URL for cloud testing
python smoke_tests/smoke_tests.py --backend-url http://localhost:9090

What gets tested:

• ✅ Package installation in isolated environment
• ✅ CLI command availability and functionality
• ✅ Library imports and basic operations
• ✅ Configuration management
• ✅ Log generation and retrieval
• ✅ Schema validation (strict and coerce modes)
• ✅ Error handling and edge cases

Benefits:

• 🛡️ No pollution to your development environment
• 🔄 Clean testing environment every time
• 🧪 Tests real installation and usage scenarios
• 🚀 Perfect for CI/CD integration

📦 Releasing New Versions

The project uses automated CI/CD for releases. When you push a version tag, the CI automatically builds and publishes to PyPI.

1. Update Version

Update the version in agent_validator/version.py:

__version__ = "1.0.1"  # Increment version number

2. Update CHANGELOG

Add a new entry to CHANGELOG.md following the Keep a Changelog format:

## [1.0.1] - 2025-01-01

### Added

- New feature description

### Changed

- Breaking change description

### Fixed

- Bug fix description

3. Run Pre-release Checks

# Install build dependencies (if not already installed)
pip install -e ".[dev]"

# Run all tests
python -m pytest tests/ -v

# Run smoke tests
python smoke_tests/smoke_tests.py

# Check code quality (matches CI)
ruff check .
mypy agent_validator cli

# Verify package builds
python -m build

4. Commit and Push Changes

# Commit version and changelog updates
git add agent_validator/version.py CHANGELOG.md
git commit -m "Release v1.0.1"
git push origin master

5. Create and Push Git Tag

# Create version tag
git tag v1.0.1

# Push tag to trigger automated release
git push origin v1.0.1

That's it! 🚀 The CI/CD pipeline will automatically:

• ✅ Run tests across Python 3.9, 3.10, 3.11, and 3.12
• ✅ Run linting and type checking
• ✅ Build the package
• ✅ Publish to PyPI (if tag starts with v)

6. Verify Release

After the CI completes (usually 2-3 minutes):

# Test installation from PyPI
pip install agent-validator==1.0.1 --force-reinstall

# Verify functionality
agent-validator --help
python -c "import agent_validator; print(agent_validator.__version__)"

7. Create GitHub Release

• Go to GitHub Releases
• Click "Create a new release"
• Select the tag you just pushed
• Copy the changelog content
• Publish the release

Release Checklist:

• Version updated in agent_validator/version.py
• CHANGELOG.md updated
• All tests pass locally
• Smoke tests pass
• Code quality checks pass
• Package builds successfully
• Changes committed and pushed to master
• Git tag created and pushed
• CI/CD pipeline completes successfully
• Installation from PyPI verified
• GitHub release created with changelog

CI/CD Pipeline Details:

The .github/workflows/ci.yml workflow automatically:

• Tests: Runs on Python 3.9, 3.10, 3.11, 3.12
• Quality: Runs ruff check and mypy
• Coverage: Generates and uploads coverage reports
• Publishing: Automatically publishes to PyPI when a tag starting with v is pushed
• Security: Uses GitHub secrets for PyPI authentication

---

🤝 Contributing

1. 🍴 Fork the repository
2. 🌿 Create a feature branch
3. ✏️ Make your changes
4. 🧪 Add tests
5. ✅ Run the test suite
6. 📤 Submit a pull request

---

📄 License

MIT License - see LICENSE file for details.

---

❓ FAQ

Q: Can I use this with any LLM/agent?
A: Yes! The library is framework-agnostic. Just pass your agent's output to the validate function.

Q: What happens if my agent returns malformed JSON?
A: In strict mode, it will fail immediately. In coerce mode, it will try to parse as JSON first, then fall back to treating it as a plain string.

Q: How do I handle sensitive data in logs?
A: Sensitive data is automatically redacted before logging. You can also add custom redaction patterns.

Q: Can I use this in production?
A: Yes! The library is designed for production use with proper error handling, logging, and monitoring capabilities.

Q: How do I access the web dashboard securely?
A: Use agent-validator dashboard which creates a secure local proxy server. Never put your license key in URLs - the CLI handles authentication securely via headers.

Q: How do I verify my license key is set correctly?
A: Use agent-validator config --show --show-secrets to display the actual license key value. By default, sensitive values are masked as *** for security.

Q: What's the performance impact?
A: Minimal. Validation is fast, and logging is asynchronous. The main overhead comes from retry attempts when validation fails.

Q: Can I use my own schema format?
A: Currently only Python dict schemas are supported. JSONSchema support is planned for v0.1.

---

🗺️ Roadmap

• JSONSchema import/export
• Pydantic model support
• Custom validators per field
• Schema composition and inheritance
• Web dashboard for monitoring
• Alerting and notifications
• Schema versioning
• Performance metrics

---