WYV-1186 Create API to replay webhook callbacks (#4)

skyvern/forge/agent.py

@@ -692,16 +692,19 @@ class ForgeAgent(Agent):
         # Wait for all tasks to complete before generating the links for the artifacts
         await app.ARTIFACT_MANAGER.wait_for_upload_aiotasks_for_task(task.task_id)
 
-        if not task.webhook_callback_url:
+        await self.execute_task_webhook(task=task, last_step=last_step, api_key=api_key)
+
+    async def execute_task_webhook(self, task: Task, last_step: Step, api_key: str | None) -> None:
+        if not api_key:
             LOG.warning(
-                "Task has no webhook callback url. Not sending task response",
+                "Request has no api key. Not sending task response",
                 task_id=task.task_id,
             )
             return
 
-        if not api_key:
+        if not task.webhook_callback_url:
             LOG.warning(
-                "Request has no api key. Not sending task response",
+                "Task has no webhook callback url. Not sending task response",
                 task_id=task.task_id,
             )
             return
@@ -732,6 +735,7 @@ class ForgeAgent(Agent):
         if not task_from_db:
             LOG.error("Failed to get task from db when sending task response")
             raise TaskNotFound(task_id=task.task_id)
+
         task = task_from_db
         if not task.webhook_callback_url:
             LOG.info("Task has no webhook callback url. Not sending task response")

skyvern/forge/sdk/routes/agent_protocol.py

@@ -218,6 +218,36 @@ async def get_task(
     )
 
 
+@base_router.post(
+    "/tasks/{task_id}/retry_webhook",
+    tags=["agent"],
+    response_model=TaskResponse,
+)
+async def retry_webhook(
+    request: Request,
+    task_id: str,
+    current_org: Organization = Depends(org_auth_service.get_current_org),
+    x_api_key: Annotated[str | None, Header()] = None,
+) -> TaskResponse:
+    agent = request["agent"]
+    task_obj = await agent.db.get_task(task_id, organization_id=current_org.organization_id)
+    if not task_obj:
+        raise HTTPException(
+            status_code=status.HTTP_404_NOT_FOUND,
+            detail=f"Task not found {task_id}",
+        )
+
+    # get latest step
+    latest_step = await agent.db.get_latest_step(task_id, organization_id=current_org.organization_id)
+    if not latest_step:
+        return task_obj.to_task_response()
+
+    # retry the webhook
+    await agent.execute_task_webhook(task=task_obj, last_step=latest_step, api_key=x_api_key)
+
+    return task_obj.to_task_response()
+
+
 @base_router.get("/internal/tasks/{task_id}", response_model=list[Task])
 async def get_task_internal(
     request: Request,