import gradio as gr import requests import json import os import time import tempfile import re from configparser import ConfigParser from typing import List, Dict, Any MCP_SERVER_URL = "https://mcp-1st-birthday-project-conductor-core-mcp.hf.space/execute" # HARDCODED PROVIDER MAPPING PROVIDER_MODEL_MAPPING = { "Google Gemini": [], "Anthropic": [], "OpenAI": [], "Custom": [] } def classify_model_by_provider(model_label: str) -> str: """Classify a model label into a provider category using keyword matching""" label_lower = model_label.lower() if 'gemini' in label_lower: return "Google Gemini" elif 'claude' in label_lower or 'anthropic' in label_lower: return "Anthropic" elif 'gpt' in label_lower or 'openai' in label_lower: return "OpenAI" else: return "Custom" def load_client_config(): config = {} coder_models = {} strategist_models = {} if os.path.exists('config.ini'): parser = ConfigParser() parser.read('config.ini') try: if 'CODER_MODELS' in parser: for label, entry in parser['CODER_MODELS'].items(): coder_models[label] = label if 'STRATEGIST_MODELS' in parser: for label, entry in parser['STRATEGIST_MODELS'].items(): strategist_models[label] = label except Exception as e: print(f"Config parsing warning: {e}") # Read defaults from config.ini [API] section if 'API' in parser: config['coder_default'] = parser['API'].get('coder_model_default_label', '').strip() config['strategist_default'] = parser['API'].get('strategist_model_default_label', '').strip() else: config['coder_default'] = '' config['strategist_default'] = '' else: config['coder_default'] = '' config['strategist_default'] = '' # Build provider-grouped structure coder_by_provider = {p: [] for p in PROVIDER_MODEL_MAPPING.keys()} strategist_by_provider = {p: [] for p in PROVIDER_MODEL_MAPPING.keys()} for label in coder_models.keys(): provider = classify_model_by_provider(label) coder_by_provider[provider].append(label) for label in strategist_models.keys(): provider = classify_model_by_provider(label) strategist_by_provider[provider].append(label) config['coder_by_provider'] = coder_by_provider config['strategist_by_provider'] = strategist_by_provider config['coder_models'] = list(coder_models.keys()) config['strategist_models'] = list(strategist_models.keys()) # Fallback to environment variables only if config.ini didn't provide values if not config['coder_default']: config['coder_default'] = os.getenv('CODER_MODEL_DEFAULT_LABEL', '') if not config['strategist_default']: config['strategist_default'] = os.getenv('STRATEGIST_MODEL_DEFAULT_LABEL', '') return config CLIENT_CONFIG = load_client_config() def format_conversation_log(history: List[Dict[str, Any]]) -> str: log_lines = [] for i, msg in enumerate(history): role_label = msg.get('role', 'UNKNOWN').upper() text_content = "" if 'parts' in msg and isinstance(msg['parts'], list): text_content = msg['parts'][0].get('text', '') log_lines.append(f"{'='*80}") log_lines.append(f"[{i}] {role_label}") log_lines.append(f"{'='*80}") log_lines.append(text_content) log_lines.append("") return "\n".join(log_lines) def build_master_log(system_log_list, coder_log_list, strategist_log_list, audit_log_list, start_time, final_code_block=None): """Build master log with enhanced tool failure reporting""" actor_styles = { 'SYSTEM': {'icon': '🔧', 'color': '#ef4444'}, 'CODER': {'icon': '💻', 'color': '#3b82f6'}, 'STRATEGIST': {'icon': '🎯', 'color': '#a855f7'}, 'TOOL': {'icon': '🛠️', 'color': '#f97316'} } master_entries = [] seq_counter = 1 for entry in system_log_list: elapsed = time.time() - start_time minutes = int(elapsed // 60) seconds = int(elapsed % 60) milliseconds = int((elapsed % 1) * 1000) timestamp = f"{minutes:02d}:{seconds:02d}:{milliseconds:03d}" style = actor_styles['SYSTEM'] html_entry = f'
{style["icon"]} [SEQ-{seq_counter:02d}] [{timestamp}] SYSTEM: {entry}
' master_entries.append(html_entry) seq_counter += 1 for msg in coder_log_list: elapsed = time.time() - start_time minutes = int(elapsed // 60) seconds = int(elapsed % 60) milliseconds = int((elapsed % 1) * 1000) timestamp = f"{minutes:02d}:{seconds:02d}:{milliseconds:03d}" role = msg.get('role', 'UNKNOWN').upper() actor = f"CODER ({role})" text_content = "" if 'parts' in msg and isinstance(msg['parts'], list): text_content = msg['parts'][0].get('text', '') style = actor_styles['CODER'] html_entry = f'
{style["icon"]} [SEQ-{seq_counter:02d}] [{timestamp}] {actor}: {text_content}
' master_entries.append(html_entry) seq_counter += 1 for msg in strategist_log_list: elapsed = time.time() - start_time minutes = int(elapsed // 60) seconds = int(elapsed % 60) milliseconds = int((elapsed % 1) * 1000) timestamp = f"{minutes:02d}:{seconds:02d}:{milliseconds:03d}" role = msg.get('role', 'UNKNOWN').upper() actor = f"STRATEGIST ({role})" text_content = "" if 'parts' in msg and isinstance(msg['parts'], list): text_content = msg['parts'][0].get('text', '') style = actor_styles['STRATEGIST'] html_entry = f'
{style["icon"]} [SEQ-{seq_counter:02d}] [{timestamp}] {actor}: {text_content}
' master_entries.append(html_entry) seq_counter += 1 # Enhanced tool failure logging for diagnostic in audit_log_list: elapsed = time.time() - start_time minutes = int(elapsed // 60) seconds = int(elapsed % 60) milliseconds = int((elapsed % 1) * 1000) timestamp = f"{minutes:02d}:{seconds:02d}:{milliseconds:03d}" tool_name = diagnostic.get('tool_name', 'UNKNOWN') status = diagnostic.get('status', 'UNKNOWN') latency = diagnostic.get('latency', 'N/A') transaction_id = diagnostic.get('transaction_id', 'N/A') request_payload = diagnostic.get('request_payload', {}) response_payload = diagnostic.get('response_payload', None) # Format based on status if status == 'FAILURE': request_str = json.dumps(request_payload, indent=2) error_msg = diagnostic.get('error_message', 'Unknown error') tool_entry = f"TOOL: {tool_name} | STATUS: FAILURE | ATTEMPTED PARAMS: {request_str}
ERROR: {error_msg}" else: request_str = json.dumps(request_payload, indent=2) response_str = json.dumps(response_payload, indent=2) if response_payload else "None" tool_entry = f"TOOL: {tool_name} | TRANSACTION: {transaction_id} | STATUS: {status} | LATENCY: {latency}s
[REQUEST]
{request_str}

