Spaces:
Sleeping
Sleeping
| """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 | |