Task Planner and Validator - Skill Guide
This skill provides a secure, step-by-step task management system for AI Agents.
Quick Installation
# 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
# Run tests
python test_basic.py
# Run examples
python examples.py
Basic Usage
1. Import and Initialize
from task_planner import TaskPlanner
# Create planner
planner = TaskPlanner(auto_approve=False)
2. Define Your Executor
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
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
# 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:
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:
success, results = planner.execute_plan(
plan,
my_executor,
dry_run=True # Simulate only
)
Save and Load Plans
Persist plans for reuse:
# 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:
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:
{
"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
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
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
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
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
# 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
{
"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
# 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
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
# Skip manual approval for automated workflows
planner = TaskPlanner(auto_approve=True)
Checkpoint System
# Checkpoints are automatically created for rollback-capable steps
# Access checkpoint history
checkpoints = planner.executor.checkpoint_stack
Execution History
# View execution history
history = planner.executor.execution_history
for entry in history:
print(f"{entry['timestamp']}: {entry['step_id']} - {entry['status']}")
Custom Validation Rules
# 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"
# 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
# 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
# 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
# Run basic tests
python test_basic.py
# Run examples
python examples.py
# Both should show "✅ ALL TESTS PASSED"
Getting Help
License
MIT License - see LICENSE file
Author
cerbug45
⭐ If you find this useful, star the repository on GitHub!
