diff --git a/docs/api-reference/openapi.json b/docs/api-reference/openapi.json index bb3dfff43..9be049b35 100644 --- a/docs/api-reference/openapi.json +++ b/docs/api-reference/openapi.json @@ -168,6 +168,7 @@ "get": { "tags": [ "Agent", + "Workflow Runs", "Workflows" ], "summary": "Get a run by id", @@ -585,6 +586,224 @@ } } }, + "/v1/workflows/{workflow_id}/runs": { + "get": { + "tags": [ + "Workflow Runs" + ], + "summary": "List runs for a workflow", + "description": "List runs for a specific workflow. Supports filtering by status, search_key, and error_code. All filters are combined with AND logic.", + "operationId": "get_workflow_runs_by_id_v1_workflows__workflow_id__runs_get", + "parameters": [ + { + "name": "workflow_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "description": "Workflow permanent ID. Workflow ID starts with `wpid_`.", + "example": "wpid_123", + "title": "Workflow Id" + }, + "description": "Workflow permanent ID. Workflow ID starts with `wpid_`." + }, + { + "name": "page", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "description": "Page number for pagination.", + "default": 1, + "title": "Page" + }, + "description": "Page number for pagination." + }, + { + "name": "page_size", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "description": "Number of runs to return per page.", + "default": 10, + "title": "Page Size" + }, + "description": "Number of runs to return per page." + }, + { + "name": "status", + "in": "query", + "required": false, + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowRunStatus" + }, + "nullable": true, + "description": "Filter by one or more run statuses.", + "title": "Status" + }, + "description": "Filter by one or more run statuses." + }, + { + "name": "search_key", + "in": "query", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "maxLength": 500, + "description": "Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers.", + "title": "Search Key" + }, + "description": "Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers." + }, + { + "name": "error_code", + "in": "query", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "maxLength": 500, + "description": "Exact-match filter on the error_code field inside each task's errors JSON array.", + "title": "Error Code" + }, + "description": "Exact-match filter on the error_code field inside each task's errors JSON array." + }, + { + "name": "x-api-key", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings.", + "title": "X-Api-Key" + }, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings." + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowRun" + }, + "title": "Response Get Workflow Runs By Id V1 Workflows Workflow Id Runs Get" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + }, + "x-fern-sdk-method-name": "get_workflow_runs_by_id" + } + }, + "/v1/workflows/runs/{workflow_run_id}/retry": { + "post": { + "tags": [ + "Workflow Runs", + "Workflows" + ], + "summary": "Retry workflow run", + "description": "Retry a workflow run using the original run parameters.", + "operationId": "retry_workflow_run_v1_workflows_runs__workflow_run_id__retry_post", + "parameters": [ + { + "name": "workflow_run_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "description": "The id of the workflow run to retry.", + "examples": [ + "wr_123" + ], + "title": "Workflow Run Id" + }, + "description": "The id of the workflow run to retry." + }, + { + "name": "x-api-key", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings.", + "title": "X-Api-Key" + }, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings." + }, + { + "name": "x-max-steps-override", + "in": "header", + "required": false, + "schema": { + "type": "integer", + "nullable": true, + "title": "X-Max-Steps-Override" + } + }, + { + "name": "x-user-agent", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "title": "X-User-Agent" + } + } + ], + "responses": { + "200": { + "description": "Successfully retried workflow run", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/WorkflowRunResponse" + } + } + } + }, + "400": { + "description": "Workflow run is not retryable" + }, + "404": { + "description": "Workflow run not found" + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + }, + "x-fern-sdk-method-name": "retry_workflow_run" + } + }, "/v1/workflows/{workflow_id}/delete": { "post": { "tags": [ @@ -844,6 +1063,7 @@ "get": { "tags": [ "Agent", + "Workflow Runs", "Workflows" ], "summary": "Get run timeline", @@ -1743,6 +1963,7 @@ }, "/v1/run/tasks/login": { "post": { + "x-excluded": true, "tags": [ "Agent" ], @@ -1761,6 +1982,16 @@ "title": "X-Api-Key" }, "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings." + }, + { + "name": "x-user-agent", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "title": "X-User-Agent" + } } ], "requestBody": { @@ -1794,11 +2025,13 @@ } } } - } + }, + "x-fern-sdk-method-name": "login" } }, "/v1/run/tasks/download_files": { "post": { + "x-excluded": true, "tags": [ "Agent" ], @@ -1817,6 +2050,16 @@ "title": "X-Api-Key" }, "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings." + }, + { + "name": "x-user-agent", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "title": "X-User-Agent" + } } ], "requestBody": { @@ -1850,7 +2093,8 @@ } } } - } + }, + "x-fern-sdk-method-name": "download_files" } }, "/v1/scripts": { diff --git a/fern/openapi/skyvern_openapi.json b/fern/openapi/skyvern_openapi.json index b0198ad3d..4e86b0168 100644 --- a/fern/openapi/skyvern_openapi.json +++ b/fern/openapi/skyvern_openapi.json @@ -228,7 +228,8 @@ "get": { "tags": [ "Agent", - "Workflow Runs" + "Workflow Runs", + "Workflows" ], "summary": "Get a run by id", "description": "Get run information (task run, workflow run)", @@ -1589,7 +1590,8 @@ "get": { "tags": [ "Agent", - "Workflow Runs" + "Workflow Runs", + "Workflows" ], "summary": "Get run timeline", "description": "Get timeline for a run (workflow run or task_v2 run)", @@ -1952,6 +1954,224 @@ "x-fern-sdk-method-name": "get_workflow_runs" } }, + "/v1/workflows/{workflow_id}/runs": { + "get": { + "tags": [ + "Workflow Runs" + ], + "summary": "List runs for a workflow", + "description": "List runs for a specific workflow. Supports filtering by status, search_key, and error_code. All filters are combined with AND logic.", + "operationId": "get_workflow_runs_by_id_v1_workflows__workflow_id__runs_get", + "parameters": [ + { + "name": "workflow_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "description": "Workflow permanent ID. Workflow ID starts with `wpid_`.", + "example": "wpid_123", + "title": "Workflow Id" + }, + "description": "Workflow permanent ID. Workflow ID starts with `wpid_`." + }, + { + "name": "page", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "description": "Page number for pagination.", + "default": 1, + "title": "Page" + }, + "description": "Page number for pagination." + }, + { + "name": "page_size", + "in": "query", + "required": false, + "schema": { + "type": "integer", + "minimum": 1, + "description": "Number of runs to return per page.", + "default": 10, + "title": "Page Size" + }, + "description": "Number of runs to return per page." + }, + { + "name": "status", + "in": "query", + "required": false, + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowRunStatus" + }, + "nullable": true, + "description": "Filter by one or more run statuses.", + "title": "Status" + }, + "description": "Filter by one or more run statuses." + }, + { + "name": "search_key", + "in": "query", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "maxLength": 500, + "description": "Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers.", + "title": "Search Key" + }, + "description": "Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers." + }, + { + "name": "error_code", + "in": "query", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "maxLength": 500, + "description": "Exact-match filter on the error_code field inside each task's errors JSON array.", + "title": "Error Code" + }, + "description": "Exact-match filter on the error_code field inside each task's errors JSON array." + }, + { + "name": "x-api-key", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings.", + "title": "X-Api-Key" + }, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings." + } + ], + "responses": { + "200": { + "description": "Successful Response", + "content": { + "application/json": { + "schema": { + "type": "array", + "items": { + "$ref": "#/components/schemas/WorkflowRun" + }, + "title": "Response Get Workflow Runs By Id V1 Workflows Workflow Id Runs Get" + } + } + } + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + }, + "x-fern-sdk-method-name": "get_workflow_runs_by_id" + } + }, + "/v1/workflows/runs/{workflow_run_id}/retry": { + "post": { + "tags": [ + "Workflow Runs", + "Workflows" + ], + "summary": "Retry workflow run", + "description": "Retry a workflow run using the original run parameters.", + "operationId": "retry_workflow_run_v1_workflows_runs__workflow_run_id__retry_post", + "parameters": [ + { + "name": "workflow_run_id", + "in": "path", + "required": true, + "schema": { + "type": "string", + "description": "The id of the workflow run to retry.", + "examples": [ + "wr_123" + ], + "title": "Workflow Run Id" + }, + "description": "The id of the workflow run to retry." + }, + { + "name": "x-api-key", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings.", + "title": "X-Api-Key" + }, + "description": "Skyvern API key for authentication. API key can be found at https://app.skyvern.com/settings." + }, + { + "name": "x-max-steps-override", + "in": "header", + "required": false, + "schema": { + "type": "integer", + "nullable": true, + "title": "X-Max-Steps-Override" + } + }, + { + "name": "x-user-agent", + "in": "header", + "required": false, + "schema": { + "type": "string", + "nullable": true, + "title": "X-User-Agent" + } + } + ], + "responses": { + "200": { + "description": "Successfully retried workflow run", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/WorkflowRunResponse" + } + } + } + }, + "400": { + "description": "Workflow run is not retryable" + }, + "404": { + "description": "Workflow run not found" + }, + "422": { + "description": "Validation Error", + "content": { + "application/json": { + "schema": { + "$ref": "#/components/schemas/HTTPValidationError" + } + } + } + } + }, + "x-fern-sdk-method-name": "retry_workflow_run" + } + }, "/v1/workflows/{workflow_permanent_id}": { "get": { "tags": [ @@ -3464,6 +3684,7 @@ }, "/v1/run/tasks/login": { "post": { + "x-excluded": true, "tags": [ "Agent" ], @@ -3573,6 +3794,7 @@ }, "/v1/run/tasks/download_files": { "post": { + "x-excluded": true, "tags": [ "Agent" ], diff --git a/skyvern/cli/mcp_tools/README.md b/skyvern/cli/mcp_tools/README.md index b52e9a89c..a00cf0158 100644 --- a/skyvern/cli/mcp_tools/README.md +++ b/skyvern/cli/mcp_tools/README.md @@ -98,7 +98,7 @@ Supports Skyvern vault, Bitwarden, 1Password, and Azure Key Vault with automatic `skyvern_state_save`, `skyvern_state_load`, `skyvern_get_session_storage`, `skyvern_set_session_storage`, `skyvern_clear_session_storage`, `skyvern_clear_local_storage`, `skyvern_clipboard_read`, `skyvern_clipboard_write` ### Workflows -`skyvern_workflow_create`, `skyvern_workflow_list`, `skyvern_workflow_get`, `skyvern_workflow_run`, `skyvern_workflow_status`, `skyvern_workflow_update`, `skyvern_workflow_delete`, `skyvern_workflow_cancel`, `skyvern_workflow_update_folder` +`skyvern_workflow_create`, `skyvern_workflow_list`, `skyvern_workflow_get`, `skyvern_workflow_run_list`, `skyvern_workflow_run`, `skyvern_workflow_status`, `skyvern_workflow_retry`, `skyvern_workflow_update`, `skyvern_workflow_delete`, `skyvern_workflow_cancel`, `skyvern_workflow_update_folder` ### Workflow Building Blocks `skyvern_block_schema`, `skyvern_block_validate` -- 23 block types for multi-step automations. diff --git a/skyvern/cli/mcp_tools/__init__.py b/skyvern/cli/mcp_tools/__init__.py index 21ff28c92..a0276e048 100644 --- a/skyvern/cli/mcp_tools/__init__.py +++ b/skyvern/cli/mcp_tools/__init__.py @@ -151,7 +151,9 @@ from .workflow import ( skyvern_workflow_delete, skyvern_workflow_get, skyvern_workflow_list, + skyvern_workflow_retry, skyvern_workflow_run, + skyvern_workflow_run_list, skyvern_workflow_status, skyvern_workflow_update, skyvern_workflow_update_folder, @@ -442,12 +444,14 @@ mcp.tool(tags={"folder"}, annotations=_dest("Delete Folder"))(skyvern_folder_del # -- Workflow management (CRUD + execution, no browser needed) -- mcp.tool(tags={"workflow"}, annotations=_ro("List Workflows"))(size_capped(skyvern_workflow_list)) mcp.tool(tags={"workflow"}, annotations=_ro("Get Workflow"))(size_capped(skyvern_workflow_get)) +mcp.tool(tags={"workflow"}, annotations=_ro("List Workflow Runs"))(size_capped(skyvern_workflow_run_list)) mcp.tool(tags={"workflow"}, annotations=_mut("Create Workflow"))(skyvern_workflow_create) mcp.tool(tags={"workflow"}, annotations=_mut("Update Workflow"))(skyvern_workflow_update) mcp.tool(tags={"workflow"}, annotations=_mut("Move Workflow to Folder"))(skyvern_workflow_update_folder) mcp.tool(tags={"workflow"}, annotations=_dest("Delete Workflow"))(skyvern_workflow_delete) mcp.tool(tags={"workflow"}, annotations=_web_dest("Run Workflow"))(skyvern_workflow_run) mcp.tool(tags={"workflow"}, annotations=_ro("Get Workflow Run Status"))(skyvern_workflow_status) +mcp.tool(tags={"workflow"}, annotations=_web_dest("Retry Workflow Run"))(skyvern_workflow_retry) mcp.tool(tags={"workflow"}, annotations=_dest("Cancel Workflow Run"))(skyvern_workflow_cancel) # -- Schedule management (no browser needed) -- @@ -554,12 +558,14 @@ __all__ = [ # Workflow management "skyvern_workflow_list", "skyvern_workflow_get", + "skyvern_workflow_run_list", "skyvern_workflow_create", "skyvern_workflow_update", "skyvern_workflow_update_folder", "skyvern_workflow_delete", "skyvern_workflow_run", "skyvern_workflow_status", + "skyvern_workflow_retry", "skyvern_workflow_cancel", # Schedule management "skyvern_schedule_list", diff --git a/skyvern/cli/mcp_tools/_validation.py b/skyvern/cli/mcp_tools/_validation.py index 473cb338b..c652027d3 100644 --- a/skyvern/cli/mcp_tools/_validation.py +++ b/skyvern/cli/mcp_tools/_validation.py @@ -77,6 +77,30 @@ def validate_run_id(run_id: str, action: str) -> dict[str, Any] | None: return None +def validate_workflow_run_id(workflow_run_id: str, action: str) -> dict[str, Any] | None: + if "/" in workflow_run_id or "\\" in workflow_run_id: + return make_result( + action, + ok=False, + error=make_error( + ErrorCode.INVALID_INPUT, + "workflow_run_id must not contain path separators", + "Provide a valid workflow run ID (starts with wr_)", + ), + ) + if not workflow_run_id.startswith("wr_"): + return make_result( + action, + ok=False, + error=make_error( + ErrorCode.INVALID_INPUT, + f"Invalid workflow_run_id format: {workflow_run_id!r}", + "Workflow run IDs start with wr_. Use skyvern_workflow_run_list to find valid IDs.", + ), + ) + return None + + def validate_schedule_id(workflow_schedule_id: str, action: str) -> dict[str, Any] | None: if "/" in workflow_schedule_id or "\\" in workflow_schedule_id: return make_result( diff --git a/skyvern/cli/mcp_tools/_workflow_http.py b/skyvern/cli/mcp_tools/_workflow_http.py index 361b90047..149be02f4 100644 --- a/skyvern/cli/mcp_tools/_workflow_http.py +++ b/skyvern/cli/mcp_tools/_workflow_http.py @@ -94,6 +94,61 @@ async def list_workflows_raw( return payload +async def list_workflow_runs_raw( + workflow_id: str, + *, + page: int, + page_size: int, + status: list[str] | None, + search_key: str | None, + error_code: str | None, +) -> list[dict[str, Any]]: + """GET /v1/workflows/{id}/runs — returns raw workflow run dicts for one workflow.""" + skyvern = get_skyvern() + params: dict[str, Any] = { + "page": page, + "page_size": page_size, + } + if status: + params["status"] = status + if search_key is not None: + params["search_key"] = search_key + if error_code is not None: + params["error_code"] = error_code + response = await skyvern._client_wrapper.httpx_client.request( + f"{PUBLIC_WORKFLOW_ROUTE}/{workflow_id}/runs", + method="GET", + params=params, + ) + if response.status_code == 404: + raise NotFoundError(body={"detail": f"Workflow {workflow_id!r} not found"}) + if response.status_code >= 400: + raise RuntimeError(f"HTTP {response.status_code}: {extract_error_detail(response)}") + payload = _decode_success_payload(response, operation="workflow runs list") + if not isinstance(payload, list): + raise RuntimeError(f"Unexpected workflow runs list payload: {type(payload).__name__}") + return payload + + +async def retry_workflow_run_raw(workflow_run_id: str) -> dict[str, Any]: + """POST /v1/workflows/runs/{id}/retry — returns the new workflow run response.""" + skyvern = get_skyvern() + response = await skyvern._client_wrapper.httpx_client.request( + f"{PUBLIC_WORKFLOW_ROUTE}/runs/{workflow_run_id}/retry", + method="POST", + ) + if response.status_code == 404: + raise NotFoundError(body={"detail": f"Workflow run {workflow_run_id!r} not found"}) + if response.status_code == 400: + raise BadRequestError(body={"detail": extract_error_detail(response)}) + if response.status_code >= 400: + raise RuntimeError(f"HTTP {response.status_code}: {extract_error_detail(response)}") + payload = _decode_success_payload(response, operation="retry_workflow_run") + if not isinstance(payload, dict): + raise RuntimeError(f"Unexpected retry_workflow_run payload: {type(payload).__name__}") + return payload + + async def create_workflow_raw( *, json_definition: dict[str, Any] | None, diff --git a/skyvern/cli/mcp_tools/workflow.py b/skyvern/cli/mcp_tools/workflow.py index af26f080b..0436c2c4f 100644 --- a/skyvern/cli/mcp_tools/workflow.py +++ b/skyvern/cli/mcp_tools/workflow.py @@ -25,13 +25,15 @@ from skyvern.utils.yaml_loader import safe_load_no_dates from ._common import ErrorCode, Timer, make_error, make_result from ._session import get_skyvern -from ._validation import validate_folder_id, validate_run_id, validate_workflow_id +from ._validation import validate_folder_id, validate_run_id, validate_workflow_id, validate_workflow_run_id from ._workflow_http import ( coerce_timestamp, create_workflow_raw, get_workflow_by_id, get_workflow_run_status, + list_workflow_runs_raw, list_workflows_raw, + retry_workflow_run_raw, update_workflow_folder_raw, update_workflow_raw, ) @@ -148,6 +150,22 @@ def _get_value(obj: Any, key: str, default: Any = None) -> Any: return getattr(obj, key, default) +def _api_error_body_message(body: Any) -> str: + detail = _get_value(body, "detail") + if isinstance(detail, str): + return detail + if detail is not None: + return json.dumps(_jsonable(detail)) + if isinstance(body, str): + return body + if body is None: + return "Bad request" + try: + return json.dumps(_jsonable(body)) + except TypeError: + return str(body) + + def _get_run_id(run: Any) -> str | None: return _get_value(run, "run_id") or _get_value(run, "workflow_run_id") @@ -895,6 +913,71 @@ async def skyvern_workflow_get( ) +async def skyvern_workflow_run_list( + workflow_id: Annotated[str, "Workflow permanent ID (starts with wpid_)"], + page: Annotated[int, Field(description="Page number (1-based)", ge=1)] = 1, + page_size: Annotated[int, Field(description="Results per page", ge=1, le=100)] = 10, + status: Annotated[list[str] | None, "Filter by one or more workflow run statuses"] = None, + search_key: Annotated[str | None, Field(description="Search workflow run IDs, parameters, and headers")] = None, + error_code: Annotated[str | None, Field(description="Filter by task error code")] = None, +) -> dict[str, Any]: + """List runs for a specific workflow. Use this when you have a workflow permanent ID + and need to browse its run history with pagination and optional filters.""" + if err := validate_workflow_id(workflow_id, "skyvern_workflow_run_list"): + return err + + with Timer() as timer: + try: + requested_page_size = page_size + runs = await list_workflow_runs_raw( + workflow_id, + page=page, + page_size=requested_page_size + 1, + status=status, + search_key=search_key, + error_code=error_code, + ) + timer.mark("sdk") + except NotFoundError: + return make_result( + "skyvern_workflow_run_list", + ok=False, + timing_ms=timer.timing_ms, + error=make_error( + ErrorCode.WORKFLOW_NOT_FOUND, + f"Workflow {workflow_id!r} not found", + "Verify the workflow ID with skyvern_workflow_list", + ), + ) + except Exception as e: + return make_result( + "skyvern_workflow_run_list", + ok=False, + timing_ms=timer.timing_ms, + error=make_error(ErrorCode.API_ERROR, str(e), "Check your API key and workflow run filters"), + ) + + has_more = len(runs) > page_size + runs = runs[:page_size] + + return make_result( + "skyvern_workflow_run_list", + data={ + "workflow_id": workflow_id, + "runs": [_serialize_run_summary(run) for run in runs], + "page": page, + "page_size": page_size, + "count": len(runs), + "has_more": has_more, + "sdk_equivalent": ( + f"await skyvern.get_workflow_runs_by_id(workflow_id={workflow_id!r}, " + f"page={page}, page_size={page_size})" + ), + }, + timing_ms=timer.timing_ms, + ) + + async def skyvern_workflow_create( definition: Annotated[str, "Workflow definition as a YAML or JSON string"], format: Annotated[ # noqa: A002 @@ -1402,3 +1485,59 @@ async def skyvern_workflow_cancel( }, timing_ms=timer.timing_ms, ) + + +async def skyvern_workflow_retry( + workflow_run_id: Annotated[str, "Workflow run ID to retry (wr_...)"], +) -> dict[str, Any]: + """Retry a terminal workflow run. Creates a new workflow run using the original run + parameters and run settings.""" + if err := validate_workflow_run_id(workflow_run_id, "skyvern_workflow_retry"): + return err + + with Timer() as timer: + try: + run = await retry_workflow_run_raw(workflow_run_id) + timer.mark("sdk") + except NotFoundError: + return make_result( + "skyvern_workflow_retry", + ok=False, + timing_ms=timer.timing_ms, + error=make_error( + ErrorCode.RUN_NOT_FOUND, + f"Workflow run {workflow_run_id!r} not found", + "Verify the workflow run ID with skyvern_workflow_run_list or skyvern_workflow_status", + ), + ) + except BadRequestError as e: + return make_result( + "skyvern_workflow_retry", + ok=False, + timing_ms=timer.timing_ms, + error=make_error( + ErrorCode.INVALID_INPUT, + _api_error_body_message(e.body), + "Only terminal workflow runs can be retried.", + ), + ) + except Exception as e: + LOG.error("workflow_retry_failed", workflow_run_id=workflow_run_id, error=str(e)) + return make_result( + "skyvern_workflow_retry", + ok=False, + timing_ms=timer.timing_ms, + error=make_error(ErrorCode.API_ERROR, str(e), "Check the workflow run ID and your API key"), + ) + + retry_run_id = _get_value(run, "run_id") or _get_value(run, "workflow_run_id") + LOG.info("workflow_retried", workflow_run_id=workflow_run_id, retry_run_id=retry_run_id) + return make_result( + "skyvern_workflow_retry", + data={ + "workflow_run_id": workflow_run_id, + "retry_run": _serialize_run_summary(run), + "sdk_equivalent": f"await skyvern.retry_workflow_run({workflow_run_id!r})", + }, + timing_ms=timer.timing_ms, + ) diff --git a/skyvern/cli/skills/skyvern/references/tool-map.md b/skyvern/cli/skills/skyvern/references/tool-map.md index f56e8d0ce..f2def6d65 100644 --- a/skyvern/cli/skills/skyvern/references/tool-map.md +++ b/skyvern/cli/skills/skyvern/references/tool-map.md @@ -54,10 +54,12 @@ | `skyvern_workflow_create` | Create a workflow definition | | `skyvern_workflow_list` | List workflows | | `skyvern_workflow_get` | Get workflow details | +| `skyvern_workflow_run_list` | List runs for a workflow | | `skyvern_workflow_update` | Update a workflow | | `skyvern_workflow_delete` | Delete a workflow | | `skyvern_workflow_run` | Execute a workflow | | `skyvern_workflow_status` | Check run status | +| `skyvern_workflow_retry` | Retry a terminal workflow run | | `skyvern_workflow_cancel` | Cancel a running workflow | ## Workflow blocks diff --git a/skyvern/cli/workflow.py b/skyvern/cli/workflow.py index 96dbc6ca9..8c6d121bf 100644 --- a/skyvern/cli/workflow.py +++ b/skyvern/cli/workflow.py @@ -26,6 +26,12 @@ async def tool_workflow_get(**kwargs: Any) -> dict[str, Any]: return await skyvern_workflow_get(**kwargs) +async def tool_workflow_run_list(**kwargs: Any) -> dict[str, Any]: + from .mcp_tools.workflow import skyvern_workflow_run_list # noqa: PLC0415 + + return await skyvern_workflow_run_list(**kwargs) + + async def tool_workflow_create(**kwargs: Any) -> dict[str, Any]: from .mcp_tools.workflow import skyvern_workflow_create # noqa: PLC0415 @@ -62,6 +68,12 @@ async def tool_workflow_cancel(**kwargs: Any) -> dict[str, Any]: return await skyvern_workflow_cancel(**kwargs) +async def tool_workflow_retry(**kwargs: Any) -> dict[str, Any]: + from .mcp_tools.workflow import skyvern_workflow_retry # noqa: PLC0415 + + return await skyvern_workflow_retry(**kwargs) + + @workflow_app.callback() def workflow_callback( api_key: str | None = typer.Option( @@ -134,6 +146,37 @@ def workflow_get( ) +@workflow_app.command("runs") +def workflow_runs( + workflow_id: str = typer.Option(..., "--id", help="Workflow permanent ID (wpid_...)."), + page: int = typer.Option(1, "--page", min=1, help="Page number (1-based)."), + page_size: int = typer.Option(10, "--page-size", min=1, max=100, help="Results per page."), + status: list[str] | None = typer.Option(None, "--status", help="Filter by workflow run status."), + search_key: str | None = typer.Option(None, "--search-key", help="Search run IDs, parameters, and headers."), + error_code: str | None = typer.Option(None, "--error-code", help="Filter by task error code."), + json_output: bool = typer.Option(False, "--json", help="Output as JSON."), +) -> None: + """List runs for a workflow.""" + + async def _run() -> dict[str, Any]: + return await tool_workflow_run_list( + workflow_id=workflow_id, + page=page, + page_size=page_size, + status=status, + search_key=search_key, + error_code=error_code, + ) + + run_tool( + _run, + json_output=json_output, + hint_on_exception="Check your API key, workflow ID, and run filters.", + action="skyvern_workflow_run_list", + telemetry_tool_name="skyvern_workflow_run_list", + ) + + @workflow_app.command("create") def workflow_create( definition: str = typer.Option( @@ -306,3 +349,22 @@ def workflow_cancel( action="skyvern_workflow_cancel", telemetry_tool_name="skyvern_workflow_cancel", ) + + +@workflow_app.command("retry") +def workflow_retry( + workflow_run_id: str = typer.Option(..., "--run-id", help="Workflow run ID to retry (wr_...)."), + json_output: bool = typer.Option(False, "--json", help="Output as JSON."), +) -> None: + """Retry a terminal workflow run.""" + + async def _run() -> dict[str, Any]: + return await tool_workflow_retry(workflow_run_id=workflow_run_id) + + run_tool( + _run, + json_output=json_output, + hint_on_exception="Check the workflow run ID and API key.", + action="skyvern_workflow_retry", + telemetry_tool_name="skyvern_workflow_retry", + ) diff --git a/skyvern/client/client.py b/skyvern/client/client.py index 74bf507d3..d7411d832 100644 --- a/skyvern/client/client.py +++ b/skyvern/client/client.py @@ -470,6 +470,53 @@ class Skyvern: ) return _response.data + def retry_workflow_run( + self, + workflow_run_id: str, + *, + max_steps_override: typing.Optional[int] = None, + user_agent: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> WorkflowRunResponse: + """ + Retry a workflow run using the original run parameters. + + Parameters + ---------- + workflow_run_id : str + The id of the workflow run to retry. + + max_steps_override : typing.Optional[int] + + user_agent : typing.Optional[str] + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + WorkflowRunResponse + Successfully retried workflow run + + Examples + -------- + from skyvern import Skyvern + + client = Skyvern( + api_key="YOUR_API_KEY", + ) + client.retry_workflow_run( + workflow_run_id="wr_123", + ) + """ + _response = self._raw_client.retry_workflow_run( + workflow_run_id, + max_steps_override=max_steps_override, + user_agent=user_agent, + request_options=request_options, + ) + return _response.data + def get_run(self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> GetRunResponse: """ Get run information (task run, workflow run) @@ -1259,6 +1306,59 @@ class Skyvern: ) return _response.data + def get_workflow_runs_by_id( + self, + workflow_id: str, + *, + page: typing.Optional[int] = None, + page_size: typing.Optional[int] = None, + status: typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] = None, + search_key: typing.Optional[str] = None, + error_code: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> typing.List[WorkflowRun]: + """ + List runs for a specific workflow. + + Parameters + ---------- + workflow_id : str + Workflow permanent ID. Workflow ID starts with `wpid_`. + + page : typing.Optional[int] + Page number for pagination. + + page_size : typing.Optional[int] + Number of runs to return per page. + + status : typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] + Filter by one or more run statuses. + + search_key : typing.Optional[str] + Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers. + + error_code : typing.Optional[str] + Exact-match filter on the error_code field inside each task's errors JSON array. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + typing.List[WorkflowRun] + Successful Response + """ + _response = self._raw_client.get_workflow_runs_by_id( + workflow_id, + page=page, + page_size=page_size, + status=status, + search_key=search_key, + error_code=error_code, + request_options=request_options, + ) + return _response.data + def get_workflow( self, workflow_permanent_id: str, @@ -2936,6 +3036,61 @@ class AsyncSkyvern: ) return _response.data + async def retry_workflow_run( + self, + workflow_run_id: str, + *, + max_steps_override: typing.Optional[int] = None, + user_agent: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> WorkflowRunResponse: + """ + Retry a workflow run using the original run parameters. + + Parameters + ---------- + workflow_run_id : str + The id of the workflow run to retry. + + max_steps_override : typing.Optional[int] + + user_agent : typing.Optional[str] + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + WorkflowRunResponse + Successfully retried workflow run + + Examples + -------- + import asyncio + + from skyvern import AsyncSkyvern + + client = AsyncSkyvern( + api_key="YOUR_API_KEY", + ) + + + async def main() -> None: + await client.retry_workflow_run( + workflow_run_id="wr_123", + ) + + + asyncio.run(main()) + """ + _response = await self._raw_client.retry_workflow_run( + workflow_run_id, + max_steps_override=max_steps_override, + user_agent=user_agent, + request_options=request_options, + ) + return _response.data + async def get_run(self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> GetRunResponse: """ Get run information (task run, workflow run) @@ -3861,6 +4016,59 @@ class AsyncSkyvern: ) return _response.data + async def get_workflow_runs_by_id( + self, + workflow_id: str, + *, + page: typing.Optional[int] = None, + page_size: typing.Optional[int] = None, + status: typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] = None, + search_key: typing.Optional[str] = None, + error_code: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> typing.List[WorkflowRun]: + """ + List runs for a specific workflow. + + Parameters + ---------- + workflow_id : str + Workflow permanent ID. Workflow ID starts with `wpid_`. + + page : typing.Optional[int] + Page number for pagination. + + page_size : typing.Optional[int] + Number of runs to return per page. + + status : typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] + Filter by one or more run statuses. + + search_key : typing.Optional[str] + Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers. + + error_code : typing.Optional[str] + Exact-match filter on the error_code field inside each task's errors JSON array. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + typing.List[WorkflowRun] + Successful Response + """ + _response = await self._raw_client.get_workflow_runs_by_id( + workflow_id, + page=page, + page_size=page_size, + status=status, + search_key=search_key, + error_code=error_code, + request_options=request_options, + ) + return _response.data + async def get_workflow( self, workflow_permanent_id: str, diff --git a/skyvern/client/raw_client.py b/skyvern/client/raw_client.py index d3f7fc15b..ef984144f 100644 --- a/skyvern/client/raw_client.py +++ b/skyvern/client/raw_client.py @@ -463,6 +463,92 @@ class RawSkyvern: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + def retry_workflow_run( + self, + workflow_run_id: str, + *, + max_steps_override: typing.Optional[int] = None, + user_agent: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> HttpResponse[WorkflowRunResponse]: + """ + Retry a workflow run using the original run parameters. + + Parameters + ---------- + workflow_run_id : str + The id of the workflow run to retry. + + max_steps_override : typing.Optional[int] + + user_agent : typing.Optional[str] + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + HttpResponse[WorkflowRunResponse] + Successfully retried workflow run + """ + _response = self._client_wrapper.httpx_client.request( + f"v1/workflows/runs/{jsonable_encoder(workflow_run_id)}/retry", + method="POST", + headers={ + "x-max-steps-override": str(max_steps_override) if max_steps_override is not None else None, + "x-user-agent": str(user_agent) if user_agent is not None else None, + }, + request_options=request_options, + omit=OMIT, + ) + try: + if 200 <= _response.status_code < 300: + _data = typing.cast( + WorkflowRunResponse, + parse_obj_as( + type_=WorkflowRunResponse, # type: ignore + object_=_response.json(), + ), + ) + return HttpResponse(response=_response, data=_data) + if _response.status_code == 400: + raise BadRequestError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + if _response.status_code == 404: + raise NotFoundError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + if _response.status_code == 422: + raise UnprocessableEntityError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + def get_run( self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> HttpResponse[GetRunResponse]: @@ -1679,6 +1765,86 @@ class RawSkyvern: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + def get_workflow_runs_by_id( + self, + workflow_id: str, + *, + page: typing.Optional[int] = None, + page_size: typing.Optional[int] = None, + status: typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] = None, + search_key: typing.Optional[str] = None, + error_code: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> HttpResponse[typing.List[WorkflowRun]]: + """ + List runs for a specific workflow. + + Parameters + ---------- + workflow_id : str + Workflow permanent ID. Workflow ID starts with `wpid_`. + + page : typing.Optional[int] + Page number for pagination. + + page_size : typing.Optional[int] + Number of runs to return per page. + + status : typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] + Filter by one or more run statuses. + + search_key : typing.Optional[str] + Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers. + + error_code : typing.Optional[str] + Exact-match filter on the error_code field inside each task's errors JSON array. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + HttpResponse[typing.List[WorkflowRun]] + Successful Response + """ + _response = self._client_wrapper.httpx_client.request( + f"v1/workflows/{jsonable_encoder(workflow_id)}/runs", + method="GET", + params={ + "page": page, + "page_size": page_size, + "status": status, + "search_key": search_key, + "error_code": error_code, + }, + request_options=request_options, + ) + try: + if 200 <= _response.status_code < 300: + _data = typing.cast( + typing.List[WorkflowRun], + parse_obj_as( + type_=typing.List[WorkflowRun], # type: ignore + object_=_response.json(), + ), + ) + return HttpResponse(response=_response, data=_data) + if _response.status_code == 422: + raise UnprocessableEntityError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + def get_workflow( self, workflow_permanent_id: str, @@ -3872,6 +4038,92 @@ class AsyncRawSkyvern: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + async def retry_workflow_run( + self, + workflow_run_id: str, + *, + max_steps_override: typing.Optional[int] = None, + user_agent: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> AsyncHttpResponse[WorkflowRunResponse]: + """ + Retry a workflow run using the original run parameters. + + Parameters + ---------- + workflow_run_id : str + The id of the workflow run to retry. + + max_steps_override : typing.Optional[int] + + user_agent : typing.Optional[str] + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + AsyncHttpResponse[WorkflowRunResponse] + Successfully retried workflow run + """ + _response = await self._client_wrapper.httpx_client.request( + f"v1/workflows/runs/{jsonable_encoder(workflow_run_id)}/retry", + method="POST", + headers={ + "x-max-steps-override": str(max_steps_override) if max_steps_override is not None else None, + "x-user-agent": str(user_agent) if user_agent is not None else None, + }, + request_options=request_options, + omit=OMIT, + ) + try: + if 200 <= _response.status_code < 300: + _data = typing.cast( + WorkflowRunResponse, + parse_obj_as( + type_=WorkflowRunResponse, # type: ignore + object_=_response.json(), + ), + ) + return AsyncHttpResponse(response=_response, data=_data) + if _response.status_code == 400: + raise BadRequestError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + if _response.status_code == 404: + raise NotFoundError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + if _response.status_code == 422: + raise UnprocessableEntityError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + async def get_run( self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None ) -> AsyncHttpResponse[GetRunResponse]: @@ -5088,6 +5340,86 @@ class AsyncRawSkyvern: raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + async def get_workflow_runs_by_id( + self, + workflow_id: str, + *, + page: typing.Optional[int] = None, + page_size: typing.Optional[int] = None, + status: typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] = None, + search_key: typing.Optional[str] = None, + error_code: typing.Optional[str] = None, + request_options: typing.Optional[RequestOptions] = None, + ) -> AsyncHttpResponse[typing.List[WorkflowRun]]: + """ + List runs for a specific workflow. + + Parameters + ---------- + workflow_id : str + Workflow permanent ID. Workflow ID starts with `wpid_`. + + page : typing.Optional[int] + Page number for pagination. + + page_size : typing.Optional[int] + Number of runs to return per page. + + status : typing.Optional[typing.Union[WorkflowRunStatus, typing.Sequence[WorkflowRunStatus]]] + Filter by one or more run statuses. + + search_key : typing.Optional[str] + Case-insensitive substring search across workflow run ID, parameter key, parameter description, run parameter value, and extra HTTP headers. + + error_code : typing.Optional[str] + Exact-match filter on the error_code field inside each task's errors JSON array. + + request_options : typing.Optional[RequestOptions] + Request-specific configuration. + + Returns + ------- + AsyncHttpResponse[typing.List[WorkflowRun]] + Successful Response + """ + _response = await self._client_wrapper.httpx_client.request( + f"v1/workflows/{jsonable_encoder(workflow_id)}/runs", + method="GET", + params={ + "page": page, + "page_size": page_size, + "status": status, + "search_key": search_key, + "error_code": error_code, + }, + request_options=request_options, + ) + try: + if 200 <= _response.status_code < 300: + _data = typing.cast( + typing.List[WorkflowRun], + parse_obj_as( + type_=typing.List[WorkflowRun], # type: ignore + object_=_response.json(), + ), + ) + return AsyncHttpResponse(response=_response, data=_data) + if _response.status_code == 422: + raise UnprocessableEntityError( + headers=dict(_response.headers), + body=typing.cast( + typing.Optional[typing.Any], + parse_obj_as( + type_=typing.Optional[typing.Any], # type: ignore + object_=_response.json(), + ), + ), + ) + _response_json = _response.json() + except JSONDecodeError: + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response.text) + raise ApiError(status_code=_response.status_code, headers=dict(_response.headers), body=_response_json) + async def get_workflow( self, workflow_permanent_id: str, diff --git a/skyvern/forge/agent_functions.py b/skyvern/forge/agent_functions.py index 1d694a27d..5ebe09371 100644 --- a/skyvern/forge/agent_functions.py +++ b/skyvern/forge/agent_functions.py @@ -504,6 +504,19 @@ class AgentFunction: """Persist per-run analytics metadata. OSS builds have no sidecar table.""" return None + async def get_workflow_run_metadata( + self, + *, + workflow_run_id: str, + organization_id: str, + ) -> dict[str, str] | None: + """Fetch per-run analytics metadata. OSS builds have no sidecar table.""" + return None + + async def is_block_scoped_workflow_run(self, workflow_run: "WorkflowRun") -> bool: + """Return whether this workflow run was created for scoped block execution.""" + return workflow_run.debug_session_id is not None + # Phrases that indicate a magic-link confirmation page meant to be closed. # Keep lowercase; matching is case-insensitive. MAGIC_LINK_CLOSE_SIGNALS: tuple[str, ...] = ( diff --git a/skyvern/forge/sdk/db/repositories/debug.py b/skyvern/forge/sdk/db/repositories/debug.py index 21ddec825..ed59819a7 100644 --- a/skyvern/forge/sdk/db/repositories/debug.py +++ b/skyvern/forge/sdk/db/repositories/debug.py @@ -88,6 +88,23 @@ class DebugRepository(BaseRepository): return BlockRun.model_validate(model) if model else None + @db_operation("has_block_run_for_workflow_run") + async def has_block_run_for_workflow_run( + self, + *, + organization_id: str, + workflow_run_id: str, + ) -> bool: + async with self.Session() as session: + model = await session.scalar( + select(BlockRunModel.workflow_run_id) + .filter_by(organization_id=organization_id) + .filter_by(workflow_run_id=workflow_run_id) + .limit(1) + ) + + return model is not None + @db_operation("create_block_run") async def create_block_run( self, diff --git a/skyvern/forge/sdk/db/repositories/workflow_runs.py b/skyvern/forge/sdk/db/repositories/workflow_runs.py index 05d42f95f..1f438edfc 100644 --- a/skyvern/forge/sdk/db/repositories/workflow_runs.py +++ b/skyvern/forge/sdk/db/repositories/workflow_runs.py @@ -966,6 +966,7 @@ class WorkflowRunsRepository(BaseRepository): status: list[WorkflowRunStatus] | None = None, search_key: str | None = None, error_code: str | None = None, + exclude_child_runs: bool = False, ) -> list[WorkflowRun]: """ Get runs for a workflow, with optional `search_key` on run ID, parameter key/description/value, @@ -980,6 +981,8 @@ class WorkflowRunsRepository(BaseRepository): .filter(WorkflowRunModel.organization_id == organization_id) .filter(WorkflowRunModel.copilot_session_id.is_(None)) ) + if exclude_child_runs: + query = query.filter(WorkflowRunModel.parent_workflow_run_id.is_(None)) query = self._apply_search_key_filter(query, search_key) query = self._apply_error_code_filter(query, error_code) if status: diff --git a/skyvern/forge/sdk/routes/agent_protocol.py b/skyvern/forge/sdk/routes/agent_protocol.py index ec5b7d062..7ed3fd022 100644 --- a/skyvern/forge/sdk/routes/agent_protocol.py +++ b/skyvern/forge/sdk/routes/agent_protocol.py @@ -32,6 +32,7 @@ from skyvern.config import settings from skyvern.exceptions import ( MissingBrowserAddressError, SkyvernHTTPException, + WorkflowNotFound, ) from skyvern.forge import app from skyvern.forge.prompts import prompt_engine @@ -510,7 +511,7 @@ async def run_workflow( @base_router.get( "/runs/{run_id}", - tags=["Agent", "Workflow Runs"], + tags=["Agent", "Workflow Runs", "Workflows"], response_model=RunResponse, description="Get run information (task run, workflow run)", summary="Get a run by id", @@ -1887,7 +1888,7 @@ async def retry_run_webhook( @base_router.get( "/runs/{run_id}/timeline", - tags=["Agent", "Workflow Runs"], + tags=["Agent", "Workflow Runs", "Workflows"], response_model=list[WorkflowRunTimeline], openapi_extra={ "x-fern-sdk-method-name": "get_run_timeline", @@ -2270,6 +2271,208 @@ async def _continue_workflow_run(workflow_run_id: str, organization_id: str) -> await app.WORKFLOW_SERVICE.mark_workflow_run_as_running(workflow_run_id) +def _workflow_request_body_from_existing_run( + workflow_run: WorkflowRun, + parameters: dict[str, Any] | None = None, + run_metadata: dict[str, str] | None = None, +) -> WorkflowRequestBody: + return WorkflowRequestBody( + data=parameters, + proxy_location=workflow_run.proxy_location, + webhook_callback_url=workflow_run.webhook_callback_url, + totp_verification_url=workflow_run.totp_verification_url, + totp_identifier=workflow_run.totp_identifier, + browser_session_id=workflow_run.browser_session_id, + browser_profile_id=workflow_run.browser_profile_id, + max_screenshot_scrolls=workflow_run.max_screenshot_scrolls, + max_elapsed_time_minutes=getattr(workflow_run, "max_elapsed_time_minutes", None), + extra_http_headers=workflow_run.extra_http_headers, + cdp_connect_headers=workflow_run.cdp_connect_headers, + browser_address=workflow_run.browser_address, + run_with=workflow_run.run_with, + ai_fallback=workflow_run.ai_fallback, + run_metadata=run_metadata, + ) + + +def _workflow_run_request_from_workflow_request( + *, + workflow_id: str, + title: str | None, + workflow_request: WorkflowRequestBody, +) -> WorkflowRunRequest: + return WorkflowRunRequest( + workflow_id=workflow_id, + title=title, + parameters=workflow_request.data, + proxy_location=workflow_request.proxy_location, + webhook_url=workflow_request.webhook_callback_url, + totp_url=workflow_request.totp_verification_url, + totp_identifier=workflow_request.totp_identifier, + browser_session_id=workflow_request.browser_session_id, + browser_profile_id=workflow_request.browser_profile_id, + max_screenshot_scrolls=workflow_request.max_screenshot_scrolls, + max_elapsed_time_minutes=getattr(workflow_request, "max_elapsed_time_minutes", None), + extra_http_headers=workflow_request.extra_http_headers, + cdp_connect_headers=workflow_request.cdp_connect_headers, + browser_address=workflow_request.browser_address, + run_with=workflow_request.run_with, + ai_fallback=workflow_request.ai_fallback, + run_metadata=workflow_request.run_metadata, + ) + + +@base_router.post( + "/workflows/runs/{workflow_run_id}/retry", + tags=["Workflow Runs", "Workflows"], + response_model=WorkflowRunResponse, + openapi_extra={ + "x-fern-sdk-method-name": "retry_workflow_run", + }, + description="Retry a workflow run using the original run parameters.", + summary="Retry workflow run", + responses={ + 200: {"description": "Successfully retried workflow run"}, + 400: {"description": "Workflow run is not retryable"}, + 404: {"description": "Workflow run not found"}, + }, +) +@base_router.post("/workflows/runs/{workflow_run_id}/retry/", include_in_schema=False) +async def retry_workflow_run( + request: Request, + background_tasks: BackgroundTasks, + workflow_run_id: str = Path(..., description="The id of the workflow run to retry.", examples=["wr_123"]), + current_org: Organization = Depends(org_auth_service.get_current_org), + x_api_key: Annotated[str | None, Header()] = None, + x_max_steps_override: Annotated[int | None, Header()] = None, + x_user_agent: Annotated[str | None, Header()] = None, +) -> WorkflowRunResponse: + analytics.capture("skyvern-oss-agent-workflow-run-retry") + original_workflow_run = await app.DATABASE.workflow_runs.get_workflow_run( + workflow_run_id=workflow_run_id, + organization_id=current_org.organization_id, + ) + + if not original_workflow_run: + raise HTTPException( + status_code=http_status.HTTP_404_NOT_FOUND, + detail=f"Workflow run not found {workflow_run_id}", + ) + + if not original_workflow_run.status.is_final(): + raise HTTPException( + status_code=http_status.HTTP_400_BAD_REQUEST, + detail="Only terminal workflow runs can be retried", + ) + + is_block_scoped_run = await app.AGENT_FUNCTION.is_block_scoped_workflow_run(original_workflow_run) + if not is_block_scoped_run: + is_block_scoped_run = await app.DATABASE.debug.has_block_run_for_workflow_run( + organization_id=current_org.organization_id, + workflow_run_id=original_workflow_run.workflow_run_id, + ) + if is_block_scoped_run: + raise HTTPException( + status_code=http_status.HTTP_400_BAD_REQUEST, + detail="Block-scoped workflow runs cannot be retried with this endpoint", + ) + + await PermissionCheckerFactory.get_instance().check( + current_org, browser_session_id=original_workflow_run.browser_session_id + ) + await app.RATE_LIMITER.rate_limit_submit_run(current_org.organization_id) + + try: + original_workflow = await app.WORKFLOW_SERVICE.get_workflow( + workflow_id=original_workflow_run.workflow_id, + organization_id=None, + ) + except WorkflowNotFound as e: + raise HTTPException( + status_code=http_status.HTTP_404_NOT_FOUND, + detail=f"Workflow not found for run {workflow_run_id}", + ) from e + + template = original_workflow.organization_id != current_org.organization_id + original_workflow_run_parameter_tuples = await app.DATABASE.workflow_runs.get_workflow_run_parameters( + workflow_run_id=original_workflow_run.workflow_run_id, + ) + original_workflow_run_parameters = { + workflow_parameter.key: workflow_run_parameter.value + for workflow_parameter, workflow_run_parameter in original_workflow_run_parameter_tuples + } + original_run_metadata = None + try: + original_run_metadata = await app.AGENT_FUNCTION.get_workflow_run_metadata( + workflow_run_id=original_workflow_run.workflow_run_id, + organization_id=current_org.organization_id, + ) + except Exception: + LOG.warning( + "Failed to fetch workflow run metadata for retry; continuing without metadata", + workflow_run_id=original_workflow_run.workflow_run_id, + organization_id=current_org.organization_id, + exc_info=True, + ) + legacy_workflow_request = _workflow_request_body_from_existing_run( + workflow_run=original_workflow_run, + parameters=original_workflow_run_parameters, + run_metadata=original_run_metadata, + ) + + context = skyvern_context.ensure_context() + trigger_type = workflow_run_trigger_type_from_user_agent(x_user_agent) + try: + workflow_run = await workflow_service.run_workflow( + workflow_id=original_workflow_run.workflow_permanent_id, + organization=current_org, + workflow_request=legacy_workflow_request, + template=template, + version=original_workflow.version, + max_steps=x_max_steps_override, + api_key=x_api_key, + request_id=context.request_id, + request=request, + background_tasks=background_tasks, + trigger_type=trigger_type, + ignore_inherited_workflow_system_prompt=original_workflow_run.ignore_inherited_workflow_system_prompt, + ) + except MissingBrowserAddressError as e: + raise HTTPException(status_code=400, detail=str(e)) from e + + if settings.OTEL_ENABLED: + span = trace.get_current_span() + if span: + if workflow_run.workflow_run_id: + span.set_attribute("workflow_run_id", workflow_run.workflow_run_id) + if workflow_run.workflow_id: + span.set_attribute("workflow_id", workflow_run.workflow_id) + + workflow_run_request_hydrated = _workflow_run_request_from_workflow_request( + workflow_id=original_workflow_run.workflow_permanent_id, + title=original_workflow.title, + workflow_request=legacy_workflow_request, + ) + + return WorkflowRunResponse( + run_id=workflow_run.workflow_run_id, + run_type=RunType.workflow_run, + status=str(workflow_run.status), + output=None, + failure_reason=workflow_run.failure_reason, + created_at=workflow_run.created_at, + modified_at=workflow_run.modified_at, + run_request=workflow_run_request_hydrated, + downloaded_files=None, + recording_url=None, + app_url=f"{settings.SKYVERN_APP_URL.rstrip('/')}/runs/{workflow_run.workflow_run_id}", + browser_session_id=workflow_run.browser_session_id, + browser_profile_id=workflow_run.browser_profile_id, + run_with=workflow_run.run_with, + ai_fallback=workflow_run.ai_fallback, + ) + + @legacy_base_router.post( "/workflows/runs/{workflow_run_id}/cancel", tags=["agent"], @@ -2803,28 +3006,55 @@ async def get_workflow_runs( ) -@legacy_base_router.get( +_WORKFLOW_RUNS_BY_ID_DESCRIPTION = ( + "List runs for a specific workflow.\n\n" + "Supports filtering by **status**, **search_key**, and **error_code**. " + "All filters are combined with **AND** logic.\n\n" + "### search_key\n\n" + "Case-insensitive substring search across: workflow run ID, " + "parameter key, parameter description, run parameter value, " + "and extra HTTP headers. Soft-deleted parameter definitions are excluded.\n\n" + "### error_code\n\n" + "Exact-match filter on the `error_code` field inside each task's `errors` JSON array. " + "A run matches if any of its tasks contains an error with a matching `error_code`." +) + + +async def _get_workflow_runs_by_id( + *, + workflow_id: str, + organization_id: str, + page: int, + page_size: int, + status: list[WorkflowRunStatus] | None, + search_key: str | None, + error_code: str | None, + exclude_child_runs: bool, +) -> list[WorkflowRun]: + analytics.capture("skyvern-oss-agent-workflow-runs-get") + return await app.WORKFLOW_SERVICE.get_workflow_runs_for_workflow_permanent_id( + workflow_permanent_id=workflow_id, + organization_id=organization_id, + page=page, + page_size=page_size, + status=status, + search_key=search_key, + error_code=error_code, + exclude_child_runs=exclude_child_runs, + ) + + +@base_router.get( "/workflows/{workflow_id}/runs", response_model=list[WorkflowRun], - tags=["agent"], - description=( - "List runs for a specific workflow.\n\n" - "Supports filtering by **status**, **search_key**, and **error_code**. " - "All filters are combined with **AND** logic.\n\n" - "### search_key\n\n" - "Case-insensitive substring search across: workflow run ID, " - "parameter key, parameter description, run parameter value, " - "and extra HTTP headers. Soft-deleted parameter definitions are excluded.\n\n" - "### error_code\n\n" - "Exact-match filter on the `error_code` field inside each task's `errors` JSON array. " - "A run matches if any of its tasks contains an error with a matching `error_code`." - ), + tags=["Workflow Runs", "Workflows"], + description=_WORKFLOW_RUNS_BY_ID_DESCRIPTION, summary="List runs for a workflow", openapi_extra={ "x-fern-sdk-method-name": "get_workflow_runs_by_id", }, ) -@legacy_base_router.get( +@base_router.get( "/workflows/{workflow_id}/runs/", response_model=list[WorkflowRun], include_in_schema=False, @@ -2860,33 +3090,78 @@ async def get_workflow_runs_by_id( """ List runs for a specific workflow permanent id. + The public API excludes child workflow runs so workflow histories only show top-level runs. All filters (**status**, **search_key**, **error_code**) are combined with AND logic. - - **search_key** performs a case-insensitive substring match across: - - `workflow_run_id` — the unique run identifier - - Parameter **key** and **description** from workflow parameter definitions (soft-deleted parameters excluded) - - Run parameter **value** — the actual value supplied when the run was created - - `extra_http_headers` — searched as raw JSON text - - **error_code** performs an exact match against the `error_code` field in the `errors` - JSON array on each task. A run matches if *any* of its tasks has a matching error code. - - **Examples:** - - All failed runs: `?status=failed` - - Failed runs with a specific error: `?status=failed&error_code=INVALID_CREDENTIALS` - - Runs matching a parameter value: `?search_key=https://example.com` - - Runs matching a run ID: `?search_key=wr_abc123` - - Combined: `?status=failed&error_code=LOGIN_FAILED&search_key=my_credential` """ - analytics.capture("skyvern-oss-agent-workflow-runs-get") - return await app.WORKFLOW_SERVICE.get_workflow_runs_for_workflow_permanent_id( - workflow_permanent_id=workflow_id, + return await _get_workflow_runs_by_id( + workflow_id=workflow_id, organization_id=current_org.organization_id, page=page, page_size=page_size, status=status, search_key=search_key, error_code=error_code, + exclude_child_runs=True, + ) + + +@legacy_base_router.get( + "/workflows/{workflow_id}/runs", + response_model=list[WorkflowRun], + tags=["agent"], + description=_WORKFLOW_RUNS_BY_ID_DESCRIPTION, + summary="List runs for a workflow", + openapi_extra={ + "x-fern-sdk-method-name": "get_workflow_runs_by_id", + }, +) +@legacy_base_router.get( + "/workflows/{workflow_id}/runs/", + response_model=list[WorkflowRun], + include_in_schema=False, +) +async def get_workflow_runs_by_id_legacy( + workflow_id: str, + page: int = Query(1, ge=1, description="Page number for pagination."), + page_size: int = Query(10, ge=1, description="Number of runs to return per page."), + status: Annotated[list[WorkflowRunStatus] | None, Query(description="Filter by one or more run statuses.")] = None, + search_key: str | None = Query( + None, + max_length=500, + description=( + "Case-insensitive substring search across: workflow run ID, " + "parameter key, parameter description, run parameter value, " + "and extra HTTP headers. A run is returned if any of these fields match. " + "Soft-deleted parameter definitions are excluded from key/description matching." + ), + examples=["login_url", "credential_value", "wr_abc123"], + ), + error_code: str | None = Query( + None, + max_length=500, + description=( + "Exact-match filter on the error_code field inside each task's errors JSON array. " + "A run matches if any of its tasks contains an error with a matching error_code. " + "Error codes are user-defined strings set during workflow execution." + ), + examples=["INVALID_CREDENTIALS", "LOGIN_FAILED", "CAPTCHA_DETECTED"], + ), + current_org: Organization = Depends(org_auth_service.get_current_org), +) -> list[WorkflowRun]: + """ + List runs for a specific workflow permanent id using legacy endpoint behavior. + + Legacy callers keep seeing child workflow runs to avoid changing existing API behavior. + """ + return await _get_workflow_runs_by_id( + workflow_id=workflow_id, + organization_id=current_org.organization_id, + page=page, + page_size=page_size, + status=status, + search_key=search_key, + error_code=error_code, + exclude_child_runs=False, ) diff --git a/skyvern/forge/sdk/routes/run_blocks.py b/skyvern/forge/sdk/routes/run_blocks.py index f5d5ff4cd..3ad0c01e1 100644 --- a/skyvern/forge/sdk/routes/run_blocks.py +++ b/skyvern/forge/sdk/routes/run_blocks.py @@ -125,6 +125,7 @@ async def _run_workflow_and_build_response( tags=["Agent"], response_model=WorkflowRunResponse, openapi_extra={ + "x-excluded": True, "x-fern-sdk-method-name": "login", "x-fern-examples": [ { @@ -291,6 +292,7 @@ async def login( tags=["Agent"], response_model=WorkflowRunResponse, openapi_extra={ + "x-excluded": True, "x-fern-sdk-method-name": "download_files", "x-fern-examples": [ { diff --git a/skyvern/forge/sdk/workflow/service.py b/skyvern/forge/sdk/workflow/service.py index 9d938c06c..217ca5082 100644 --- a/skyvern/forge/sdk/workflow/service.py +++ b/skyvern/forge/sdk/workflow/service.py @@ -4207,6 +4207,7 @@ class WorkflowService: status: list[WorkflowRunStatus] | None = None, search_key: str | None = None, error_code: str | None = None, + exclude_child_runs: bool = False, ) -> list[WorkflowRun]: return await app.DATABASE.workflow_runs.get_workflow_runs_for_workflow_permanent_id( workflow_permanent_id=workflow_permanent_id, @@ -4216,6 +4217,7 @@ class WorkflowService: status=status, search_key=search_key, error_code=error_code, + exclude_child_runs=exclude_child_runs, ) async def get_workflow_runs_for_browser_session( diff --git a/tests/unit/forge/sdk/db/test_workflow_runs_repository.py b/tests/unit/forge/sdk/db/test_workflow_runs_repository.py index b6deffb17..55b4d64ad 100644 --- a/tests/unit/forge/sdk/db/test_workflow_runs_repository.py +++ b/tests/unit/forge/sdk/db/test_workflow_runs_repository.py @@ -306,6 +306,7 @@ async def test_workflow_run_history_queries_exclude_copilot_session_runs() -> No await repo.get_workflow_runs_for_workflow_permanent_id( workflow_permanent_id="wpid_test", organization_id="o_test", + exclude_child_runs=True, ) assert len(captured_queries) == 2 @@ -313,6 +314,31 @@ async def test_workflow_run_history_queries_exclude_copilot_session_runs() -> No where_clause = _where_clause_sql(query) assert "workflow_runs.copilot_session_id IS NULL" in where_clause _assert_not_filtering_copilot_authored_workflows(where_clause) + workflow_runs_for_workflow_clause = _where_clause_sql(captured_queries[1]) + assert "workflow_runs.parent_workflow_run_id IS NULL" in workflow_runs_for_workflow_clause + + +@pytest.mark.asyncio +async def test_get_workflow_runs_for_workflow_permanent_id_keeps_child_runs_by_default() -> None: + captured: dict[str, Any] = {} + + async def _execute(query): + captured["query"] = query + return _EmptyExecuteResult() + + session = MagicMock() + session.execute = AsyncMock(side_effect=_execute) + + repo = WorkflowRunsRepository(session_factory=lambda: _SessionContext(session), debug_enabled=False) + + await repo.get_workflow_runs_for_workflow_permanent_id( + workflow_permanent_id="wpid_test", + organization_id="o_test", + ) + + where_clause = _where_clause_sql(captured["query"]) + assert "workflow_runs.parent_workflow_run_id IS NULL" not in where_clause + assert "workflow_runs.copilot_session_id IS NULL" in where_clause @pytest.mark.asyncio diff --git a/tests/unit/test_mcp_workflow_route_prefixes.py b/tests/unit/test_mcp_workflow_route_prefixes.py index dee6972bb..5e90ebd24 100644 --- a/tests/unit/test_mcp_workflow_route_prefixes.py +++ b/tests/unit/test_mcp_workflow_route_prefixes.py @@ -1,8 +1,8 @@ """Guard test for MCP workflow raw-HTTP route prefix choices. Keeps public vs internal route prefix assignments stable: -- list / create / update / update_folder → ``v1/workflows`` (public, mirrors the - Fern-generated raw client at ``skyvern/client/raw_client.py``). +- list / create / update / update_folder / retry → ``v1/workflows`` (public, + mirrors the Fern-generated raw client at ``skyvern/client/raw_client.py``). - get / run-status → ``api/v1/workflows`` (internal, used only where no public Fern SDK equivalent exists yet). @@ -17,11 +17,14 @@ from __future__ import annotations import ast import inspect +from types import SimpleNamespace from unittest.mock import AsyncMock, MagicMock import pytest from skyvern.cli.mcp_tools import _workflow_http +from skyvern.cli.mcp_tools import workflow as workflow_tools +from skyvern.client.errors import BadRequestError @pytest.fixture @@ -83,6 +86,60 @@ async def test_update_workflow_folder_uses_public_route(capture_request: MagicMo assert _route(capture_request).startswith(_workflow_http.PUBLIC_WORKFLOW_ROUTE + "/") +@pytest.mark.asyncio +async def test_retry_workflow_run_uses_public_route(capture_request: MagicMock) -> None: + await _workflow_http.retry_workflow_run_raw("wr_x") + assert _route(capture_request).startswith(_workflow_http.PUBLIC_WORKFLOW_ROUTE + "/") + + +@pytest.mark.asyncio +async def test_retry_workflow_run_bad_request_uses_detail(monkeypatch: pytest.MonkeyPatch) -> None: + async def raise_bad_request(workflow_run_id: str) -> dict[str, object]: + raise BadRequestError(body={"detail": f"Workflow run {workflow_run_id} is not terminal"}) + + monkeypatch.setattr(workflow_tools, "retry_workflow_run_raw", raise_bad_request) + + result = await workflow_tools.skyvern_workflow_retry("wr_x") + + assert result["ok"] is False + assert result["error"]["message"] == "Workflow run wr_x is not terminal" + + +@pytest.mark.asyncio +async def test_workflow_run_list_fetches_extra_row_for_has_more(monkeypatch: pytest.MonkeyPatch) -> None: + list_mock = AsyncMock( + return_value=[ + SimpleNamespace(workflow_run_id="wr_1", status="completed"), + SimpleNamespace(workflow_run_id="wr_2", status="completed"), + SimpleNamespace(workflow_run_id="wr_3", status="completed"), + ] + ) + monkeypatch.setattr(workflow_tools, "list_workflow_runs_raw", list_mock) + + result = await workflow_tools.skyvern_workflow_run_list("wpid_x", page=1, page_size=2) + + assert result["ok"] is True + list_mock.assert_awaited_once_with( + "wpid_x", + page=1, + page_size=3, + status=None, + search_key=None, + error_code=None, + ) + assert result["data"]["count"] == 2 + assert result["data"]["has_more"] is True + assert [run["run_id"] for run in result["data"]["runs"]] == ["wr_1", "wr_2"] + + +@pytest.mark.asyncio +async def test_retry_workflow_run_rejects_task_run_ids() -> None: + result = await workflow_tools.skyvern_workflow_retry("tsk_v2_x") + + assert result["ok"] is False + assert result["error"]["message"] == "Invalid workflow_run_id format: 'tsk_v2_x'" + + @pytest.mark.asyncio async def test_get_workflow_by_id_uses_internal_route(capture_request: MagicMock) -> None: await _workflow_http.get_workflow_by_id("wpid_x") diff --git a/tests/unit/test_runs_v2_route.py b/tests/unit/test_runs_v2_route.py index 80cae86ac..d0795b057 100644 --- a/tests/unit/test_runs_v2_route.py +++ b/tests/unit/test_runs_v2_route.py @@ -1,10 +1,16 @@ +from datetime import datetime, timezone from types import SimpleNamespace +from typing import Any from unittest.mock import AsyncMock import orjson import pytest +from fastapi import BackgroundTasks, HTTPException +from skyvern.exceptions import WorkflowNotFound +from skyvern.forge.sdk.db.enums import WorkflowRunTriggerType from skyvern.forge.sdk.routes import agent_protocol +from skyvern.forge.sdk.workflow.models.workflow import WorkflowRequestBody, WorkflowRunStatus @pytest.mark.asyncio @@ -61,3 +67,407 @@ async def test_get_runs_v2_serializes_mapping_rows_from_database(monkeypatch: py "script_run": False, } ] + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("handler", "expected_exclude_child_runs"), + [ + (agent_protocol.get_workflow_runs_by_id, True), + (agent_protocol.get_workflow_runs_by_id_legacy, False), + ], +) +async def test_get_workflow_runs_by_id_child_filter_depends_on_route( + monkeypatch: pytest.MonkeyPatch, + handler: Any, + expected_exclude_child_runs: bool, +) -> None: + mock_service = SimpleNamespace( + get_workflow_runs_for_workflow_permanent_id=AsyncMock(return_value=[]), + ) + monkeypatch.setattr(agent_protocol.app, "WORKFLOW_SERVICE", mock_service) + monkeypatch.setattr(agent_protocol.analytics, "capture", lambda *args, **kwargs: None) + + response = await handler( + workflow_id="wpid_123", + page=2, + page_size=5, + status=[WorkflowRunStatus.failed], + search_key="login", + error_code="LOGIN_FAILED", + current_org=SimpleNamespace(organization_id="org_123"), + ) + + assert response == [] + mock_service.get_workflow_runs_for_workflow_permanent_id.assert_awaited_once_with( + workflow_permanent_id="wpid_123", + organization_id="org_123", + page=2, + page_size=5, + status=[WorkflowRunStatus.failed], + search_key="login", + error_code="LOGIN_FAILED", + exclude_child_runs=expected_exclude_child_runs, + ) + + +@pytest.mark.asyncio +async def test_retry_workflow_run_replays_original_run_parameters(monkeypatch: pytest.MonkeyPatch) -> None: + created_at = datetime(2026, 4, 1, tzinfo=timezone.utc) + original_run = SimpleNamespace( + workflow_run_id="wr_original", + workflow_id="wf_original", + workflow_permanent_id="wpid_123", + status=WorkflowRunStatus.failed, + proxy_location=None, + webhook_callback_url="https://example.com/webhook", + totp_verification_url="https://example.com/totp", + totp_identifier="account@example.com", + browser_session_id="pbs_123", + browser_profile_id="bprof_123", + max_screenshot_scrolls=3, + extra_http_headers={"X-Test": "1"}, + cdp_connect_headers={"X-CDP-Auth": "secret"}, + browser_address="http://127.0.0.1:9222", + run_with="code", + ai_fallback=True, + debug_session_id=None, + code_gen=None, + ignore_inherited_workflow_system_prompt=True, + ) + retried_run = SimpleNamespace( + workflow_run_id="wr_retry", + workflow_id="wf_original", + status=WorkflowRunStatus.created, + failure_reason=None, + created_at=created_at, + modified_at=created_at, + browser_session_id="pbs_123", + browser_profile_id="bprof_123", + run_with="code", + ai_fallback=True, + ) + + mock_workflow_runs = SimpleNamespace( + get_workflow_run=AsyncMock(return_value=original_run), + get_workflow_run_parameters=AsyncMock( + return_value=[(SimpleNamespace(key="customer"), SimpleNamespace(value="acme"))] + ), + ) + mock_debug = SimpleNamespace(has_block_run_for_workflow_run=AsyncMock(return_value=False)) + mock_database = SimpleNamespace(workflow_runs=mock_workflow_runs, debug=mock_debug) + mock_workflow_service = SimpleNamespace( + get_workflow=AsyncMock( + return_value=SimpleNamespace(version=7, title="Original workflow title", organization_id="org_123") + ), + ) + mock_rate_limiter = SimpleNamespace(rate_limit_submit_run=AsyncMock()) + monkeypatch.setattr(agent_protocol.app, "DATABASE", mock_database) + monkeypatch.setattr(agent_protocol.app, "WORKFLOW_SERVICE", mock_workflow_service) + app_instance = object.__getattribute__(agent_protocol.app, "_inst") + monkeypatch.setattr(app_instance, "RATE_LIMITER", mock_rate_limiter, raising=False) + monkeypatch.setattr(agent_protocol.analytics, "capture", lambda *args, **kwargs: None) + monkeypatch.setattr(agent_protocol.skyvern_context, "ensure_context", lambda: SimpleNamespace(request_id="req_123")) + mock_agent_function = SimpleNamespace( + is_block_scoped_workflow_run=AsyncMock(return_value=False), + get_workflow_run_metadata=AsyncMock(return_value={"env": "prod"}), + ) + monkeypatch.setattr(app_instance, "AGENT_FUNCTION", mock_agent_function, raising=False) + + mock_permission_checker = SimpleNamespace(check=AsyncMock()) + monkeypatch.setattr( + agent_protocol.PermissionCheckerFactory, + "get_instance", + lambda: mock_permission_checker, + ) + + run_workflow_mock = AsyncMock(return_value=retried_run) + monkeypatch.setattr(agent_protocol.workflow_service, "run_workflow", run_workflow_mock) + + response = await agent_protocol.retry_workflow_run( + request=SimpleNamespace(), + background_tasks=BackgroundTasks(), + workflow_run_id="wr_original", + current_org=SimpleNamespace(organization_id="org_123"), + x_api_key="api-key", + x_max_steps_override=10, + x_user_agent="skyvern-ui", + ) + + mock_workflow_runs.get_workflow_run.assert_awaited_once_with( + workflow_run_id="wr_original", + organization_id="org_123", + ) + mock_debug.has_block_run_for_workflow_run.assert_awaited_once_with( + organization_id="org_123", + workflow_run_id="wr_original", + ) + mock_agent_function.is_block_scoped_workflow_run.assert_awaited_once_with(original_run) + mock_permission_checker.check.assert_awaited_once_with( + SimpleNamespace(organization_id="org_123"), browser_session_id="pbs_123" + ) + mock_rate_limiter.rate_limit_submit_run.assert_awaited_once_with("org_123") + mock_workflow_service.get_workflow.assert_awaited_once_with( + workflow_id="wf_original", + organization_id=None, + ) + mock_workflow_runs.get_workflow_run_parameters.assert_awaited_once_with( + workflow_run_id="wr_original", + ) + mock_agent_function.get_workflow_run_metadata.assert_awaited_once_with( + workflow_run_id="wr_original", + organization_id="org_123", + ) + + run_workflow_mock.assert_awaited_once() + call_kwargs = run_workflow_mock.call_args.kwargs + assert call_kwargs["workflow_id"] == "wpid_123" + assert call_kwargs["template"] is False + assert call_kwargs["version"] == 7 + assert call_kwargs["max_steps"] == 10 + assert call_kwargs["api_key"] == "api-key" + assert call_kwargs["request_id"] == "req_123" + assert call_kwargs["trigger_type"] == WorkflowRunTriggerType.manual + assert call_kwargs["ignore_inherited_workflow_system_prompt"] is True + assert isinstance(call_kwargs["workflow_request"], WorkflowRequestBody) + assert call_kwargs["workflow_request"].data == {"customer": "acme"} + assert call_kwargs["workflow_request"].webhook_callback_url == "https://example.com/webhook" + assert call_kwargs["workflow_request"].totp_verification_url == "https://example.com/totp" + assert call_kwargs["workflow_request"].totp_identifier == "account@example.com" + assert call_kwargs["workflow_request"].browser_session_id == "pbs_123" + assert call_kwargs["workflow_request"].browser_profile_id == "bprof_123" + assert call_kwargs["workflow_request"].max_screenshot_scrolls == 3 + assert call_kwargs["workflow_request"].extra_http_headers == {"X-Test": "1"} + assert call_kwargs["workflow_request"].cdp_connect_headers == {"X-CDP-Auth": "secret"} + assert call_kwargs["workflow_request"].browser_address == "http://127.0.0.1:9222" + assert call_kwargs["workflow_request"].run_with == "code" + assert call_kwargs["workflow_request"].ai_fallback is True + assert call_kwargs["workflow_request"].run_metadata == {"env": "prod"} + + assert response.run_id == "wr_retry" + assert response.run_request is not None + assert response.run_request.workflow_id == "wpid_123" + assert response.run_request.title == "Original workflow title" + assert response.run_request.parameters == {"customer": "acme"} + + +@pytest.mark.asyncio +async def test_retry_workflow_run_rejects_block_scoped_run(monkeypatch: pytest.MonkeyPatch) -> None: + mock_debug = SimpleNamespace(has_block_run_for_workflow_run=AsyncMock(return_value=True)) + mock_workflow_runs = SimpleNamespace( + get_workflow_run=AsyncMock( + return_value=SimpleNamespace( + workflow_run_id="wr_block", + status=WorkflowRunStatus.failed, + debug_session_id=None, + code_gen=None, + ) + ) + ) + monkeypatch.setattr( + agent_protocol.app, + "DATABASE", + SimpleNamespace(workflow_runs=mock_workflow_runs, debug=mock_debug), + ) + app_instance = object.__getattribute__(agent_protocol.app, "_inst") + monkeypatch.setattr( + app_instance, + "AGENT_FUNCTION", + SimpleNamespace(is_block_scoped_workflow_run=AsyncMock(return_value=False)), + raising=False, + ) + monkeypatch.setattr(agent_protocol.analytics, "capture", lambda *args, **kwargs: None) + + with pytest.raises(HTTPException) as exc_info: + await agent_protocol.retry_workflow_run( + request=SimpleNamespace(), + background_tasks=BackgroundTasks(), + workflow_run_id="wr_block", + current_org=SimpleNamespace(organization_id="org_123"), + x_api_key=None, + x_max_steps_override=None, + x_user_agent=None, + ) + + assert exc_info.value.status_code == 400 + assert exc_info.value.detail == "Block-scoped workflow runs cannot be retried with this endpoint" + mock_debug.has_block_run_for_workflow_run.assert_awaited_once_with( + organization_id="org_123", + workflow_run_id="wr_block", + ) + + +@pytest.mark.asyncio +async def test_retry_workflow_run_replays_template_runs_as_templates(monkeypatch: pytest.MonkeyPatch) -> None: + now = datetime(2026, 4, 1, tzinfo=timezone.utc) + original_run = SimpleNamespace( + workflow_run_id="wr_template", + workflow_id="wf_template", + workflow_permanent_id="wpid_template", + status=WorkflowRunStatus.completed, + proxy_location=None, + webhook_callback_url=None, + totp_verification_url=None, + totp_identifier=None, + browser_session_id=None, + browser_profile_id=None, + max_screenshot_scrolls=None, + extra_http_headers=None, + cdp_connect_headers=None, + browser_address=None, + run_with=None, + ai_fallback=None, + debug_session_id=None, + code_gen=None, + ignore_inherited_workflow_system_prompt=False, + ) + retried_run = SimpleNamespace( + workflow_run_id="wr_template_retry", + workflow_id="wf_template", + status=WorkflowRunStatus.created, + failure_reason=None, + created_at=now, + modified_at=now, + browser_session_id=None, + browser_profile_id=None, + run_with=None, + ai_fallback=None, + ) + mock_workflow_runs = SimpleNamespace( + get_workflow_run=AsyncMock(return_value=original_run), + get_workflow_run_parameters=AsyncMock(return_value=[]), + ) + mock_debug = SimpleNamespace(has_block_run_for_workflow_run=AsyncMock(return_value=False)) + mock_database = SimpleNamespace(workflow_runs=mock_workflow_runs, debug=mock_debug) + mock_workflow_service = SimpleNamespace( + get_workflow=AsyncMock( + return_value=SimpleNamespace(version=3, title="Template title", organization_id="template_org") + ), + ) + mock_rate_limiter = SimpleNamespace(rate_limit_submit_run=AsyncMock()) + monkeypatch.setattr(agent_protocol.app, "DATABASE", mock_database) + monkeypatch.setattr(agent_protocol.app, "WORKFLOW_SERVICE", mock_workflow_service) + app_instance = object.__getattribute__(agent_protocol.app, "_inst") + monkeypatch.setattr(app_instance, "RATE_LIMITER", mock_rate_limiter, raising=False) + monkeypatch.setattr( + app_instance, + "AGENT_FUNCTION", + SimpleNamespace( + is_block_scoped_workflow_run=AsyncMock(return_value=False), + get_workflow_run_metadata=AsyncMock(side_effect=RuntimeError("metadata unavailable")), + ), + raising=False, + ) + monkeypatch.setattr(agent_protocol.analytics, "capture", lambda *args, **kwargs: None) + monkeypatch.setattr(agent_protocol.skyvern_context, "ensure_context", lambda: SimpleNamespace(request_id="req_123")) + monkeypatch.setattr( + agent_protocol.PermissionCheckerFactory, + "get_instance", + lambda: SimpleNamespace(check=AsyncMock()), + ) + + run_workflow_mock = AsyncMock(return_value=retried_run) + monkeypatch.setattr(agent_protocol.workflow_service, "run_workflow", run_workflow_mock) + + await agent_protocol.retry_workflow_run( + request=SimpleNamespace(), + background_tasks=BackgroundTasks(), + workflow_run_id="wr_template", + current_org=SimpleNamespace(organization_id="org_123"), + x_api_key=None, + x_max_steps_override=None, + x_user_agent=None, + ) + + run_workflow_mock.assert_awaited_once() + assert run_workflow_mock.call_args.kwargs["template"] is True + assert run_workflow_mock.call_args.kwargs["version"] == 3 + assert run_workflow_mock.call_args.kwargs["ignore_inherited_workflow_system_prompt"] is False + assert run_workflow_mock.call_args.kwargs["workflow_request"].run_metadata is None + mock_workflow_service.get_workflow.assert_awaited_once_with( + workflow_id="wf_template", + organization_id=None, + ) + mock_workflow_runs.get_workflow_run_parameters.assert_awaited_once_with( + workflow_run_id="wr_template", + ) + app_instance.AGENT_FUNCTION.is_block_scoped_workflow_run.assert_awaited_once_with(original_run) + + +@pytest.mark.asyncio +async def test_retry_workflow_run_rejects_missing_workflow(monkeypatch: pytest.MonkeyPatch) -> None: + original_run = SimpleNamespace( + workflow_run_id="wr_missing_workflow", + workflow_id="wf_missing", + workflow_permanent_id="wpid_missing", + status=WorkflowRunStatus.failed, + browser_session_id=None, + debug_session_id=None, + code_gen=None, + ) + mock_database = SimpleNamespace( + workflow_runs=SimpleNamespace( + get_workflow_run=AsyncMock(return_value=original_run), + get_workflow_run_parameters=AsyncMock(), + ), + debug=SimpleNamespace(has_block_run_for_workflow_run=AsyncMock(return_value=False)), + ) + mock_workflow_service = SimpleNamespace( + get_workflow=AsyncMock(side_effect=WorkflowNotFound(workflow_id="wf_missing")), + ) + monkeypatch.setattr(agent_protocol.app, "DATABASE", mock_database) + monkeypatch.setattr(agent_protocol.app, "WORKFLOW_SERVICE", mock_workflow_service) + app_instance = object.__getattribute__(agent_protocol.app, "_inst") + monkeypatch.setattr( + app_instance, + "AGENT_FUNCTION", + SimpleNamespace(is_block_scoped_workflow_run=AsyncMock(return_value=False)), + raising=False, + ) + monkeypatch.setattr( + agent_protocol.PermissionCheckerFactory, + "get_instance", + lambda: SimpleNamespace(check=AsyncMock()), + ) + monkeypatch.setattr(app_instance, "RATE_LIMITER", SimpleNamespace(rate_limit_submit_run=AsyncMock()), raising=False) + monkeypatch.setattr(agent_protocol.analytics, "capture", lambda *args, **kwargs: None) + + with pytest.raises(HTTPException) as exc_info: + await agent_protocol.retry_workflow_run( + request=SimpleNamespace(), + background_tasks=BackgroundTasks(), + workflow_run_id="wr_missing_workflow", + current_org=SimpleNamespace(organization_id="org_123"), + x_api_key=None, + x_max_steps_override=None, + x_user_agent=None, + ) + + assert exc_info.value.status_code == 404 + assert exc_info.value.detail == "Workflow not found for run wr_missing_workflow" + mock_database.workflow_runs.get_workflow_run_parameters.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_retry_workflow_run_rejects_active_run(monkeypatch: pytest.MonkeyPatch) -> None: + mock_workflow_runs = SimpleNamespace( + get_workflow_run=AsyncMock( + return_value=SimpleNamespace(workflow_run_id="wr_running", status=WorkflowRunStatus.running) + ) + ) + monkeypatch.setattr(agent_protocol.app, "DATABASE", SimpleNamespace(workflow_runs=mock_workflow_runs)) + monkeypatch.setattr(agent_protocol.analytics, "capture", lambda *args, **kwargs: None) + + with pytest.raises(HTTPException) as exc_info: + await agent_protocol.retry_workflow_run( + request=SimpleNamespace(), + background_tasks=BackgroundTasks(), + workflow_run_id="wr_running", + current_org=SimpleNamespace(organization_id="org_123"), + x_api_key=None, + x_max_steps_override=None, + x_user_agent=None, + ) + + assert exc_info.value.status_code == 400 + assert exc_info.value.detail == "Only terminal workflow runs can be retried" diff --git a/tests/unit/test_sdk_run_metadata.py b/tests/unit/test_sdk_run_metadata.py index 83b77a7ab..ee4ee94a0 100644 --- a/tests/unit/test_sdk_run_metadata.py +++ b/tests/unit/test_sdk_run_metadata.py @@ -2,8 +2,9 @@ import json from datetime import UTC, datetime import httpx +import pytest -from skyvern.client import Skyvern +from skyvern.client import AsyncSkyvern, Skyvern from skyvern.client.types.workflow_run_request import WorkflowRunRequest @@ -46,3 +47,144 @@ def test_run_workflow_sends_run_metadata() -> None: "run_metadata": {"customer": "acme", "tier": "enterprise"}, } ] + + +def _workflow_run_payload() -> dict[str, object]: + now = datetime.now(UTC).isoformat() + return { + "workflow_run_id": "wr_123", + "workflow_id": "wf_123", + "workflow_permanent_id": "wpid_123", + "organization_id": "org_123", + "status": "completed", + "created_at": now, + "modified_at": now, + } + + +def _workflow_run_response_payload() -> dict[str, object]: + now = datetime.now(UTC).isoformat() + return { + "run_id": "wr_retry", + "status": "queued", + "created_at": now, + "modified_at": now, + } + + +def test_retry_workflow_run_uses_retry_route() -> None: + captured_requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_requests.append(request) + return httpx.Response(status_code=200, json=_workflow_run_response_payload()) + + client = Skyvern( + base_url="https://api.example.test", + api_key="test-key", + httpx_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + response = client.retry_workflow_run( + "wr_original", + max_steps_override=12, + user_agent="skyvern-ui", + ) + + assert response.run_id == "wr_retry" + assert captured_requests[0].method == "POST" + assert captured_requests[0].url.path == "/v1/workflows/runs/wr_original/retry" + assert captured_requests[0].headers["x-max-steps-override"] == "12" + assert captured_requests[0].headers["x-user-agent"] == "skyvern-ui" + + +@pytest.mark.asyncio +async def test_async_retry_workflow_run_uses_retry_route() -> None: + captured_requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + captured_requests.append(request) + return httpx.Response(status_code=200, json=_workflow_run_response_payload()) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as httpx_client: + client = AsyncSkyvern( + base_url="https://api.example.test", + api_key="test-key", + httpx_client=httpx_client, + ) + response = await client.retry_workflow_run( + "wr_original", + max_steps_override=8, + user_agent="skyvern-ui", + ) + + assert response.run_id == "wr_retry" + assert captured_requests[0].method == "POST" + assert captured_requests[0].url.path == "/v1/workflows/runs/wr_original/retry" + assert captured_requests[0].headers["x-max-steps-override"] == "8" + assert captured_requests[0].headers["x-user-agent"] == "skyvern-ui" + + +def test_get_workflow_runs_by_id_uses_workflow_scoped_route() -> None: + captured_requests: list[httpx.Request] = [] + + def handler(request: httpx.Request) -> httpx.Response: + captured_requests.append(request) + return httpx.Response(status_code=200, json=[_workflow_run_payload()]) + + client = Skyvern( + base_url="https://api.example.test", + api_key="test-key", + httpx_client=httpx.Client(transport=httpx.MockTransport(handler)), + ) + + runs = client.get_workflow_runs_by_id( + "wpid_123", + page=1, + page_size=10, + status="completed", + search_key="acme", + error_code="LOGIN_FAILED", + ) + + assert runs[0].workflow_run_id == "wr_123" + assert captured_requests[0].method == "GET" + assert captured_requests[0].url.path == "/v1/workflows/wpid_123/runs" + assert captured_requests[0].url.params["page"] == "1" + assert captured_requests[0].url.params["page_size"] == "10" + assert captured_requests[0].url.params["status"] == "completed" + assert captured_requests[0].url.params["search_key"] == "acme" + assert captured_requests[0].url.params["error_code"] == "LOGIN_FAILED" + + +@pytest.mark.asyncio +async def test_async_get_workflow_runs_by_id_uses_workflow_scoped_route() -> None: + captured_requests: list[httpx.Request] = [] + + async def handler(request: httpx.Request) -> httpx.Response: + captured_requests.append(request) + return httpx.Response(status_code=200, json=[_workflow_run_payload()]) + + async with httpx.AsyncClient(transport=httpx.MockTransport(handler)) as httpx_client: + client = AsyncSkyvern( + base_url="https://api.example.test", + api_key="test-key", + httpx_client=httpx_client, + ) + runs = await client.get_workflow_runs_by_id( + "wpid_123", + page=2, + page_size=5, + status="failed", + search_key="prod", + error_code="TIMEOUT", + ) + + assert runs[0].workflow_run_id == "wr_123" + assert captured_requests[0].method == "GET" + assert captured_requests[0].url.path == "/v1/workflows/wpid_123/runs" + assert captured_requests[0].url.params["page"] == "2" + assert captured_requests[0].url.params["page_size"] == "5" + assert captured_requests[0].url.params["status"] == "failed" + assert captured_requests[0].url.params["search_key"] == "prod" + assert captured_requests[0].url.params["error_code"] == "TIMEOUT"