Spaces:
Sleeping
Sleeping
File size: 12,183 Bytes
8b02e7c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 236 237 238 239 240 241 242 243 244 245 246 247 248 249 250 251 252 253 254 255 256 257 258 259 260 261 262 263 264 265 266 267 268 269 270 271 272 273 274 275 276 277 278 279 280 281 282 283 284 285 286 287 288 289 290 291 292 293 294 295 296 297 298 299 300 301 302 303 304 305 306 307 308 309 310 311 312 313 314 315 316 317 318 319 320 321 322 323 324 325 326 327 328 329 330 331 332 333 334 335 | """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
|