Skip to content
Permalink

Comparing changes

This is a direct comparison between two commits made in this repository or its related repositories. View the default comparison for this range or learn more about diff comparisons.

Open a pull request

Create a new pull request by comparing changes across two branches. If you need to, you can also . Learn more about diff comparisons here.
base repository: Skyvern-AI/skyvern
Failed to load repositories. Confirm that selected base ref is valid, then try again.
Loading
base: d7b8117226e638555f43550a0edf9bb6fe8d774f
Choose a base ref
..
head repository: Skyvern-AI/skyvern
Failed to load repositories. Confirm that selected head ref is valid, then try again.
Loading
compare: 9d93949ee8d3f29f300d55270f9317903bc4769c
Choose a head ref
Showing with 33 additions and 12 deletions.
  1. +2 −2 skyvern/forge/agent.py
  2. +14 −1 skyvern/forge/sdk/routes/agent_protocol.py
  3. +17 −9 skyvern/forge/sdk/workflow/service.py
4 changes: 2 additions & 2 deletions skyvern/forge/agent.py
Original file line number Diff line number Diff line change
@@ -285,7 +285,7 @@ async def execute_step(
task=task,
last_step=step,
api_key=api_key,
need_call_webhook=False,
need_call_webhook=True,
)
return step, None, None

@@ -1543,7 +1543,7 @@ async def clean_up_task(
async def execute_task_webhook(
self,
task: Task,
last_step: Step,
last_step: Step | None,
api_key: str | None,
) -> None:
if not api_key:
15 changes: 14 additions & 1 deletion skyvern/forge/sdk/routes/agent_protocol.py
Original file line number Diff line number Diff line change
@@ -284,6 +284,7 @@ async def get_task(
async def cancel_task(
task_id: str,
current_org: Organization = Depends(org_auth_service.get_current_org),
x_api_key: Annotated[str | None, Header()] = None,
) -> None:
analytics.capture("skyvern-oss-agent-task-get")
task_obj = await app.DATABASE.get_task(task_id, organization_id=current_org.organization_id)
@@ -292,16 +293,28 @@ async def cancel_task(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Task not found {task_id}",
)
await app.agent.update_task(task_obj, status=TaskStatus.canceled)
task = await app.agent.update_task(task_obj, status=TaskStatus.canceled)
# get latest step
latest_step = await app.DATABASE.get_latest_step(task_id, organization_id=current_org.organization_id)
# retry the webhook
await app.agent.execute_task_webhook(task=task, last_step=latest_step, api_key=x_api_key)


@base_router.post("/workflows/runs/{workflow_run_id}/cancel")
@base_router.post("/workflows/runs/{workflow_run_id}/cancel/", include_in_schema=False)
async def cancel_workflow_run(
workflow_run_id: str,
current_org: Organization = Depends(org_auth_service.get_current_org),
x_api_key: Annotated[str | None, Header()] = None,
) -> None:
workflow_run = await app.DATABASE.get_workflow_run(workflow_run_id=workflow_run_id)
if not workflow_run:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail=f"Workflow run not found {workflow_run_id}",
)
await app.WORKFLOW_SERVICE.mark_workflow_run_as_canceled(workflow_run_id)
await app.WORKFLOW_SERVICE.execute_workflow_webhook(workflow_run, api_key=x_api_key)


@base_router.post(
26 changes: 17 additions & 9 deletions skyvern/forge/sdk/workflow/service.py
Original file line number Diff line number Diff line change
@@ -233,7 +233,7 @@ async def execute_workflow(
workflow=workflow,
workflow_run=workflow_run,
api_key=api_key,
need_call_webhook=False,
need_call_webhook=True,
)
return workflow_run
parameters = block.get_all_parameters(workflow_run_id)
@@ -881,10 +881,18 @@ async def clean_up_workflow(
if not need_call_webhook:
return

await self.execute_workflow_webhook(workflow_run, api_key)

async def execute_workflow_webhook(
self,
workflow_run: WorkflowRun,
api_key: str | None = None,
) -> None:
workflow_id = workflow_run.workflow_id
workflow_run_status_response = await self.build_workflow_run_status_response(
workflow_permanent_id=workflow.workflow_permanent_id,
workflow_permanent_id=workflow_run.workflow_permanent_id,
workflow_run_id=workflow_run.workflow_run_id,
organization_id=workflow.organization_id,
organization_id=workflow_run.organization_id,
)
LOG.info(
"Built workflow run status response",
@@ -894,15 +902,15 @@ async def clean_up_workflow(
if not workflow_run.webhook_callback_url:
LOG.warning(
"Workflow has no webhook callback url. Not sending workflow response",
workflow_id=workflow.workflow_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run.workflow_run_id,
)
return

if not api_key:
LOG.warning(
"Request has no api key. Not sending workflow response",
workflow_id=workflow.workflow_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run.workflow_run_id,
)
return
@@ -921,7 +929,7 @@ async def clean_up_workflow(
}
LOG.info(
"Sending webhook run status to webhook callback url",
workflow_id=workflow.workflow_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run.workflow_run_id,
webhook_callback_url=workflow_run.webhook_callback_url,
payload=payload,
@@ -934,15 +942,15 @@ async def clean_up_workflow(
if resp.status_code == 200:
LOG.info(
"Webhook sent successfully",
workflow_id=workflow.workflow_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run.workflow_run_id,
resp_code=resp.status_code,
resp_text=resp.text,
)
else:
LOG.info(
"Webhook failed",
workflow_id=workflow.workflow_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run.workflow_run_id,
webhook_data=payload,
resp=resp,
@@ -951,7 +959,7 @@ async def clean_up_workflow(
)
except Exception as e:
raise FailedToSendWebhook(
workflow_id=workflow.workflow_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run.workflow_run_id,
) from e