Manages workflow states and coordinates between agent and frontend.
| 15 | |
| 16 | |
| 17 | class WorkflowManager: |
| 18 | """Manages workflow states and coordinates between agent and frontend.""" |
| 19 | |
| 20 | def __init__(self, connection_manager: ConnectionManager): |
| 21 | self.workflows: Dict[str, WorkflowState] = {} |
| 22 | self.agent = ScrapeCraftAgent() |
| 23 | self.connection_manager = connection_manager |
| 24 | self.approval_callbacks: Dict[str, asyncio.Event] = {} |
| 25 | |
| 26 | def get_workflow(self, pipeline_id: str) -> Optional[WorkflowState]: |
| 27 | """Get workflow state for a pipeline.""" |
| 28 | return self.workflows.get(pipeline_id) |
| 29 | |
| 30 | def create_workflow(self, pipeline_id: str, user: str = "system") -> WorkflowState: |
| 31 | """Create a new workflow state.""" |
| 32 | workflow = WorkflowState( |
| 33 | pipeline_id=pipeline_id, |
| 34 | created_by=user |
| 35 | ) |
| 36 | self.workflows[pipeline_id] = workflow |
| 37 | return workflow |
| 38 | |
| 39 | async def process_message( |
| 40 | self, |
| 41 | pipeline_id: str, |
| 42 | message: str, |
| 43 | user: str = "user" |
| 44 | ) -> Dict[str, Any]: |
| 45 | """Process a message through the workflow.""" |
| 46 | # Get or create workflow |
| 47 | workflow = self.get_workflow(pipeline_id) or self.create_workflow(pipeline_id, user) |
| 48 | |
| 49 | # Process through agent |
| 50 | result = await self.agent.process_message(message, pipeline_id) |
| 51 | |
| 52 | # Update workflow state based on result |
| 53 | await self._update_workflow_from_result(workflow, result) |
| 54 | |
| 55 | # Broadcast state update |
| 56 | await self._broadcast_workflow_update(workflow) |
| 57 | |
| 58 | return { |
| 59 | "response": result["response"], |
| 60 | "workflow_state": workflow.model_dump(mode='json'), |
| 61 | "requires_action": len(workflow.pending_approvals) > 0 |
| 62 | } |
| 63 | |
| 64 | async def update_urls( |
| 65 | self, |
| 66 | pipeline_id: str, |
| 67 | urls: List[Dict[str, Any]], |
| 68 | user: str = "user" |
| 69 | ) -> WorkflowState: |
| 70 | """Manually update URLs in the workflow.""" |
| 71 | workflow = self.get_workflow(pipeline_id) |
| 72 | if not workflow: |
| 73 | raise ValueError(f"Workflow not found for pipeline {pipeline_id}") |
| 74 |
no outgoing calls
no test coverage detected