Update core_mcp.py

core_mcp.py

@@ -5,10 +5,12 @@ import time
 import uuid
 import requests
 import uvicorn
+import asyncio
 from datetime import datetime
 from configparser import ConfigParser
 from typing import List, Dict, Tuple, Optional, Any
 from fastapi import FastAPI, HTTPException, BackgroundTasks
+from fastapi.responses import StreamingResponse
 from pydantic import BaseModel
 
 # --- CONFIGURATION & REGISTRY ---
@@ -471,7 +473,18 @@ class ConductorEngine:
         gemini_role = 'model' if role == 'assistant' else 'user'
         self.strategist_history.append({"role": gemini_role, "parts": [{"text": content}]})
 
+    def _current_state(self) -> Dict[str, Any]:
+        """Package current state for transmission"""
+        return {
+            "system_log": self.system_log,
+            "coder_log": self.coder_history,
+            "strategist_log": self.strategist_history,
+            "audit_log": self.audit_log,
+            "final_output": ""
+        }
+
     def execute_workflow(self, blueprint: str, intent: str) -> Dict[str, Any]:
+        """ORIGINAL: Non-streaming version (kept for backwards compatibility)"""
         self.coder_history = []
         self.strategist_history = []
         self.system_log = []
@@ -531,6 +544,84 @@ class ConductorEngine:
         self.log_system("FATAL: Loop limit reached.")
         return self._compile_results(coder_response)
 
+    def execute_workflow_streaming(self, blueprint: str, intent: str):
+        """NEW: Streaming version that yields updates as they happen"""
+        self.coder_history = []
+        self.strategist_history = []
+        self.system_log = []
+        self.audit_log = []
+        self.loop_count = 0
+
+        self.log_system("INITIATING CONDUCTOR ENGINE...")
+        yield self._current_state()
+        
+        for log_msg in self.model_resolution_logs:
+            self.log_system(log_msg)
+            yield self._current_state()
+
+        if self.fatal_init_error:
+            self.log_system("FATAL: Model configuration invalid.")
+            yield self._current_state()
+            return
+
+        self.log_system(f"Engaged: Coder ({self.coder_model_id}), Strategist ({self.strategist_model_id})")
+        yield self._current_state()
+        
+        initial_prompt = format_initial_prompt(blueprint)
+        self.log_coder("user", initial_prompt)
+        yield self._current_state()
+        
+        while self.loop_count < self.max_loops:
+            self.loop_count += 1
+            self.log_system(f"Executing Cycle {self.loop_count}/{self.max_loops}")
+            yield self._current_state()
+            
+            coder_response = call_ai_api(self.coder_url, CONFIG['CODER_API_KEY'], self.coder_history)
+            self.log_coder("assistant", coder_response)
+            yield self._current_state()
+            
+            if "DETECTED" not in coder_response:
+                self.log_system("MISSION COMPLETE: Ambiguity resolved or Code generated.")
+                yield self._current_state()
+                return
+            
+            self.log_system("Ambiguity DETECTED. Escalating to Strategist.")
+            yield self._current_state()
+            
+            strategist_prompt = format_strategist_prompt(blueprint, intent, coder_response)
+            self.log_strategist("user", strategist_prompt)
+            yield self._current_state()
+            
+            strategist_response = call_ai_api(self.strategist_url, CONFIG['STRATEGIST_API_KEY'], self.strategist_history)
+            self.log_strategist("assistant", strategist_response)
+            yield self._current_state()
+            
+            tool_request = detect_tool_usage(strategist_response)
+            if tool_request:
+                self.log_system(f"TOOL REQUEST: {tool_request['tool_name']}")
+                yield self._current_state()
+                
+                diagnostic = execute_tool(tool_request['tool_name'], tool_request['params'])
+                self.audit_log.append(diagnostic)
+                self.log_system(f"Tool Result: {diagnostic['status']}")
+                yield self._current_state()
+                
+                tool_result_text = f"TOOL: {diagnostic['tool_name']}\nSTATUS: {diagnostic['status']}\nOUTPUT: {json.dumps(diagnostic['response_payload'])}"
+                synthesis_prompt = format_strategist_tool_synthesis_prompt(blueprint, intent, coder_response, tool_result_text)
+                self.log_strategist("user", synthesis_prompt)
+                yield self._current_state()
+                
+                strategist_response = call_ai_api(self.strategist_url, CONFIG['STRATEGIST_API_KEY'], self.strategist_history)
+                self.log_strategist("assistant", strategist_response)
+                yield self._current_state()
+
+            resolution_prompt = format_resolution_prompt(blueprint, strategist_response)
+            self.log_coder("user", resolution_prompt)
+            yield self._current_state()
+
+        self.log_system("FATAL: Loop limit reached.")
+        yield self._current_state()
+
     def _compile_results(self, final_output: str) -> Dict[str, Any]:
         return {
             "system_log": self.system_log,
@@ -559,9 +650,25 @@ class ExecutionResponse(BaseModel):
 
 @app.post("/execute", response_model=ExecutionResponse)
 async def execute_conductor(req: ExecutionRequest):
+    """ORIGINAL: Non-streaming endpoint (kept for backwards compatibility)"""
     engine = ConductorEngine(req.coder_model_label, req.strategist_model_label)
     result = engine.execute_workflow(req.blueprint, req.intent)
     return result
 
+@app.post("/execute_stream")
+async def execute_conductor_stream(req: ExecutionRequest):
+    """NEW: Streaming endpoint that yields real-time updates"""
+    
+    async def generate_updates():
+        engine = ConductorEngine(req.coder_model_label, req.strategist_model_label)
+        
+        for update in engine.execute_workflow_streaming(req.blueprint, req.intent):
+            yield f"data: {json.dumps(update)}\n\n"
+            await asyncio.sleep(0)
+        
+        yield "data: {\"status\": \"complete\"}\n\n"
+    
+    return StreamingResponse(generate_updates(), media_type="text/event-stream")
+
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=7860)