Function run_workflow_sync

server/routes/execute_sync.py:153–253

(request: WorkflowRunRequest, http_request: Request)

@router.post("/api/workflow/run")
async def run_workflow_sync(request: WorkflowRunRequest, http_request: Request):
    try:
        resolved_log_level: Optional[LogLevel] = None
        if request.log_level:
            resolved_log_level = LogLevel(request.log_level)
    except ValueError:
        raise HTTPException(
            status_code=400,
            detail="log_level must be one of DEBUG, INFO, WARNING, ERROR, CRITICAL",
        )

    accepts_stream = _SSE_CONTENT_TYPE in (http_request.headers.get("accept") or "")
    if not accepts_stream:
        try:
            result = await run_in_threadpool(
                run_workflow,
                request.yaml_file,
                task_prompt=request.task_prompt,
                attachments=request.attachments,
                session_name=request.session_name,
                variables=request.variables,
                log_level=resolved_log_level,
            )
        except FileNotFoundError as exc:
            raise HTTPException(status_code=404, detail=str(exc))
        except ValidationError as exc:
            raise HTTPException(status_code=400, detail=str(exc))
        except Exception as exc:
            logger = get_server_logger()
            logger.log_exception(exc, "Failed to run workflow via sync API")
            raise WorkflowExecutionError(f"Failed to run workflow: {exc}")

        final_message = result.final_message.text_content() if result.final_message else ""
        meta = result.meta_info

        logger = get_server_logger()
        logger.info(
            "Workflow execution completed via sync API",
            log_type=LogType.WORKFLOW,
            session_id=meta.session_name,
            yaml_path=meta.yaml_file,
        )

        return {
            "status": "completed",
            "final_message": final_message,
            "token_usage": meta.token_usage,
            "output_dir": str(meta.output_dir.resolve()),
        }

    event_queue: queue.Queue[tuple[str, Any]] = queue.Queue()
    done_event = threading.Event()

    def enqueue(event_type: str, data: Any) -> None:
        event_queue.put((event_type, data))

    def worker() -> None:
        try: