A comprehensive Python framework for orchestrating multiple specialized AI agents to automate complex tasks. Break down goals, spawn specialist agents, and organize all work in clean, date-based folders for personal research, content creation, and development.
- Multi-Agent Coordination: Seamlessly coordinate 6 specialized agent types
- Intelligent Task Planning: Automatically breaks down goals into logical workflow steps
- Workspace Isolation: Each task runs in its own organized workspace with proper file isolation
- Human-in-the-Loop: Optional user approvals at critical steps
- Comprehensive Logging: Complete workflow history with detailed execution tracking
- Production Ready: Robust async support, error handling, and configuration management
- Extensible: Easy to add new agent types and custom tools
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β AI Orchestrator β
β β
β Breaks down goals β Plans workflow β Spawns agents β
β βββββββββββββββββββ βββββββββββββββββββββββββββββββ β
β β Workspace Mgr β β Task Planner β β
β β β β β β
β β β’ Date-based β β β’ Analyze goals β β
β β folders β β β’ Select agents β β
β β β’ Context mgmt β β β’ Plan steps β β
β βββββββββββββββββββ βββββββββββββββββββββββββββββββ β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β
βΌ
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
β Specialist Agents β
β β
β π¬ Researcher β π» Coder β βοΈ Writer β π Analyst β
β β β β β
β β’ Web search β β’ Code β β’ Content β β’ Data β
β β’ Synthesize β gen β β’ Docs β analysisβ
β β’ Fact-find β β’ Testing β β’ Editing β β’ Trends β
β β β β β
β ποΈ Designer β π QA Tester β
β β β
β β’ Architecture β β’ Validation β
β β’ Systems plan β β’ Quality check β
β β’ Blueprints β β’ Testing β
βββββββββββββββββββββββββββββββββββββββββββββββββββββββββββ
- Python 3.9 or higher
- Claude Agent SDK (requires API key)
- asyncio compatible environment
# Clone the repository git clone https://github.com/your-org/ai-orchestrator-sdk.git cd ai-orchestrator-sdk # Install dependencies pip install -r requirements.txt # Install the SDK in development mode pip install -e .
# Core dependencies pip install claude-agent-sdk aiofiles # Optional dependencies for enhanced functionality pip install pandas matplotlib seaborn # For data analysis pip install requests beautifulsoup4 # For enhanced web scraping pip install jinja2 # For template rendering
# Basic goal execution python main.py "Research AI trends for my podcast" # Development task python main.py "Build a Python script to analyze sales data" # Writing task python main.py "Write documentation for my REST API" # List recent workspaces python main.py --list # Non-interactive mode (no prompts) python main.py --non-interactive "Generate weekly report" # Resume an existing workspace python main.py --resume workspaces/week-47/2024-11-23/my-task # Clean up old workspaces python main.py --cleanup 30
from src.orchestrator import Orchestrator # Create orchestrator instance orchestrator = Orchestrator( base_dir="workspaces", # Base directory for workspaces interactive=True # Prompt for user approvals ) # Execute a goal result = await orchestrator.execute_goal( "Create a data analysis dashboard for sales metrics" ) # Check results print(f"β Completed {len(result['results'])} steps") print(f"π Workspace: {result['workspace']}") print(f"π Success rate: {result['summary']['success_rate']}")
Each task creates an organized workspace:
workspaces/
βββ week-47/
βββ 2024εΉ΄11ζ23ζ₯/
βββ analyze-sales-data-143022/
βββ context-main.md # Main context and progress
βββ workflow-history.json # Detailed execution log
βββ workflow-state.json # Current workflow state
βββ workspace-info.json # Workspace metadata
βββ execution-summary.md # Final results summary
βββ agent_outputs/ # Individual agent logs
β βββ researcher_143025.log
β βββ coder_143042.log
β βββ analyst_143058.log
βββ research_notes.md # Research findings
βββ architecture_design.md # System architecture
βββ implementation.md # Generated code
βββ analysis_report.md # Final analysis
- Purpose: Gather and synthesize information
- Tools: Web search, file reading, markdown writing
- Use Cases: Market research, fact-finding, trend analysis
- Max Turns: 8
# Spawn researcher directly from src.agent_manager import AgentManager, AgentType agent_manager = AgentManager(workspace_path) result = await agent_manager.spawn_agent( agent_type=AgentType.RESEARCHER, task="Research Python web scraping libraries", output_file="scraping_research.md" )
- Purpose: Generate and implement code
- Tools: File operations, code generation, testing
- Use Cases: Scripts, APIs, utilities, automation
- Max Turns: 6
- Purpose: Create and polish content
- Tools: Markdown writing, editing, formatting
- Use Cases: Documentation, articles, scripts, reports
- Max Turns: 6
- Purpose: Analyze data and identify patterns
- Tools: Data analysis, statistics, trend identification
- Use Cases: Business insights, data trends, metrics
- Max Turns: 7
- Purpose: Design architectures and systems
- Tools: Architecture planning, system design
- Use Cases: System architecture, workflow design
- Max Turns: 5
- Purpose: Validate and test deliverables
- Tools: Testing, validation, quality assurance
- Use Cases: Code testing, document review
- Max Turns: 5
Set up your environment:
# Required: Claude API Key export CLAUDE_API_KEY="your-api-key-here" # Optional: Custom workspace directory export ORCHESTRATOR_WORKSPACE__BASE_DIR="my-workspaces" # Optional: Default agent settings export ORCHESTRATOR_AGENT__DEFAULT_MAX_TURNS="10" export ORCHESTRATOR_AGENT__ENABLE_LOGGING="true" # Optional: Security settings export ORCHESTRATOR_SECURITY__SANDBOX_MODE="true" export ORCHESTRATOR_SECURITY__RESTRICT_FILE_ACCESS="true"
Create config/orchestrator.json:
{
"workspace": {
"base_dir": "workspaces",
"max_workspace_age_days": 30,
"auto_cleanup": false
},
"agent": {
"default_max_turns": 6,
"default_timeout": 300,
"cost_tracking": true,
"enable_logging": true
},
"security": {
"allow_external_apis": true,
"restrict_file_access": true,
"allowed_file_extensions": [
".md", ".txt", ".json", ".csv", ".py", ".js",
".html", ".css", ".yaml", ".yml"
],
"sandbox_mode": false
}
}# Research AI trends for podcast result = await orchestrator.execute_goal( "Research the latest AI developments for this week's tech podcast" ) # Output files created: # - research_notes.md # Initial research findings # - research_synthesis.md # Analyzed and synthesized data # - podcast_script.md # Draft podcast episode # - show_notes.md # Supporting materials
# Build complete data analysis solution result = await orchestrator.execute_goal( "Create a Python dashboard to analyze monthly sales data with charts" ) # Workflow executed: # 1. Designer: Plans dashboard architecture # 2. Researcher: Finds best visualization libraries # 3. Coder: Implements dashboard script # 4. QA Tester: Validates and tests
# Generate comprehensive API docs result = await orchestrator.execute_goal( "Write complete documentation for my REST API including examples" ) # Custom workflow with specific instructions: await orchestrator.agent_manager.spawn_agent( agent_type=AgentType.RESEARCHER, task="Analyze the API endpoints and understand functionality", custom_instructions="Focus on user perspective and use cases" )
Run the test suite:
# Run all tests python -m pytest tests/ -v # Run specific test python -m pytest tests/test_orchestrator.py -v # Run with coverage python -m pytest tests/ --cov=src --cov-report=html
- β Workspace Management
- β Agent Spawning and Coordination
- β Task Planning
- β File Operations
- β Configuration Loading
- β Error Handling
The SDK includes several security features:
- Workspace Isolation: Agents can only access files within their workspace
- File Type Restrictions: Configurable allowed file extensions
- Sandbox Mode: Optional restricted execution environment
- No Code Execution: Agents generate code but don't execute it directly
- User Approvals: Optional prompts before critical operations
- Control Costs: Limit agent turns and parallel executions
- Reuse Workspaces: Resume existing workspaces when possible
- Batch Operations: Group related tasks in single goals
- Monitor Logs: Check workflow-history.json for inefficiencies
- Clean Up: Regularly remove old workspaces
We welcome contributions! See CONTRIBUTING.md for guidelines.
# Clone and install git clone https://github.com/your-org/ai-orchestrator-sdk.git cd ai-orchestrator-sdk pip install -e ".[dev]" # Run tests python -m pytest # Run linting flake8 src/ black src/
-
"Agent failed to spawn"
- Check API key configuration
- Verify network connectivity
- Review agent permissions
-
"Workspace creation failed"
- Check directory permissions
- Verify disk space
- Check base directory exists
-
"Context not updating"
- Check file write permissions
- Verify workspace integrity
- Review workflow history for errors
# Enable verbose logging import logging logging.basicConfig(level=logging.DEBUG) # Or use CLI flag python main.py --verbose "Your task here"
This project is licensed under the MIT License - see LICENSE file for details.
- Claude Agent SDK for agent management
- OpenAI for powerful language models
- The Python async community for excellent async support
- π§ Email: support@ai-orchestrator.com
- π¬ Discord: Join our community
- π Docs: Full documentation
- π Issues: GitHub Issues