Delegation MCP

"""Configuration tab for Gradio UI - manage agents and routing rules.""" try: import gradio as gr GRADIO_AVAILABLE = True except ImportError: GRADIO_AVAILABLE = False # Mock gr for type hinting if needed, or just handle availability check gr = None # type: ignore from pathlib import Path from .config_manager import ConfigurationManager, AgentStatus def create_config_tab(config_manager: ConfigurationManager | None = None) -> gr.Blocks | None: """Create configuration tab for runtime agent and routing management. Args: config_manager: Optional ConfigurationManager instance (creates new if None) Returns: Gradio Blocks interface for configuration """ if not GRADIO_AVAILABLE: print("Error: Gradio is not installed. Please install with `pip install .[ui]`") return None if config_manager is None: config_manager = ConfigurationManager() with gr.Blocks() as config_tab: gr.Markdown(""" # ⚙️ Agent Configuration Configure which agents are active and how tasks are routed between them. Changes take effect immediately after saving. """) # Status message for feedback status_msg = gr.Markdown("", visible=False) # Main layout with gr.Row(): # Left column: Agent Management with gr.Column(scale=1): gr.Markdown("## 🤖 Available Agents") # Primary orchestrator selection agent_statuses = config_manager.get_agent_statuses() agent_names = [status.name for status in agent_statuses if status.installed and status.enabled] primary_dropdown = gr.Dropdown( choices=agent_names, value=config_manager.primary_orchestrator, label="Primary Orchestrator", info="Default agent when no routing rules match", interactive=True, ) gr.Markdown("### Agent Status") # Create agent toggles and status displays agent_components = {} for status in agent_statuses: with gr.Row(): with gr.Column(scale=2): # Agent name and status gr.Markdown(f"**{status.status_icon} {status.name}**") gr.Markdown(f"*{status.status_text}*") with gr.Column(scale=1): # Toggle switch toggle = gr.Checkbox( value=status.enabled, label="Enabled", interactive=status.installed, # Only allow toggle if installed elem_id=f"agent_{status.name}_toggle", ) agent_components[status.name] = toggle with gr.Column(scale=1): # Capabilities info button with gr.Accordion(f"ℹ️", open=False): caps = config_manager.get_agent_capabilities(status.name) gr.Markdown(f""" **Command:** `{caps.get('command', 'N/A')}` **Args:** {' '.join(caps.get('args', []))} **Timeout:** {caps.get('timeout', 300)}s **Max Retries:** {caps.get('max_retries', 3)} """) # Right column: Routing Rules with gr.Column(scale=2): gr.Markdown("## 🔀 Routing Rules") # Rules editor rules_yaml = config_manager.get_rules_yaml() rules_editor = gr.TextArea( value=rules_yaml, label="Routing Rules (YAML)", placeholder="Enter routing rules in YAML format...", lines=15, max_lines=25, ) # Validation message validation_msg = gr.Markdown("", visible=False) # Buttons row with gr.Row(): validate_btn = gr.Button("✓ Validate Rules", variant="secondary") preview_btn = gr.Button("👁️ Preview", variant="secondary") # Preview panel with gr.Accordion("📋 Routing Preview", open=False) as preview_accordion: preview_text = gr.Markdown("") # Bottom buttons with gr.Row(): save_btn = gr.Button("💾 Save Configuration", variant="primary", size="lg") reset_btn = gr.Button("🔄 Reset to Defaults", variant="stop") # === Event Handlers === def update_agent_toggle(agent_name: str, enabled: bool): """Handle agent toggle.""" success, message = config_manager.toggle_agent(agent_name, enabled) if not success: # Revert the toggle return { status_msg: gr.Markdown( f"❌ **Error:** {message}", visible=True ), agent_components[agent_name]: gr.Checkbox(value=not enabled), } # Update primary dropdown choices if needed agent_statuses = config_manager.get_agent_statuses() active_agents = [s.name for s in agent_statuses if s.installed and s.enabled] return { status_msg: gr.Markdown( f"✅ {message}", visible=True ), primary_dropdown: gr.Dropdown(choices=active_agents), } def set_primary_orchestrator(agent_name: str): """Handle primary orchestrator selection.""" success, message = config_manager.set_primary_orchestrator(agent_name) if not success: return { status_msg: gr.Markdown( f"❌ **Error:** {message}", visible=True ), primary_dropdown: gr.Dropdown(value=config_manager.primary_orchestrator), } return status_msg.update( value=f"✅ {message}", visible=True ) def validate_rules(yaml_text: str): """Validate routing rules.""" is_valid, message, _ = config_manager.validate_routing_rules(yaml_text) if is_valid: return validation_msg.update( value=f"✅ {message}", visible=True ) else: return validation_msg.update( value=f"❌ **Validation Error:** {message}", visible=True ) def preview_rules(yaml_text: str): """Generate routing rules preview.""" is_valid, message, parsed_rules = config_manager.validate_routing_rules(yaml_text) if not is_valid: return { preview_text: gr.Markdown(f"❌ **Cannot preview:** {message}"), preview_accordion: gr.Accordion(open=True), } preview = config_manager.preview_routing_rules(parsed_rules or []) return { preview_text: gr.Markdown(preview), preview_accordion: gr.Accordion(open=True), } def save_configuration(yaml_text: str): """Save all configuration changes.""" # Validate rules first is_valid, message, _ = config_manager.validate_routing_rules(yaml_text) if not is_valid: return status_msg.update( value=f"❌ **Cannot save:** {message}", visible=True ) # Save configuration success, message = config_manager.save_configurations(rules_yaml=yaml_text) if success: return status_msg.update( value=f"✅ **Configuration saved successfully!** Changes are now active.", visible=True ) else: return status_msg.update( value=f"❌ **Save failed:** {message}", visible=True ) def reset_configuration(): """Reset configuration to defaults.""" success, message = config_manager.reset_to_defaults() if not success: return { status_msg: gr.Markdown( f"❌ **Reset failed:** {message}", visible=True ), } # Reload UI with defaults agent_statuses = config_manager.get_agent_statuses() active_agents = [s.name for s in agent_statuses if s.installed and s.enabled] rules_yaml = config_manager.get_rules_yaml() # Build update dictionary updates = { status_msg: gr.Markdown( f"✅ {message}", visible=True ), primary_dropdown: gr.Dropdown( value=config_manager.primary_orchestrator, choices=active_agents, ), rules_editor: gr.TextArea(value=rules_yaml), validation_msg: gr.Markdown(visible=False), preview_text: gr.Markdown(""), } # Update agent toggles for status in agent_statuses: updates[agent_components[status.name]] = gr.Checkbox(value=status.enabled) return updates # Wire up event handlers primary_dropdown.change( fn=set_primary_orchestrator, inputs=[primary_dropdown], outputs=[status_msg], ) # Wire up agent toggles for agent_name, toggle in agent_components.items(): toggle.change( fn=lambda enabled, name=agent_name: update_agent_toggle(name, enabled), inputs=[toggle], outputs=[status_msg, agent_components[agent_name], primary_dropdown], ) validate_btn.click( fn=validate_rules, inputs=[rules_editor], outputs=[validation_msg], ) preview_btn.click( fn=preview_rules, inputs=[rules_editor], outputs=[preview_text, preview_accordion], ) save_btn.click( fn=save_configuration, inputs=[rules_editor], outputs=[status_msg], ) # Reset button all_outputs = [ status_msg, primary_dropdown, rules_editor, validation_msg, preview_text, ] + list(agent_components.values()) reset_btn.click( fn=reset_configuration, outputs=all_outputs, ) # Add helpful tooltips gr.Markdown(""" --- ### 💡 Tips - **Primary Orchestrator**: Handles all tasks unless a routing rule matches - **Routing Rules**: Use regex patterns to delegate specific tasks to appropriate agents - **Pattern Examples**: - `security|audit|vulnerability` - Security-related tasks - `refactor|redesign` - Code refactoring - `test|pytest|jest` - Testing tasks - **Priority**: Higher priority rules are evaluated first (0-10) - **Agent Status**: - 🟢 Active - Enabled and installed - 🔴 Disabled - Toggled off - ⚠️ Not Installed - Command not found in PATH Changes take effect immediately. See [GitHub](https://github.com/carlosduplar/multi-agent-mcp) for docs. """) return config_tab