Delegation MCP

"""System instructions for the orchestrator agent.""" from .task_mapper import TASK_CATEGORIES def get_system_instructions( orchestrator_name: str = "Claude", task_mappings: dict[str, str] | None = None, selected_agents: list[str] | None = None, ) -> str: """ Generate ultra-concise system instructions for orchestrator. Args: orchestrator_name: The name of the orchestrator agent (e.g., "Claude", "Gemini"). task_mappings: Dictionary mapping task categories to agent names selected_agents: List of selected agent names Returns: Formatted system instructions string (optimized for minimal tokens). """ orchestrator_name = orchestrator_name.capitalize() task_mappings = task_mappings or {} # Build concise routing table routing_rows = [] category_map = {cat["key"]: cat for cat in TASK_CATEGORIES} for task_key, agent_name in task_mappings.items(): category = category_map.get(task_key) if category and task_key != "general": routing_rows.append(f"- {category['description']} → **{agent_name}**") routing_table = "\n".join(routing_rows) if routing_rows else "No specific mappings configured" instructions = f"""# {orchestrator_name} - Orchestrator ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ ⚠️ STOP - READ BEFORE ANY CODE/FILE/TECHNICAL OPERATION ⚠️ ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Before reading files, editing code, analyzing systems, or ANY technical work: ❓ Did you call `delegate_task(query="...", guidance_only=True)` for this request? □ YES - I called it. Server returned: ________________ □ NO - I have NOT called it yet If you checked NO or left blank: **STOP NOW. Go back. Call the tool.** If you checked YES: What did the server return? □ "claude" → OK to proceed (you handle it) □ "[agent_name]" (gemini/aider/copilot) → MUST delegate, do NOT proceed **VIOLATION CHECK**: If you are reading code/files WITHOUT having called the tool above, you are violating the protocol. STOP IMMEDIATELY. **MULTI-STEP TASKS**: If request has multiple distinct phases (e.g., "review + fix + commit"), call delegate_task separately for each: 1. delegate_task("phase 1") → get results 2. delegate_task("phase 2: [specific from results]") → execute 3. delegate_task("phase 3") → finalize Example: "Review backend, fix issues, commit changes" → Call delegate_task("review backend for scalability") → Gemini analyzes → Call delegate_task("fix critical query issues in api.ts") → Aider implements fixes → Call delegate_task("stage, commit, push with message") → Aider handles git ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ **Core Function**: Delegate specialized tasks, handle architecture/planning yourself. ## Delegation Protocol **For ANY code/technical request**: 1. Call `delegate_task(query=full_user_request, guidance_only=True)` FIRST 2. Tool returns agent name: "gemini" / "aider" / "copilot" (delegate) or "claude" (you handle) 3. Follow tool response (do NOT override it) **Why**: Server has routing logic (keywords + capability scoring). Always consult it. ## Routing Table {routing_table} ## Tools - `delegate_task(query, guidance_only?, force_delegate?)` - Get routing guidance or execute delegation - `list_orchestrators()` - Show available agents - `discover_agents()` - Find new agents ## Rules 1. **ALWAYS call tool first**: `delegate_task(guidance_only=True)` before ANY code work 2. **Trust server routing**: Uses keyword matching + capability scoring 3. **Handle directly ONLY if**: Tool explicitly returns "HANDLE_DIRECTLY" 4. **Never bypass**: Even if task seems "simple" or "quick" - ALWAYS consult tool **Critical**: If you catch yourself thinking "this is quick, I'll just..." → STOP. Call the tool. """ return instructions def _build_routing_table(task_mappings: dict[str, str]) -> list[str]: """Build routing table rows from task mappings.""" rows = [] category_map = {cat["key"]: cat for cat in TASK_CATEGORIES} for task_key, agent_name in task_mappings.items(): category = category_map.get(task_key) if not category or task_key == "general": continue # Format examples for table examples = ", ".join(category["pattern_examples"][:3]) # First 3 examples row = f"| {category['description']} | **{agent_name}** | {examples} |" rows.append(row) return rows def _build_agent_summary(selected_agents: list[str]) -> list[str]: """Build agent capabilities summary. Note: This is intentionally kept simple since actual capabilities are shown in the routing table based on user's task mappings. """ if not selected_agents: return [] summary = [] for agent in selected_agents: summary.append(f"- **{agent}**") return summary def _build_prohibitions(task_mappings: dict[str, str]) -> list[str]: """Build prohibition list dynamically from task mappings. Only prohibit tasks that the user has explicitly mapped to other agents. """ if not task_mappings: return [] prohibitions = [] category_map = {cat["key"]: cat for cat in TASK_CATEGORIES} # Map task types to prohibition descriptions prohibition_templates = { "git_operations": "❌ Execute git commands (git checkout, git pull, git commit, etc.)", "shell_tasks": "❌ Run shell/terminal commands directly", "refactoring": "❌ Perform code refactoring directly", "security_audit": "❌ Conduct security reviews or audits", "testing": "❌ Write or execute tests directly", "performance": "❌ Perform performance analysis or optimization", "browser_interaction": "❌ Execute browser automation tasks", "code_review": "❌ Conduct code reviews", "exploration": "❌ Explore codebase or trace implementations (delegate to save tokens)", "debugging": "❌ Debug issues or investigate errors (delegate to save tokens)", "impact_analysis": "❌ Analyze dependencies or find usages (delegate to save tokens)", } for task_key, agent_name in task_mappings.items(): if task_key == "general": continue # Get prohibition template for this task type prohibition = prohibition_templates.get(task_key) if prohibition: prohibitions.append(f"- {prohibition} → **Delegate to {agent_name}**") return prohibitions def _build_routing_examples(task_mappings: dict[str, str]) -> list[str]: """Build routing examples dynamically from actual task mappings.""" examples = [] category_map = {cat["key"]: cat for cat in TASK_CATEGORIES} # Build examples from actual mappings for task_key, agent_name in task_mappings.items(): category = category_map.get(task_key) if not category or task_key == "general": continue # Use first pattern example from the category if category["pattern_examples"]: example_query = category["pattern_examples"][0] examples.append(f'| "{example_query}" | DELEGATE | {agent_name} |') # Add a general handling example examples.append('| "What\'s the best database for this use case?" | HANDLE | (yourself) |') return examples