"""Workflow API routes - CRUD, execution, approval, and real-time progress""" from __future__ import annotations import asyncio import json import logging import re import uuid from datetime import datetime, timezone from typing import Any from fastapi import APIRouter, Depends, HTTPException, Request, WebSocket, WebSocketDisconnect, Security from fastapi.security import APIKeyHeader, APIKeyQuery from agentkit.orchestrator.workflow_schema import ( ApproveRequest, CreateWorkflowRequest, ExecuteWorkflowRequest, WorkflowDefinition, WorkflowExecution, WorkflowStage, WorkflowSummary, ) logger = logging.getLogger(__name__) router = APIRouter(tags=["workflows"]) # --------------------------------------------------------------------------- # API Key Authentication # --------------------------------------------------------------------------- _api_key_header = APIKeyHeader(name="X-API-Key", auto_error=False) _api_key_query = APIKeyQuery(name="api_key", auto_error=False) async def _verify_api_key( request: Request, api_key_header: str | None = Security(_api_key_header), api_key_query: str | None = Security(_api_key_query), ) -> None: """Verify API key for REST endpoints. Raises HTTPException if invalid.""" configured_api_key: str | None = None if hasattr(request.app.state, "server_config") and request.app.state.server_config: configured_api_key = request.app.state.server_config.api_key if configured_api_key is None and hasattr(request.app.state, "api_key"): configured_api_key = request.app.state.api_key # If no API key is configured, allow all requests (backwards compat) if configured_api_key is None: return provided = api_key_header or api_key_query if provided != configured_api_key: raise HTTPException( status_code=401, detail="Invalid or missing API key. Provide via X-API-Key header or api_key query parameter.", ) # --------------------------------------------------------------------------- # In-memory Workflow Store # --------------------------------------------------------------------------- class WorkflowStore: """In-memory workflow store.""" def __init__(self, max_workflows: int = 500, max_executions: int = 1000): self._workflows: dict[str, WorkflowDefinition] = {} self._executions: dict[str, WorkflowExecution] = {} self._max_workflows = max_workflows self._max_executions = max_executions self._approval_events: dict[str, asyncio.Event] = {} # key: f"{execution_id}:{stage_name}" def save(self, workflow: WorkflowDefinition) -> WorkflowDefinition: workflow.updated_at = datetime.now(timezone.utc).isoformat() self._workflows[workflow.workflow_id] = workflow # Evict oldest if over limit if len(self._workflows) > self._max_workflows: oldest_id = min( self._workflows, key=lambda k: self._workflows[k].updated_at ) del self._workflows[oldest_id] return workflow def get(self, workflow_id: str) -> WorkflowDefinition | None: return self._workflows.get(workflow_id) def list(self, limit: int = 50) -> list[WorkflowSummary]: sorted_wf = sorted( self._workflows.values(), key=lambda w: w.updated_at, reverse=True, ) summaries = [] for w in sorted_wf[:limit]: summaries.append( WorkflowSummary( workflow_id=w.workflow_id, name=w.name, version=w.version, stage_count=len(w.stages), trigger_count=len(w.triggers), created_at=w.created_at, updated_at=w.updated_at, ) ) return summaries def delete(self, workflow_id: str) -> bool: if workflow_id in self._workflows: del self._workflows[workflow_id] return True return False def create_execution(self, workflow_id: str) -> WorkflowExecution: execution = WorkflowExecution( execution_id=str(uuid.uuid4()), workflow_id=workflow_id, status="pending", started_at=datetime.now(timezone.utc).isoformat(), ) self._executions[execution.execution_id] = execution # Evict oldest if over limit if len(self._executions) > self._max_executions: oldest_id = min( self._executions, key=lambda k: self._executions[k].started_at or "", ) del self._executions[oldest_id] return execution def get_execution(self, execution_id: str) -> WorkflowExecution | None: return self._executions.get(execution_id) def update_execution(self, execution_id: str, **kwargs: Any) -> WorkflowExecution: execution = self._executions.get(execution_id) if execution is None: raise KeyError(f"Execution '{execution_id}' not found") for key, value in kwargs.items(): if hasattr(execution, key): setattr(execution, key, value) return execution # Module-level singleton _workflow_store = WorkflowStore() # WebSocket subscribers for real-time execution progress _ws_subscribers: list[WebSocket] = [] def _get_store(request: Request) -> WorkflowStore: """Get the workflow store from app state or use the module-level singleton.""" store = getattr(request.app.state, "workflow_store", None) return store if store is not None else _workflow_store # --------------------------------------------------------------------------- # Validation helpers # --------------------------------------------------------------------------- def _validate_workflow_stages(stages: list[WorkflowStage]) -> None: """Validate workflow stages for missing dependencies and circular deps.""" stage_names = {s.name for s in stages} for stage in stages: for dep in stage.depends_on: if dep not in stage_names: raise HTTPException( status_code=400, detail=f"阶段 '{stage.name}' 依赖了不存在的阶段 '{dep}'", ) # Check for circular dependencies in_degree: dict[str, int] = {s.name: 0 for s in stages} dependents: dict[str, list[str]] = {s.name: [] for s in stages} for s in stages: for dep in s.depends_on: in_degree[s.name] += 1 dependents[dep].append(s.name) remaining = set(in_degree.keys()) while remaining: current_level = [name for name in remaining if in_degree[name] == 0] if not current_level: raise HTTPException( status_code=400, detail="工作流存在循环依赖", ) for name in current_level: remaining.remove(name) for dep in dependents[name]: in_degree[dep] -= 1 # --------------------------------------------------------------------------- # Workflow execution engine (simplified) # --------------------------------------------------------------------------- async def _execute_workflow( workflow: WorkflowDefinition, execution: WorkflowExecution, variables: dict[str, Any], store: WorkflowStore | None = None, ) -> None: """Execute a workflow by running its stages in topological order.""" _store = store or _workflow_store execution.status = "running" _store.update_execution(execution.execution_id, status="running") # Topological sort stage_map = {s.name: s for s in workflow.stages} in_degree: dict[str, int] = {s.name: 0 for s in workflow.stages} dependents: dict[str, list[str]] = {s.name: [] for s in workflow.stages} for s in workflow.stages: for dep in s.depends_on: in_degree[s.name] += 1 dependents[dep].append(s.name) ordered: list[str] = [] remaining = set(in_degree.keys()) while remaining: current_level = [name for name in remaining if in_degree[name] == 0] if not current_level: execution.status = "failed" execution.error = "循环依赖" execution.completed_at = datetime.now(timezone.utc).isoformat() _store.update_execution( execution.execution_id, status="failed", error="循环依赖", completed_at=execution.completed_at, ) return for name in sorted(current_level): ordered.append(name) remaining.remove(name) for dep in dependents[name]: in_degree[dep] -= 1 # Execute stages in order for stage_name in ordered: stage = stage_map[stage_name] execution.current_stage = stage_name _store.update_execution( execution.execution_id, current_stage=stage_name, ) # Notify WebSocket subscribers await _broadcast_ws({ "event": "stage_started", "execution_id": execution.execution_id, "stage": stage_name, }) try: if stage.type == "approval": # Pause execution and wait for approval via asyncio.Event event_key = f"{execution.execution_id}:{stage_name}" approval_event = asyncio.Event() _store._approval_events[event_key] = approval_event execution.status = "paused" execution.current_stage = stage_name _store.update_execution( execution.execution_id, status="paused", current_stage=stage_name, ) await _broadcast_ws({ "event": "approval_required", "execution_id": execution.execution_id, "stage": stage_name, }) # Wait for approval with timeout try: approval_timeout = stage.config.get("approval_timeout", 3600) await asyncio.wait_for(approval_event.wait(), timeout=approval_timeout) # Check if execution was cancelled/rejected while waiting if execution.status == "cancelled": await _broadcast_ws({ "event": "stage_failed", "execution_id": execution.execution_id, "stage": stage_name, "error": "Approval rejected", }) return # Approval was granted — the /approve endpoint already set stage_results # Only update status to running if not already set if execution.status != "running": execution.status = "running" _store.update_execution( execution.execution_id, status="running", ) except asyncio.TimeoutError: execution.stage_results[stage_name] = { "status": "timeout", "approver": "none", "comment": "审批超时", } execution.status = "failed" execution.error = f"Approval timeout for stage {stage_name}" execution.completed_at = datetime.now(timezone.utc).isoformat() _store.update_execution( execution.execution_id, status="failed", error=execution.error, completed_at=execution.completed_at, stage_results=execution.stage_results, ) await _broadcast_ws({ "event": "stage_failed", "execution_id": execution.execution_id, "stage": stage_name, "error": "Approval timeout", }) return finally: _store._approval_events.pop(event_key, None) elif stage.type == "condition": # Evaluate condition expression condition_expr = stage.config.get("expression", "") result = _evaluate_condition(condition_expr, variables) execution.stage_results[stage_name] = { "status": "completed", "condition_result": result, } _store.update_execution( execution.execution_id, stage_results=execution.stage_results, ) else: # Skill or parallel stage - simulate execution execution.stage_results[stage_name] = { "status": "completed", "output": {"dry_run": True, "action": stage.action}, } _store.update_execution( execution.execution_id, stage_results=execution.stage_results, ) await _broadcast_ws({ "event": "stage_completed", "execution_id": execution.execution_id, "stage": stage_name, }) except Exception as e: execution.stage_results[stage_name] = { "status": "failed", "error": str(e), } execution.status = "failed" execution.error = f"阶段 '{stage_name}' 执行失败: {e}" execution.completed_at = datetime.now(timezone.utc).isoformat() _store.update_execution( execution.execution_id, status="failed", error=execution.error, completed_at=execution.completed_at, stage_results=execution.stage_results, ) await _broadcast_ws({ "event": "stage_failed", "execution_id": execution.execution_id, "stage": stage_name, "error": str(e), }) return execution.status = "completed" execution.completed_at = datetime.now(timezone.utc).isoformat() execution.current_stage = None _store.update_execution( execution.execution_id, status="completed", completed_at=execution.completed_at, current_stage=None, ) await _broadcast_ws({ "event": "execution_completed", "execution_id": execution.execution_id, }) _SAFE_VAR_PATTERN = re.compile(r'^[a-zA-Z_][a-zA-Z0-9_]*$') _SAFE_OPERATORS = {"==", "!=", ">", "<", ">=", "<="} def _evaluate_condition(expression: str, variables: dict[str, Any]) -> bool: """Evaluate a condition expression safely.""" expression = expression.strip() if not expression: return True # Try each operator (longer operators first to avoid partial matches) for op in sorted(_SAFE_OPERATORS, key=len, reverse=True): if op in expression: parts = expression.split(op, 1) if len(parts) != 2: continue left = parts[0].strip() right = parts[1].strip() # Validate variable names if left and not _SAFE_VAR_PATTERN.match(left): raise ValueError(f"Invalid variable name in condition: {left}") if left and left not in variables: # Missing variable: treat as None/empty — condition evaluates to False left_val = None else: left_val = variables.get(left, left) # Strip quotes from right side if present if right.startswith('"') and right.endswith('"'): right_val = right[1:-1] elif right.startswith("'") and right.endswith("'"): right_val = right[1:-1] elif right and _SAFE_VAR_PATTERN.match(right): right_val = variables.get(right, right) else: right_val = right # Compare based on operator if op == "==": return str(left_val) == str(right_val) if op == "!=": return str(left_val) != str(right_val) try: left_num = float(left_val) right_num = float(right_val) except (ValueError, TypeError): return False if op == ">": return left_num > right_num if op == "<": return left_num < right_num if op == ">=": return left_num >= right_num if op == "<=": return left_num <= right_num # Boolean check for variable existence if _SAFE_VAR_PATTERN.match(expression): return bool(variables.get(expression)) raise ValueError(f"Invalid condition expression: {expression}") async def _broadcast_ws(message: dict[str, Any]) -> None: """Broadcast a message to all WebSocket subscribers.""" disconnected = [] for ws in _ws_subscribers: try: await ws.send_json(message) except Exception: disconnected.append(ws) for ws in disconnected: _ws_subscribers.remove(ws) # --------------------------------------------------------------------------- # Endpoints # --------------------------------------------------------------------------- @router.get("/workflows") async def list_workflows(request: Request, limit: int = 50, _auth: None = Depends(_verify_api_key)): """List all workflows.""" store = _get_store(request) summaries = store.list(limit=limit) return {"workflows": [s.model_dump() for s in summaries], "total": len(summaries)} @router.post("/workflows", status_code=201) async def create_workflow(request: Request, body: CreateWorkflowRequest, _auth: None = Depends(_verify_api_key)): """Create a new workflow.""" store = _get_store(request) _validate_workflow_stages(body.stages) workflow = WorkflowDefinition( workflow_id=str(uuid.uuid4()), name=body.name, stages=body.stages, triggers=body.triggers, variables_schema=body.variables_schema, output_schema=body.output_schema, ) saved = store.save(workflow) return saved.model_dump() @router.get("/workflows/{workflow_id}") async def get_workflow(request: Request, workflow_id: str, _auth: None = Depends(_verify_api_key)): """Get a workflow by ID.""" store = _get_store(request) workflow = store.get(workflow_id) if workflow is None: raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在") return workflow.model_dump() @router.put("/workflows/{workflow_id}") async def update_workflow( request: Request, workflow_id: str, body: CreateWorkflowRequest, _auth: None = Depends(_verify_api_key), ): """Update an existing workflow.""" store = _get_store(request) existing = store.get(workflow_id) if existing is None: raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在") _validate_workflow_stages(body.stages) existing.name = body.name existing.stages = body.stages existing.triggers = body.triggers existing.variables_schema = body.variables_schema existing.output_schema = body.output_schema existing.version += 1 saved = store.save(existing) return saved.model_dump() @router.delete("/workflows/{workflow_id}") async def delete_workflow(request: Request, workflow_id: str, _auth: None = Depends(_verify_api_key)): """Delete a workflow.""" store = _get_store(request) deleted = store.delete(workflow_id) if not deleted: raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在") return {"message": "已删除"} @router.post("/workflows/{workflow_id}/execute") async def execute_workflow( request: Request, workflow_id: str, body: ExecuteWorkflowRequest ): """Execute a workflow.""" store = _get_store(request) workflow = store.get(workflow_id) if workflow is None: raise HTTPException(status_code=404, detail=f"工作流 '{workflow_id}' 不存在") execution = store.create_execution(workflow_id) execution.variables = body.variables # Start execution in background task = asyncio.create_task( _execute_workflow(workflow, execution, body.variables, store=store) ) store._running_tasks = getattr(store, "_running_tasks", {}) store._running_tasks[execution.execution_id] = task task.add_done_callback(lambda t: store._running_tasks.pop(execution.execution_id, None)) return { "execution_id": execution.execution_id, "workflow_id": workflow_id, "status": execution.status, } @router.get("/workflows/executions/{execution_id}") async def get_execution(request: Request, execution_id: str): """Get execution status.""" store = _get_store(request) execution = store.get_execution(execution_id) if execution is None: raise HTTPException( status_code=404, detail=f"执行记录 '{execution_id}' 不存在" ) return execution.model_dump() @router.post("/workflows/executions/{execution_id}/approve") async def approve_execution( request: Request, execution_id: str, body: ApproveRequest, _auth: None = Depends(_verify_api_key), ): """Approve a paused approval node.""" store = _get_store(request) execution = store.get_execution(execution_id) if execution is None: raise HTTPException( status_code=404, detail=f"执行记录 '{execution_id}' 不存在" ) if execution.status != "paused": raise HTTPException( status_code=400, detail="当前执行状态不是等待审批" ) if body.approved: if execution.current_stage: execution.stage_results[execution.current_stage] = { "status": "approved", "approver": "user", "comment": body.comment, } execution.status = "running" store.update_execution( execution.execution_id, status="running", stage_results=execution.stage_results, ) # Resume the waiting execution by setting the approval event stage_name = execution.current_stage if stage_name: event_key = f"{execution_id}:{stage_name}" if event_key in store._approval_events: store._approval_events[event_key].set() else: execution.status = "cancelled" execution.completed_at = datetime.now(timezone.utc).isoformat() if execution.current_stage: execution.stage_results[execution.current_stage] = { "status": "rejected", "approver": "user", "comment": body.comment, } store.update_execution( execution.execution_id, status="cancelled", completed_at=execution.completed_at, stage_results=execution.stage_results, ) # Set the approval event so the waiting coroutine can observe the cancelled state stage_name = execution.current_stage if stage_name: event_key = f"{execution_id}:{stage_name}" if event_key in store._approval_events: store._approval_events[event_key].set() return execution.model_dump() @router.post("/workflows/executions/{execution_id}/cancel") async def cancel_execution(request: Request, execution_id: str, _auth: None = Depends(_verify_api_key)): """Cancel a running execution.""" store = _get_store(request) execution = store.get_execution(execution_id) if execution is None: raise HTTPException( status_code=404, detail=f"执行记录 '{execution_id}' 不存在" ) if execution.status not in ("running", "paused", "pending"): raise HTTPException( status_code=400, detail="当前执行状态无法取消" ) execution.status = "cancelled" execution.completed_at = datetime.now(timezone.utc).isoformat() store.update_execution( execution.execution_id, status="cancelled", completed_at=execution.completed_at, ) # Set any pending approval event so a paused workflow can observe the cancelled state if hasattr(execution, "current_stage") and execution.current_stage: event_key = f"{execution_id}:{execution.current_stage}" if event_key in store._approval_events: store._approval_events[event_key].set() return execution.model_dump() @router.websocket("/workflows/ws") async def workflow_websocket(websocket: WebSocket): """Real-time workflow execution progress WebSocket.""" # Authentication configured_api_key: str | None = None if hasattr(websocket.app.state, "server_config") and websocket.app.state.server_config: configured_api_key = websocket.app.state.server_config.api_key if configured_api_key is None and hasattr(websocket.app.state, "api_key"): configured_api_key = websocket.app.state.api_key if configured_api_key: provided = websocket.query_params.get("api_key") if provided != configured_api_key: await websocket.accept() await websocket.send_json( {"event": "error", "data": {"message": "Invalid or missing api_key"}} ) await websocket.close(code=4001, reason="Invalid or missing api_key") return await websocket.accept() _ws_subscribers.append(websocket) try: while True: try: raw = await asyncio.wait_for(websocket.receive_text(), timeout=120.0) except asyncio.TimeoutError: await websocket.close(code=1000, reason="Heartbeat timeout") return # Keep connection alive - messages are primarily server-push except WebSocketDisconnect: logger.debug("Workflow WebSocket disconnected") except Exception as e: logger.error(f"Workflow WebSocket error: {e}") finally: if websocket in _ws_subscribers: _ws_subscribers.remove(websocket)