What this skill does

# 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!