[RESPONSE]
{response_str}
" style = actor_styles['TOOL'] html_entry = f'
{style["icon"]} [SEQ-{seq_counter:02d}] [{timestamp}] TOOL: {tool_entry}
' master_entries.append(html_entry) seq_counter += 1 # Append final code block if provided if final_code_block: code_html = f'''

🎯 GENERATED CODE:

{final_code_block}
''' master_entries.append(code_html) if not master_entries: return '
No log data captured.
' html_output = f'''
{"".join(master_entries)}
''' return html_output def execute_remote_workflow(blueprint, intent, coder_model_label, strategist_model_label, enable_audit): """Execute workflow with final_code and master_log tracking""" mission_start_time = time.time() final_code = None payload = { "blueprint": blueprint, "intent": intent, "coder_model_label": coder_model_label, "strategist_model_label": strategist_model_label, "enable_audit": enable_audit } try: target_url = MCP_SERVER_URL.replace("/execute", "/execute_stream") print(f"DEBUG: Connecting to streaming endpoint: {target_url}") with requests.post(target_url, json=payload, stream=True, timeout=300) as response: if response.status_code == 404: yield f"ERROR 404: Streaming endpoint not found at {target_url}. Check server version.", "", "", "", None, None return response.raise_for_status() for line in response.iter_lines(): if not line: continue decoded = line.decode('utf-8') if not decoded.startswith('data: '): continue data_str = decoded[6:] try: data = json.loads(data_str) except json.JSONDecodeError: continue if data.get("status") == "complete": break # Watch for final_code key if "final_code" in data: final_code = data["final_code"] system_log_list = data.get("system_log", []) coder_log_list = data.get("coder_log", []) strategist_log_list = data.get("strategist_log", []) audit_log_list = data.get("audit_log", []) sys_log = "\n".join(system_log_list) cod_log = format_conversation_log(coder_log_list) strat_log = format_conversation_log(strategist_log_list) master_log = build_master_log(system_log_list, coder_log_list, strategist_log_list, audit_log_list, mission_start_time, final_code) # DIRECTIVE #3: Update both master_log_output AND master_log_state yield sys_log, cod_log, strat_log, master_log, final_code, master_log except requests.exceptions.RequestException as e: error_detail = str(e) if hasattr(e, 'response') and e.response is not None: error_detail += f"\nServer Response: {e.response.text}" error_msg = f"CONNECTION ERROR: Could not connect to MCP Server.\nDetails: {error_detail}" yield error_msg, "", "", "", None, None except Exception as e: error_msg = f"CLIENT ERROR: {str(e)}" yield error_msg, "", "", "", None, None def create_interface(): DEMO_BLUEPRINT = """# Gradio Application Blueprint Produce a pro level styled and tuned them GUI that has 1 button that when pressed displays a celebration announcement "Happy Birthday MCP!". USE specified colors or themes appropriate for the event. You must incorporate specific colors for design pieces and thematic layouts providing a fun and exciting announcement.""" DEMO_INTENT = """A simple test button with a clean look. That when pressed creates a simple happy birthday MCP response on the screen. The gui should be gradio and contain colors that are used in professional builds and match the Gradio, MCP or other color theme or as close as possible.""" # Use config defaults explicitly coder_default = CLIENT_CONFIG['coder_default'] strategist_default = CLIENT_CONFIG['strategist_default'] # Determine initial provider for each agent initial_coder_provider = "Google Gemini" initial_strategist_provider = "Google Gemini" if coder_default: initial_coder_provider = classify_model_by_provider(coder_default) if strategist_default: initial_strategist_provider = classify_model_by_provider(strategist_default) # Get initial model lists for providers initial_coder_models = CLIENT_CONFIG['coder_by_provider'].get(initial_coder_provider, []) initial_strategist_models = CLIENT_CONFIG['strategist_by_provider'].get(initial_strategist_provider, []) # Ensure default value is set or fallback to first available coder_initial_value = coder_default if coder_default and coder_default in initial_coder_models else (initial_coder_models[0] if initial_coder_models else None) strategist_initial_value = strategist_default if strategist_default and strategist_default in initial_strategist_models else (initial_strategist_models[0] if initial_strategist_models else None) def load_demo_data(): """Switch Coder to gemini 2.5 pro when demo loaded""" demo_coder_model = None for model_label in CLIENT_CONFIG['coder_models']: if '2.5' in model_label and 'pro' in model_label.lower(): demo_coder_model = model_label break if demo_coder_model: return DEMO_BLUEPRINT, DEMO_INTENT, gr.update(value=demo_coder_model) else: return DEMO_BLUEPRINT, DEMO_INTENT, gr.update() def start_execution_sequence(): return gr.update(visible=False), gr.update(visible=True), gr.update(visible=False) # DIRECTIVE #3: Button visibility control in end sequence def end_execution_sequence(sys_log, cod_log, strat_log, master_log, final_code, master_log_html): # Buttons become visible only when final_code exists button_visibility = final_code is not None and final_code.strip() != "" return ( sys_log, cod_log, strat_log, master_log, final_code, master_log_html, gr.update(visible=False), # working_video gr.update(visible=True), # celebrate_video gr.update(visible=button_visibility) # download_buttons_row ) def reset_to_default(): return gr.update(visible=True), gr.update(visible=False), gr.update(visible=False) # Cascading selector update functions def update_coder_models(provider): models = CLIENT_CONFIG['coder_by_provider'].get(provider, []) if not models: return gr.update(choices=[], value=None) return gr.update(choices=models, value=models[0]) def update_strategist_models(provider): models = CLIENT_CONFIG['strategist_by_provider'].get(provider, []) if not models: return gr.update(choices=[], value=None) return gr.update(choices=models, value=models[0]) # DIRECTIVE #1: NO LOGIC CHANGE - Existing download_code function preserved def download_code(final_code_content): """Generate downloadable Python file from final code""" if not final_code_content or final_code_content.strip() == "": return None # Create temporary file with timestamp timestamp = time.strftime("%Y%m%d_%H%M%S") filename = f"generated_code_{timestamp}.py" # Write to temporary file and return filepath temp_dir = tempfile.gettempdir() filepath = os.path.join(temp_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(final_code_content) return filepath # DIRECTIVE #2: NEW FUNCTION - Download master log def download_master_log(master_log_html): """Generate downloadable text file from master log HTML""" if not master_log_html or master_log_html.strip() == "": return None # Strip all HTML tags using regex clean_text = re.sub(r'<[^>]+>', '', master_log_html) # Decode HTML entities clean_text = clean_text.replace('<', '<').replace('>', '>').replace('&', '&') # Clean up extra whitespace clean_text = re.sub(r'\n\s*\n', '\n\n', clean_text) clean_text = clean_text.strip() # Create temporary file with timestamp timestamp = time.strftime("%Y%m%d_%H%M%S") filename = f"master_log_{timestamp}.txt" # Write to temporary file and return filepath temp_dir = tempfile.gettempdir() filepath = os.path.join(temp_dir, filename) with open(filepath, 'w', encoding='utf-8') as f: f.write(clean_text) return filepath # RESET MISSION function def reset_mission(): """Total system reset: clear all fields and reset to defaults""" coder_models_for_reset = CLIENT_CONFIG['coder_by_provider'].get(initial_coder_provider, []) strategist_models_for_reset = CLIENT_CONFIG['strategist_by_provider'].get(initial_strategist_provider, []) coder_reset_value = coder_initial_value strategist_reset_value = strategist_initial_value return ( "", # blueprint_input "", # intent_input "", # system_terminal "", # coder_log "", # strategist_log '
*Mission reset complete. All systems cleared.*
', # master_log gr.update(value=initial_coder_provider), # coder_provider gr.update(choices=coder_models_for_reset, value=coder_reset_value), # coder_model gr.update(value=initial_strategist_provider), # strategist_provider gr.update(choices=strategist_models_for_reset, value=strategist_reset_value), # strategist_model gr.update(value=False), # audit_checkbox None, # final_code_state None, # master_log_state gr.update(visible=False) # DIRECTIVE #3: Hide download buttons row on reset ) with gr.Blocks(title="Project: Conductor V4.0 Client") as interface: gr.Markdown("# Project: Conductor V4.0 (Client Mode)") gr.Markdown(f"### The Command Center: Connected to MCP Server at `{MCP_SERVER_URL}`") # DIRECTIVE #3: Hidden states for final code and master log final_code_state = gr.State(value=None) master_log_state = gr.State(value=None) # PRIMARY INTERFACE ROW - Three Columns with gr.Row(): # COLUMN 1 - CODER (Left Flank) with gr.Column(): gr.Markdown("### Coder Agent") coder_provider = gr.Dropdown( choices=list(PROVIDER_MODEL_MAPPING.keys()), value=initial_coder_provider, label="Provider", interactive=True ) coder_model = gr.Dropdown( choices=initial_coder_models, value=coder_initial_value, label="Model", interactive=True ) blueprint_input = gr.Textbox( label="Architect's Blueprint", placeholder="Enter detailed technical specifications...", lines=12 ) # COLUMN 2 - MEDIA BOX (Center Stage) with gr.Column(): gr.Markdown("### Media Box") default_image = gr.Image( value="ai_creative_tech .png", show_label=False, visible=True, interactive=False, type="filepath", height=450 ) working_video = gr.Video( value="ai_ct_working.mp4", show_label=False, visible=False, autoplay=True, loop=True, height=450 ) celebrate_video = gr.Video( value="ai_creative_tech_celebrate.mp4", show_label=False, visible=False, autoplay=True, loop=False, height=450 ) # COLUMN 3 - STRATEGIST (Right Flank) with gr.Column(): gr.Markdown("### Strategist Agent") strategist_provider = gr.Dropdown( choices=list(PROVIDER_MODEL_MAPPING.keys()), value=initial_strategist_provider, label="Provider", interactive=True ) strategist_model = gr.Dropdown( choices=initial_strategist_models, value=strategist_initial_value, label="Model", interactive=True ) intent_input = gr.Textbox( label="Architect's Intent", placeholder="Enter project goals and context...", lines=12 ) # ACTION ROW - Buttons with gr.Row(): demo_btn = gr.Button("LOAD DEMO MODE", variant="secondary") execute_btn = gr.Button("EXECUTE VIA MCP", variant="primary") reset_btn = gr.Button("RESET MISSION", variant="secondary") audit_checkbox = gr.Checkbox(label="AUDIT", value=False) # LOG ROWS with gr.Row(): system_terminal = gr.Textbox(label="SYSTEM LOG", interactive=False, lines=15, max_lines=15) coder_log = gr.Textbox(label="CODER LOG", interactive=False, lines=15, max_lines=15) strategist_log = gr.Textbox(label="STRATEGIST LOG", interactive=False, lines=15, max_lines=15) with gr.Row(): master_log_output = gr.HTML( value='
No mission log data available.
', label="MASTER LOG WINDOW" ) # DIRECTIVE #3: Two-button layout with clean, constrained width with gr.Row(visible=False) as download_buttons_row: with gr.Column(): download_code_btn = gr.DownloadButton( label="DOWNLOAD GENERATED CODE", variant="primary" ) with gr.Column(): download_log_btn = gr.DownloadButton( label="DOWNLOAD MASTER LOG", variant="secondary" ) # EVENT HANDLERS # Cascading model selector logic coder_provider.change( fn=update_coder_models, inputs=[coder_provider], outputs=[coder_model] ) strategist_provider.change( fn=update_strategist_models, inputs=[strategist_provider], outputs=[strategist_model] ) # Demo button demo_btn.click( fn=load_demo_data, inputs=None, outputs=[blueprint_input, intent_input, coder_model] ) # Execute button execute_btn.click( fn=start_execution_sequence, inputs=None, outputs=[default_image, working_video, celebrate_video] ).then( fn=execute_remote_workflow, inputs=[blueprint_input, intent_input, coder_model, strategist_model, audit_checkbox], outputs=[system_terminal, coder_log, strategist_log, master_log_output, final_code_state, master_log_state] ).then( fn=end_execution_sequence, inputs=[system_terminal, coder_log, strategist_log, master_log_output, final_code_state, master_log_state], outputs=[system_terminal, coder_log, strategist_log, master_log_output, final_code_state, master_log_state, working_video, celebrate_video, download_buttons_row] ) # DIRECTIVE #1 & #2: Download button handlers download_code_btn.click( fn=download_code, inputs=[final_code_state], outputs=[download_code_btn] ) download_log_btn.click( fn=download_master_log, inputs=[master_log_state], outputs=[download_log_btn] ) # Celebrate video auto-reset celebrate_video.stop( fn=reset_to_default, inputs=None, outputs=[default_image, working_video, celebrate_video] ) # RESET MISSION button with download buttons row visibility control reset_btn.click( fn=reset_mission, inputs=[], outputs=[ blueprint_input, intent_input, system_terminal, coder_log, strategist_log, master_log_output, coder_provider, coder_model, strategist_provider, strategist_model, audit_checkbox, final_code_state, master_log_state, download_buttons_row ] ) gr.Markdown("---") gr.Markdown("**Note:** This client requires the Conductor MCP Server V4.0 with streaming support. Logs now update in real-time with color-coded visual differentiation. **AUDIT mode will loop code back to Strategist for review.**") return interface if __name__ == "__main__": app = create_interface() app.queue() app.launch()