| name | task-panner-validator |
| description | Task Panner Validator for Agents |
Task Planner and Validator - Skill Guide
This skill provides a secure, step-by-step task management system for AI Agents.
Quick Installation
git clone https://github.com/cerbug45/task-planner-validator.git
cd task-planner-validator
Verify Installation
python test_basic.py
python examples.py
Basic Usage
1. Import and Initialize
from task_planner import TaskPlanner
planner = TaskPlanner(auto_approve=False)
2. Define Your Executor
def my_executor(action: str, parameters: dict):
"""Your custom execution logic"""
if action == "fetch_data":
return {"data": [1, 2, 3]}
elif action == "process_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
is_valid, warnings = planner.validate_plan(plan)
if warnings:
print("Warnings:", warnings)
planner.approve_plan(plan, approved_by="admin")
success, results = planner.execute_plan(plan, my_executor)
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",
"parameters": {"path": "/data/old"},
"safety_check": True,
"rollback_possible": False
}
]
Dry Run Mode
Test without executing:
success, results = planner.execute_plan(
plan,
my_executor,
dry_run=True
)
Save and Load Plans
Persist plans for reuse:
planner.save_plan(plan, "my_plan.json")
loaded_plan = planner.load_plan("my_plan.json")
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
)
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,
"action": str,
"parameters": dict,
"expected_output": str,
"safety_check": bool,
"rollback_possible": bool,
"max_retries": int
}
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
{
"description": "Fetch active users from PostgreSQL production database",
"action": "fetch_active_users_postgres_prod",
...
}
{
"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,
"rollback_possible": False
}
4. Test with Dry Run
success, results = planner.execute_plan(plan, my_executor, dry_run=True)
if success:
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
Advanced Features
Auto-Approve for Automation
planner = TaskPlanner(auto_approve=True)
Checkpoint System
checkpoints = planner.executor.checkpoint_stack
Execution History
history = planner.executor.execution_history
for entry in history:
print(f"{entry['timestamp']}: {entry['step_id']} - {entry['status']}")
Custom Validation Rules
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"
planner.approve_plan(plan, approved_by="admin")
planner = TaskPlanner(auto_approve=True)
Safety validation warnings
is_valid, warnings = planner.validate_plan(plan)
for warning in warnings:
print(warning)
if is_valid:
planner.approve_plan(plan)
Steps executing out of order
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
python test_basic.py
python examples.py
Getting Help
License
MIT License - see LICENSE file
Author
cerbug45
⭐ If you find this useful, star the repository on GitHub!