Overview
Once you’ve created a workflow, you can run it multiple times with different parameter values. Workflows execute asynchronously in the background, and you can retrieve results via polling or webhooks.
Running a Workflow
Using Python SDK
import os
import asyncio
from skyvern import Skyvern
async def main ():
client = Skyvern( api_key = os.getenv( "SKYVERN_API_KEY" ))
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = {
"company_name" : "Acme Corp" ,
"start_date" : "2026-01-01"
}
)
print ( f "Run ID: { run.run_id } " )
print ( f "Status: { run.status } " )
asyncio.run(main())
Using TypeScript SDK
import { SkyvernClient } from "@skyvern/client" ;
async function main () {
const client = new SkyvernClient ({
apiKey: process . env . SKYVERN_API_KEY ,
});
const run = await client . runWorkflow ({
body: {
workflow_id: "wpid_123456789" ,
parameters: {
company_name: "Acme Corp" ,
start_date: "2026-01-01" ,
},
},
});
console . log ( `Run ID: ${ run . run_id } ` );
console . log ( `Status: ${ run . status } ` );
}
main ();
Using REST API
curl -X POST "https://api.skyvern.com/v1/run/workflows" \
-H "x-api-key: $SKYVERN_API_KEY " \
-H "Content-Type: application/json" \
-d '{
"workflow_id": "wpid_123456789",
"parameters": {
"company_name": "Acme Corp",
"start_date": "2026-01-01"
}
}'
Run Parameters
Required Parameters
The workflow_permanent_id returned when creating the workflow.
Values for the workflow’s input parameters. Keys must match the key field in the workflow’s parameter definitions.
Optional Parameters
Display name for this specific workflow run.
Override the workflow’s default proxy location for this run.
URL to receive a POST request when the workflow completes.
ID of an existing browser session to reuse.
ID of a browser profile to reuse.
Example: Running with All Parameters
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
title = "January 2026 Invoice Batch" ,
parameters = {
"vendor_portal_url" : "https://vendor.example.com" ,
"start_date" : "2026-01-01" ,
"end_date" : "2026-01-31" ,
"notification_email" : "[email protected] "
},
proxy_location = "RESIDENTIAL_GB" ,
webhook_url = "https://your-server.com/webhook/invoice-complete"
)
Passing Parameters
Simple Parameters
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = {
"first_name" : "John" ,
"last_name" : "Smith" ,
"age" : 35 ,
"is_premium" : True
}
)
JSON Parameters
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = {
"company_info" : {
"name" : "Acme Corp" ,
"tax_id" : "12-3456789" ,
"address" : {
"street" : "123 Main St" ,
"city" : "San Francisco" ,
"state" : "CA" ,
"zip" : "94102"
}
}
}
)
Array Parameters
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = {
"job_urls" : [
"https://jobs.lever.co/company/position-1" ,
"https://jobs.lever.co/company/position-2" ,
"https://jobs.lever.co/company/position-3"
],
"recipient_emails" : [
"[email protected] " ,
"[email protected] "
]
}
)
File URL Parameters
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = {
"resume" : "https://example.com/resumes/john-smith.pdf" ,
"transcript" : "https://example.com/transcripts/john-smith.pdf"
}
)
Run Response
The initial response contains basic information about the run:
{
"run_id" : "wr_486305187432193510" ,
"status" : "created"
}
Unique identifier for this run (format: wr_*).
Current status: created, queued, running, completed, failed, terminated, timed_out, canceled.
Retrieving Results
Workflows run asynchronously. Use polling or webhooks to get results.
Option 1: Polling
Poll the get_run endpoint until the status reaches a terminal state.
Python:
import asyncio
run_id = run.run_id
while True :
result = await client.get_run(run_id)
if result.status in [ "completed" , "failed" , "terminated" , "timed_out" , "canceled" ]:
break
print ( f "Status: { result.status } " )
await asyncio.sleep( 5 )
print ( f "Final status: { result.status } " )
print ( f "Output: { result.output } " )
if result.status == "completed" :
print ( f "Recording: { result.recording_url } " )
else :
print ( f "Failure reason: { result.failure_reason } " )
TypeScript:
const runId = run . run_id ;
while ( true ) {
const result = await client . getRun ( runId );
if ([ "completed" , "failed" , "terminated" , "timed_out" , "canceled" ]. includes ( result . status )) {
console . log ( `Final status: ${ result . status } ` );
console . log ( `Output: ${ JSON . stringify ( result . output , null , 2 ) } ` );
if ( result . status === "completed" ) {
console . log ( `Recording: ${ result . recording_url } ` );
} else {
console . log ( `Failure reason: ${ result . failure_reason } ` );
}
break ;
}
console . log ( `Status: ${ result . status } ` );
await new Promise (( resolve ) => setTimeout ( resolve , 5000 ));
}
cURL:
#!/bin/bash
RUN_ID = "wr_486305187432193510"
while true ; do
RESPONSE = $( curl -s -X GET "https://api.skyvern.com/v1/runs/ $RUN_ID " \
-H "x-api-key: $SKYVERN_API_KEY " )
STATUS = $( echo " $RESPONSE " | jq -r '.status' )
echo "Status: $STATUS "
if [[ " $STATUS " == "completed" || " $STATUS " == "failed" || " $STATUS " == "terminated" || " $STATUS " == "timed_out" || " $STATUS " == "canceled" ]]; then
echo " $RESPONSE " | jq '.output'
break
fi
sleep 5
done
Option 2: Webhooks
Pass a webhook_url when running the workflow. Skyvern sends a POST request when the workflow reaches a terminal state.
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = { "company_name" : "Acme Corp" },
webhook_url = "https://your-server.com/webhook"
)
The webhook payload contains the same data as the polling response:
{
"run_id" : "wr_486305187432193510" ,
"status" : "completed" ,
"output" : {
"parse_resume_output" : { ... },
"apply_to_job_output" : { ... }
},
"recording_url" : "https://..." ,
"failure_reason" : null
}
See Webhooks for authentication and retry options.
Understanding the Response
The complete run response contains:
{
"run_id" : "wr_486305187432193510" ,
"run_type" : "workflow_run" ,
"status" : "completed" ,
"output" : {
"parse_resume_output" : {
"name" : "John Smith" ,
"email" : "[email protected] " ,
"work_experience" : [ ... ]
},
"apply_to_job_output" : {
"task_id" : "tsk_123456" ,
"status" : "completed"
}
},
"screenshot_urls" : [
"https://skyvern-artifacts.s3.amazonaws.com/.../screenshot_final.png"
],
"recording_url" : "https://skyvern-artifacts.s3.amazonaws.com/.../recording.webm" ,
"failure_reason" : null ,
"errors" : null ,
"downloaded_files" : [
{
"file_name" : "ein_confirmation.pdf" ,
"file_url" : "https://..."
}
],
"created_at" : "2026-03-02T12:00:00.000000" ,
"started_at" : "2026-03-02T12:00:05.000000" ,
"finished_at" : "2026-03-02T12:05:00.000000" ,
"app_url" : "https://app.skyvern.com/workflows/wpid_123456789/runs/wr_486305187432193510"
}
Response Fields
Unique identifier for this run.
Always workflow_run for workflow runs.
Current status of the run.
Output from each block, keyed by {label}_output.
Final screenshots from the last blocks.
Video recording of the browser session.
Error description if the run failed.
List of errors encountered during the run.
Files downloaded during the run.
When the run was created.
Link to view this run in the Skyvern UI.
Status Values
Status Description createdRun has been created but not yet queued queuedRun is waiting for execution runningRun is currently executing completedRun completed successfully failedRun failed due to an error terminatedRun was terminated by a TERMINATE condition timed_outRun exceeded the timeout limit canceledRun was manually canceled
Block Outputs
Each block’s output appears in the output object with the key {label}_output.
Navigation/Action/Login blocks:
{
"apply_to_job_output" : {
"task_id" : "tsk_123456" ,
"status" : "completed" ,
"extracted_information" : null ,
"failure_reason" : null
}
}
Extraction blocks:
{
"extract_credentials_output" : {
"asha_account_number" : "12345678" ,
"certification_status" : "active" ,
"valid_through" : "2027-12-31"
}
}
For Loop blocks:
{
"download_invoices_output" : [
{ "task_id" : "tsk_111" , "status" : "completed" },
{ "task_id" : "tsk_222" , "status" : "completed" },
{ "task_id" : "tsk_333" , "status" : "completed" }
]
}
Code blocks:
{
"calculate_totals_output" : {
"subtotal" : 99.99 ,
"tax" : 8.00 ,
"total" : 107.99
}
}
Proxy Configuration
Override the workflow’s default proxy location for a specific run:
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = { ... },
proxy_location = "RESIDENTIAL_GB" # Use UK residential proxy
)
Available proxy locations:
Value Location RESIDENTIALUnited States (default) RESIDENTIAL_ISPUnited States (static ISP IPs) RESIDENTIAL_ARArgentina RESIDENTIAL_AUAustralia RESIDENTIAL_BRBrazil RESIDENTIAL_CACanada RESIDENTIAL_DEGermany RESIDENTIAL_ESSpain RESIDENTIAL_FRFrance RESIDENTIAL_GBUnited Kingdom RESIDENTIAL_ITItaly RESIDENTIAL_JPJapan RESIDENTIAL_MXMexico RESIDENTIAL_NLNetherlands RESIDENTIAL_INIndia RESIDENTIAL_KRSouth Korea NONENo proxy
City-level targeting:
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = { ... },
proxy_location = {
"country" : "US" ,
"subdivision" : "CA" ,
"city" : "San Francisco"
}
)
See Proxy & Geo Targeting for more details.
Browser Sessions
Reusing Browser Sessions
Continue from an existing browser state:
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
browser_session_id = "pbs_abc123" ,
parameters = { ... }
)
The workflow starts from the current page in that session.
Reusing Browser Profiles
Persist cookies and storage across runs:
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
browser_profile_id = "bp_xyz789" ,
parameters = { ... }
)
This is useful for workflows that need to stay logged in.
Error Handling
Check the status and failure_reason fields:
result = await client.get_run(run_id)
if result.status == "completed" :
print ( f "Success! Output: { result.output } " )
elif result.status == "failed" :
print ( f "Failed: { result.failure_reason } " )
if result.errors:
for error in result.errors:
print ( f "- { error } " )
elif result.status == "terminated" :
print ( f "Terminated: { result.failure_reason } " )
Common Failure Reasons
Navigation timeout : Block exceeded max_steps_per_run
Element not found : Required element not present on page
Authentication failed : Invalid credentials
Download failed : File download did not complete
Validation failed : Validation criterion not met
Network error : Connection or proxy issues
Monitoring & Debugging
View in Skyvern UI
Use the app_url from the response to view the run in the Skyvern dashboard:
result = await client.get_run(run_id)
print ( f "View run: { result.app_url } " )
Recording & Screenshots
Every run includes:
Recording URL : Video of the entire browser session
Screenshot URLs : Final screenshots from each block
result = await client.get_run(run_id)
if result.status == "completed" :
print ( f "Recording: { result.recording_url } " )
for i, screenshot_url in enumerate (result.screenshot_urls):
print ( f "Screenshot { i + 1 } : { screenshot_url } " )
Downloaded Files
Access files downloaded during the workflow:
result = await client.get_run(run_id)
if result.downloaded_files:
for file in result.downloaded_files:
print ( f "Downloaded: { file [ 'file_name' ] } " )
print ( f "URL: { file [ 'file_url' ] } " )
Complete Example
Here’s a complete script that creates, runs, and monitors a workflow:
import os
import asyncio
from skyvern import Skyvern
async def main ():
client = Skyvern( api_key = os.getenv( "SKYVERN_API_KEY" ))
# Create workflow
workflow = await client.create_workflow(
json_definition = {
"title" : "Credential Verification" ,
"workflow_definition" : {
"parameters" : [
{ "key" : "first_name" , "parameter_type" : "workflow" , "workflow_parameter_type" : "string" },
{ "key" : "last_name" , "parameter_type" : "workflow" , "workflow_parameter_type" : "string" },
{ "key" : "state" , "parameter_type" : "workflow" , "workflow_parameter_type" : "string" }
],
"blocks" : [
{
"label" : "search_registry" ,
"block_type" : "navigation" ,
"url" : "https://asha.org/verify" ,
"navigation_goal" : "Search for {{ first_name }} {{ last_name }} in {{ state }} . COMPLETE when results shown." ,
"parameter_keys" : [ "first_name" , "last_name" , "state" ]
},
{
"label" : "extract_credentials" ,
"block_type" : "extraction" ,
"data_extraction_goal" : "Extract ASHA account number and certification status." ,
"data_schema" : {
"type" : "object" ,
"properties" : {
"asha_account_number" : { "type" : "string" },
"certification_status" : { "type" : "string" }
}
}
}
]
}
}
)
print ( f "Created workflow: { workflow.workflow_permanent_id } " )
# Run workflow
run = await client.run_workflow(
workflow_id = workflow.workflow_permanent_id,
parameters = {
"first_name" : "John" ,
"last_name" : "Smith" ,
"state" : "California"
}
)
print ( f "Started run: { run.run_id } " )
# Poll for results
while True :
result = await client.get_run(run.run_id)
if result.status in [ "completed" , "failed" , "terminated" , "timed_out" , "canceled" ]:
break
print ( f "Status: { result.status } " )
await asyncio.sleep( 5 )
# Display results
print ( f " \n Final status: { result.status } " )
if result.status == "completed" :
print ( f " \n Extracted credentials:" )
print ( f " Account: { result.output[ 'extract_credentials_output' ][ 'asha_account_number' ] } " )
print ( f " Status: { result.output[ 'extract_credentials_output' ][ 'certification_status' ] } " )
print ( f " \n Recording: { result.recording_url } " )
print ( f "View in UI: { result.app_url } " )
else :
print ( f " \n Failure reason: { result.failure_reason } " )
asyncio.run(main())
Best Practices
1. Use Descriptive Run Titles
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
title = f "Invoice Download - { vendor_name } - { batch_date } " ,
parameters = { ... }
)
2. Set Reasonable Poll Intervals
# Good: 5-10 seconds
await asyncio.sleep( 5 )
# Too frequent: wastes API calls
await asyncio.sleep( 0.5 )
# Too slow: delays result processing
await asyncio.sleep( 60 )
3. Handle Errors Gracefully
try :
run = await client.run_workflow(
workflow_id = workflow_id,
parameters = parameters
)
except Exception as e:
print ( f "Failed to start workflow: { e } " )
return
result = await poll_until_complete(client, run.run_id)
if result.status == "completed" :
process_successful_run(result)
elif result.status == "failed" :
notify_failure(result.failure_reason)
retry_if_appropriate()
4. Use Webhooks for Production
# Instead of polling in production, use webhooks
run = await client.run_workflow(
workflow_id = "wpid_123456789" ,
parameters = { ... },
webhook_url = "https://your-server.com/webhook/workflow-complete"
)
# Your webhook handler receives the result
# No need to poll
5. Log Run IDs
run = await client.run_workflow(
workflow_id = workflow_id,
parameters = parameters
)
logger.info( f "Started workflow run { run.run_id } for { customer_name } " )
# Store run_id in your database for tracking
Next Steps
Workflow Overview Understand workflow architecture and concepts
Creating Workflows Build workflows with SDK, API, or UI
Workflow Blocks Complete reference of all block types
Webhooks Set up webhook authentication and retries