back
loading skill details...
Provides secure task planning, validation, approval, and execution for AI agents with safety checks, rollback, dry runs, and error handling using pure Python.
# Task Planner and Validator - Skill Guide
This skill provides a secure, step-by-step task management system for AI Agents.
## Quick Installation
```bash
# Clone the repository
git clone https://github.com/cerbug45/task-planner-validator.git
cd task-planner-validator
# That's it! No dependencies needed - pure Python standard library
```
## Verify Installation
```bash
# Run tests
python test_basic.py
# Run examples
python examples.py
```
## Basic Usage
### 1. Import and Initialize
```python
from task_planner import TaskPlanner
# Create planner
planner = TaskPlanner(auto_approve=False)
```
### 2. Define Your Executor
```python
def my_executor(action: str, parameters: dict):
"""Your custom execution logic"""
if action == "fetch_data":
# Fetch data from API, database, etc.
return {"data": [1, 2, 3]}
elif action == "process_data":
# Process the data
return {"processed": True}
else:
return {"status": "completed"}
```
### 3. Create a Plan
```python
steps = [
{
"description": "Fetch user data",
"action": "fetch_data",
"parameters": {"source": "database"},
"expected_output": "List of users"
},
{
"description": "Process users",
"action": "process_data",
"parameters": {"validation": True},
"expected_output": "Processed data"
}
]
plan = planner.create_plan(
title="Data Processing Pipeline",
description="Fetch and process user data",
steps=steps
)
```
### 4. Validate and Execute
```python
# Validate
is_valid, warnings = planner.validate_plan(plan)
if warnings:
print("Warnings:", warnings)
# Approve
planner.approve_plan(plan, approved_by="admin")
# Execute
success, results = planner.execute_plan(plan, my_executor)
# Get summary
summary = planner.get_execution_summary(plan)
print(f"Progress: {summary['progress_percentage']}%")
```
## Key Features
### Safety Validation
Automatically detects dangerous operations:
```python
steps = [
{
"description": "Delete old files",
"action": "delete_files", # ⚠️ Dangerous!
"parameters": {"path": "/data/old"},
"safety_check": True, # System will warn
"rollback_possible": False # Cannot undo
}
]
```
### Dry Run Mode
Test without executing:
```python
success, results = planner.execute_plan(
plan,
my_executor,
dry_run=True # Simulate only
)
```
### Save and Load Plans
Persist plans for reuse:
```python
# Save
planner.save_plan(plan, "my_plan.json")
# Load later
loaded_plan = planner.load_plan("my_plan.json")
# Verify integrity
if loaded_plan.verify_integrity():
planner.execute_plan(loaded_plan, my_executor)
```
### Error Handling
Control error behavior:
```python
success, results = planner.execute_plan(
plan,
my_executor,
stop_on_error=False # Continue on failures
)
# Check results
for result in results:
if not result['success']:
print(f"Step {result['order']} failed: {result['error']}")
```
## Step Configuration
Each step supports these parameters:
```python
{
"description": str, # Required: Human-readable description
"action": str, # Required: Action identifier
"parameters": dict, # Required: Action parameters
"expected_output": str, # Required: Expected result
"safety_check": bool, # Optional: Enable validation (default: True)
"rollback_possible": bool, # Optional: Can be rolled back (default: True)
"max_retries": int # Optional: Retry attempts (default: 3)
}
```
## Common Use Cases
### API Orchestration
```python
steps = [
{
"description": "Authenticate",
"action": "api_auth",
"parameters": {"service": "github"},
"expected_output": "Auth token"
},
{
"description": "Fetch data",
"action": "api_fetch",
"parameters": {"endpoint": "/repos"},
"expected_output": "Repository list"
}
]
```
### Data Pipeline
```python
steps = [
{
"description": "Extract data",
"action": "extract",
"parameters": {"source": "database"},
"expected_output": "Raw data"
},
{
"description": "Transform data",
"action": "transform",
"parameters": {"rules": ["normalize", "validate"]},
"expected_output": "Clean data"
},
{
"description": "Load data",
"action": "load",
"parameters": {"destination": "warehouse"},
"expected_output": "Success confirmation"
}
]
```
### System Automation
```python
steps = [
{
"description": "Backup database",
"action": "backup",
"parameters": {"target": "postgres"},
"expected_output": "Backup file path",
"rollback_possible": True
},
{
"description": "Update schema",
"action": "migrate",
"parameters": {"version": "2.0"},
"expected_output": "Migration complete",
"rollback_possible": True
},
{
"description": "Verify integrity",
"action": "verify",
"parameters": {"checks": ["all"]},
"expected_output": "All checks passed"
}
]
```
## Best Practices
### 1. Always Validate First
```python
is_valid, warnings = planner.validate_plan(plan)
if not is_valid:
print("Plan validation failed!")
for warning in warnings:
print(f" - {warning}")
exit(1)
```
### 2. Use Descriptive Names
```python
# Good ✅
{
"description": "Fetch active users from PostgreSQL production database",
"action": "fetch_active_users_postgres_prod",
...
}
# Bad ❌
{
"description": "Get data",
"action": "get",
...
}
```
### 3. Mark Dangerous Operations
```python
{
"description": "Delete temporary files older than 30 days",
"action": "cleanup_temp_files",
"parameters": {"age_days": 30, "path": "/tmp"},
"safety_check": True, # ⚠️ Will trigger warnings
"rollback_possible": False # ⚠️ Cannot undo!
}
```
### 4. Test with Dry Run
```python
# Always test first
success, results = planner.execute_plan(plan, my_executor, dry_run=True)
if success:
# Now run for real
success, results = planner.execute_plan(plan, my_executor, dry_run=False)
```
### 5. Handle Errors Gracefully
```python
def safe_executor(action: str, parameters: dict):
try:
result = execute_action(action, parameters)
return result
except Exception as e:
logging.error(f"Failed to execute {action}: {e}")
raise # Re-raise to let planner handle it
```
## Advanced Features
### Auto-Approve for Automation
```python
# Skip manual approval for automated workflows
planner = TaskPlanner(auto_approve=True)
```
### Checkpoint System
```python
# Checkpoints are automatically created for rollback-capable steps
# Access checkpoint history
checkpoints = planner.executor.checkpoint_stack
```
### Execution History
```python
# View execution history
history = planner.executor.execution_history
for entry in history:
print(f"{entry['timestamp']}: {entry['step_id']} - {entry['status']}")
```
### Custom Validation Rules
```python
# Add custom validation to SafetyValidator
planner.safety_validator.dangerous_operations.append('my_dangerous_op')
planner.safety_validator.sensitive_paths.append('/my/sensitive/path')
```
## Troubleshooting
### "Plan must be approved before execution"
```python
# Solution: Approve the plan first
planner.approve_plan(plan, approved_by="admin")
# Or use auto-approve mode
planner = TaskPlanner(auto_approve=True)
```
### Safety validation warnings
```python
# Review warnings and ensure operations are intentional
is_valid, warnings = planner.validate_plan(plan)
for warning in warnings:
print(warning)
# If operations are safe, approve anyway
if is_valid: # Still valid, just warnings
planner.approve_plan(plan)
```
### Steps executing out of order
```python
# Ensure order values are sequential
steps[0]['order'] = 1
steps[1]['order'] = 2
steps[2]['order'] = 3
```
## File Structure
```
task-planner-validator/
├── task_planner.py # Main library
├── examples.py # Usage examples
├── test_basic.py # Test suite
├── README.md # Full documentation
├── QUICKSTART.md # Quick start guide
├── API.md # API reference
├── SKILL.md # This file
└── LICENSE # MIT License
```
## Requirements
- Python 3.8 or higher
- No external dependencies!
## Testing
```bash
# Run basic tests
python test_basic.py
# Run examples
python examples.py
# Both should show "✅ ALL TESTS PASSED"
```
## Getting Help
- 📖 Read full documentation in [README.md](README.md)
- 🚀 Check [QUICKSTART.md](QUICKSTART.md) for quick examples
- 📚 See [API.md](API.md) for complete API reference
- 💡 Browse [examples.py](examples.py) for real code
- 🐛 Report issues on GitHub
## License
MIT License - see [LICENSE](LICENSE) file
## Author
**cerbug45**
- GitHub: [@cerbug45](https://github.com/cerbug45)
---
⭐ If you find this useful, star the repository on GitHub!
don't have the plugin yet? install it then click "run inline in claude" again.