vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2026-07-09 16:09:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6
Skyvern/skyvern/client/client.py
Suchintan 720d47d754
SKY-11678 Add local credential vault persistence to OSS Docker (#7003) 
Co-authored-by: Codex <codex@openai.com>
2026-07-02 21:24:17 +00:00

6004 lines
196 KiB
Python
Raw Blame History

# This file was auto-generated by Fern from our API Definition.
from __future__ import annotations
import datetime as dt
import typing
import httpx
from .core.client_wrapper import AsyncClientWrapper, SyncClientWrapper
from .core.request_options import RequestOptions
from .environment import SkyvernEnvironment
from .raw_client import AsyncRawSkyvern, RawSkyvern
from .types.artifact import Artifact
from .types.artifact_type import ArtifactType
from .types.browser_profile import BrowserProfile
from .types.browser_session_response import BrowserSessionResponse
from .types.bulk_cancel_runs_response import BulkCancelRunsResponse
from .types.create_browser_session_request_proxy_location import CreateBrowserSessionRequestProxyLocation
from .types.create_credential_request_credential import CreateCredentialRequestCredential
from .types.create_script_response import CreateScriptResponse
from .types.credential_response import CredentialResponse
from .types.credential_vault_type import CredentialVaultType
from .types.extensions import Extensions
from .types.folder import Folder
from .types.get_run_response import GetRunResponse
from .types.otp_type import OtpType
from .types.persistent_browser_type import PersistentBrowserType
from .types.proxy_location import ProxyLocation
from .types.retry_run_webhook_request import RetryRunWebhookRequest
from .types.run_engine import RunEngine
from .types.run_sdk_action_request_action import RunSdkActionRequestAction
from .types.run_sdk_action_response import RunSdkActionResponse
from .types.run_status import RunStatus
from .types.run_webhook_replay_response import RunWebhookReplayResponse
from .types.script import Script
from .types.script_file_create import ScriptFileCreate
from .types.skyvern_forge_sdk_schemas_credentials_credential_type import SkyvernForgeSdkSchemasCredentialsCredentialType
from .types.skyvern_schemas_credential_type_credential_type import SkyvernSchemasCredentialTypeCredentialType
from .types.task_run_list_item import TaskRunListItem
from .types.task_run_request_input_data_extraction_schema import TaskRunRequestInputDataExtractionSchema
from .types.task_run_request_input_proxy_location import TaskRunRequestInputProxyLocation
from .types.task_run_response import TaskRunResponse
from .types.totp_code import TotpCode
from .types.upload_file_response import UploadFileResponse
from .types.workflow import Workflow
from .types.workflow_create_yaml_request import WorkflowCreateYamlRequest
from .types.workflow_run import WorkflowRun
from .types.workflow_run_request_input_proxy_location import WorkflowRunRequestInputProxyLocation
from .types.workflow_run_response import WorkflowRunResponse
from .types.workflow_run_status import WorkflowRunStatus
from .types.workflow_run_timeline import WorkflowRunTimeline
from .types.workflow_status import WorkflowStatus
if typing.TYPE_CHECKING:
from .artifacts.client import ArtifactsClient, AsyncArtifactsClient
from .schedules.client import AsyncSchedulesClient, SchedulesClient
from .scripts.client import AsyncScriptsClient, ScriptsClient
from .server.client import AsyncServerClient, ServerClient
from .workflows.client import AsyncWorkflowsClient, WorkflowsClient
# this is used as the default value for optional parameters
OMIT = typing.cast(typing.Any, ...)
class Skyvern:
"""
Use this class to access the different functions within the SDK. You can instantiate any number of clients with different configuration that will propagate to these functions.
Parameters
----------
base_url : typing.Optional[str]
The base url to use for requests from the client.
environment : SkyvernEnvironment
The environment to use for requests from the client. from .environment import SkyvernEnvironment
Defaults to SkyvernEnvironment.CLOUD
api_key : typing.Optional[str]
headers : typing.Optional[typing.Dict[str, str]]
Additional headers to send with every request.
timeout : typing.Optional[float]
The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx client is used, in which case this default is not enforced.
follow_redirects : typing.Optional[bool]
Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in.
httpx_client : typing.Optional[httpx.Client]
The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration.
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
"""
def __init__(
self,
*,
base_url: typing.Optional[str] = None,
environment: SkyvernEnvironment = SkyvernEnvironment.CLOUD,
api_key: typing.Optional[str] = None,
headers: typing.Optional[typing.Dict[str, str]] = None,
timeout: typing.Optional[float] = None,
follow_redirects: typing.Optional[bool] = True,
httpx_client: typing.Optional[httpx.Client] = None,
):
_defaulted_timeout = (
timeout if timeout is not None else 60 if httpx_client is None else httpx_client.timeout.read
)
self._client_wrapper = SyncClientWrapper(
base_url=_get_base_url(base_url=base_url, environment=environment),
api_key=api_key,
headers=headers,
httpx_client=httpx_client
if httpx_client is not None
else httpx.Client(timeout=_defaulted_timeout, follow_redirects=follow_redirects)
if follow_redirects is not None
else httpx.Client(timeout=_defaulted_timeout),
timeout=_defaulted_timeout,
)
self._raw_client = RawSkyvern(client_wrapper=self._client_wrapper)
self._artifacts: typing.Optional[ArtifactsClient] = None
self._server: typing.Optional[ServerClient] = None
self._workflows: typing.Optional[WorkflowsClient] = None
self._scripts: typing.Optional[ScriptsClient] = None
self._schedules: typing.Optional[SchedulesClient] = None
@property
def with_raw_response(self) -> RawSkyvern:
"""
Retrieves a raw implementation of this client that returns raw responses.
Returns
-------
RawSkyvern
"""
return self._raw_client
def run_task(
self,
*,
prompt: str,
user_agent: typing.Optional[str] = None,
url: typing.Optional[str] = OMIT,
engine: typing.Optional[RunEngine] = OMIT,
title: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[TaskRunRequestInputProxyLocation] = OMIT,
data_extraction_schema: typing.Optional[TaskRunRequestInputDataExtractionSchema] = OMIT,
error_code_mapping: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_steps: typing.Optional[int] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
model: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
publish_workflow: typing.Optional[bool] = OMIT,
include_action_history_in_verification: typing.Optional[bool] = OMIT,
max_screenshot_scrolls: typing.Optional[int] = OMIT,
browser_address: typing.Optional[str] = OMIT,
run_with: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> TaskRunResponse:
"""
Run a task
Parameters
----------
prompt : str
The goal or task description for Skyvern to accomplish
user_agent : typing.Optional[str]
url : typing.Optional[str]
The starting URL for the task. If not provided, Skyvern will attempt to determine an appropriate URL
engine : typing.Optional[RunEngine]
The engine that powers the agent task. The default value is `skyvern-1.0`, which is good for simple tasks like filling a form, or searching for information on Google. `skyvern-2.0` remains available for existing V2 workflows and explicitly requested V2 task runs. The `openai-cua` engine uses OpenAI's CUA model. The `anthropic-cua` uses Anthropic's Claude Sonnet 3.7 model with the computer use tool.
title : typing.Optional[str]
The title for the task
proxy_location : typing.Optional[TaskRunRequestInputProxyLocation]
Geographic Proxy location to route the browser traffic through. This is only available in Skyvern Cloud.
Available geotargeting options:
- RESIDENTIAL: the default value. Skyvern Cloud uses a random US residential proxy.
- RESIDENTIAL_ES: Spain
- RESIDENTIAL_IE: Ireland
- RESIDENTIAL_GB: United Kingdom
- RESIDENTIAL_IN: India
- RESIDENTIAL_JP: Japan
- RESIDENTIAL_FR: France
- RESIDENTIAL_DE: Germany
- RESIDENTIAL_NZ: New Zealand
- RESIDENTIAL_PH: Philippines
- RESIDENTIAL_KR: South Korea
- RESIDENTIAL_SA: Saudi Arabia
- RESIDENTIAL_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_BR: Brazil
- RESIDENTIAL_TR: Turkey
- RESIDENTIAL_CA: Canada
- RESIDENTIAL_MX: Mexico
- RESIDENTIAL_IT: Italy
- RESIDENTIAL_NL: Netherlands
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California (deprecated, routes through RESIDENTIAL_ISP)
- US-NY: New York (deprecated, routes through RESIDENTIAL_ISP)
- US-TX: Texas (deprecated, routes through RESIDENTIAL_ISP)
- US-FL: Florida (deprecated, routes through RESIDENTIAL_ISP)
- US-WA: Washington (deprecated, routes through RESIDENTIAL_ISP)
- NONE: No proxy
For self-hosted deployments, you can pass a custom proxy URL as a dict: {"url": "http://user:password@proxy.example.com:8080"}. This routes the browser through your own proxy server and takes precedence over any globally configured proxy pool.
Can also be a GeoTarget object for granular city/state targeting: {"country": "US", "subdivision": "CA", "city": "San Francisco"}
data_extraction_schema : typing.Optional[TaskRunRequestInputDataExtractionSchema]
The schema for data to be extracted from the webpage. If you're looking for consistent data schema being returned by the agent, it's highly recommended to use https://json-schema.org/.
error_code_mapping : typing.Optional[typing.Dict[str, typing.Optional[str]]]
Custom mapping of error codes to error messages if Skyvern encounters an error.
max_steps : typing.Optional[int]
Maximum number of steps the task can take. Task will fail if it exceeds this number. Cautions: you are charged per step so please set this number to a reasonable value. Contact sales@skyvern.com for custom pricing.
webhook_url : typing.Optional[str]
After a run is finished, send an update to this URL. Refer to https://www.skyvern.com/docs/running-tasks/webhooks-faq for more details.
totp_identifier : typing.Optional[str]
Identifier for the TOTP/2FA/MFA code when the code is pushed to Skyvern. Refer to https://www.skyvern.com/docs/credentials/totp#option-3-push-code-to-skyvern for more details.
totp_url : typing.Optional[str]
URL that serves TOTP/2FA/MFA codes for Skyvern to use during the workflow run. Refer to https://www.skyvern.com/docs/credentials/totp#option-2-get-code-from-your-endpoint for more details.
browser_session_id : typing.Optional[str]
Run the task or workflow in the specific Skyvern browser session. Having a browser session can persist the real-time state of the browser, so that the next run can continue from where the previous run left off.
model : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
Optional model configuration.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
The extra HTTP headers for the requests in browser.
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Use this for browser-provider auth (e.g., x-api-key for Skyvern Cloud, Browserless, or similar). These headers are NEVER forwarded to target websites.
publish_workflow : typing.Optional[bool]
Deprecated. Whether to publish a `skyvern-2.0` task as a reusable workflow. For backwards compatibility, this routes the request through the legacy `skyvern-2.0` publish path. Prefer creating reusable workflows through the workflow APIs.
include_action_history_in_verification : typing.Optional[bool]
Whether to include action history when verifying that the task is complete
max_screenshot_scrolls : typing.Optional[int]
The maximum number of scrolls for the post action screenshot. When it's None or 0, it takes the current viewpoint screenshot.
browser_address : typing.Optional[str]
The CDP address for the task.
run_with : typing.Optional[str]
Whether to run the task with agent or code. Null means use the default.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
TaskRunResponse
Successfully run task
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.run_task(
user_agent="x-user-agent",
prompt="Find the top 3 posts on Hacker News.",
)
"""
_response = self._raw_client.run_task(
prompt=prompt,
user_agent=user_agent,
url=url,
engine=engine,
title=title,
proxy_location=proxy_location,
data_extraction_schema=data_extraction_schema,
error_code_mapping=error_code_mapping,
max_steps=max_steps,
webhook_url=webhook_url,
totp_identifier=totp_identifier,
totp_url=totp_url,
browser_session_id=browser_session_id,
model=model,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
publish_workflow=publish_workflow,
include_action_history_in_verification=include_action_history_in_verification,
max_screenshot_scrolls=max_screenshot_scrolls,
browser_address=browser_address,
run_with=run_with,
request_options=request_options,
)
return _response.data
def run_workflow(
self,
*,
workflow_id: str,
template: typing.Optional[bool] = None,
max_steps_override: typing.Optional[int] = None,
user_agent: typing.Optional[str] = None,
parameters: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
title: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[WorkflowRunRequestInputProxyLocation] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
max_screenshot_scrolls: typing.Optional[int] = OMIT,
max_elapsed_time_minutes: typing.Optional[int] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
browser_address: typing.Optional[str] = OMIT,
ai_fallback: typing.Optional[bool] = OMIT,
run_with: typing.Optional[str] = OMIT,
run_metadata: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> WorkflowRunResponse:
"""
Run a workflow
Parameters
----------
workflow_id : str
ID of the workflow to run. Workflow ID starts with `wpid_`.
template : typing.Optional[bool]
max_steps_override : typing.Optional[int]
user_agent : typing.Optional[str]
parameters : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
Parameters to pass to the workflow
title : typing.Optional[str]
The title for this workflow run
proxy_location : typing.Optional[WorkflowRunRequestInputProxyLocation]
Geographic Proxy location to route the browser traffic through. This is only available in Skyvern Cloud.
Available geotargeting options:
- RESIDENTIAL: the default value. Skyvern Cloud uses a random US residential proxy.
- RESIDENTIAL_ES: Spain
- RESIDENTIAL_IE: Ireland
- RESIDENTIAL_GB: United Kingdom
- RESIDENTIAL_IN: India
- RESIDENTIAL_JP: Japan
- RESIDENTIAL_FR: France
- RESIDENTIAL_DE: Germany
- RESIDENTIAL_NZ: New Zealand
- RESIDENTIAL_PH: Philippines
- RESIDENTIAL_KR: South Korea
- RESIDENTIAL_SA: Saudi Arabia
- RESIDENTIAL_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_BR: Brazil
- RESIDENTIAL_TR: Turkey
- RESIDENTIAL_CA: Canada
- RESIDENTIAL_MX: Mexico
- RESIDENTIAL_IT: Italy
- RESIDENTIAL_NL: Netherlands
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California (deprecated, routes through RESIDENTIAL_ISP)
- US-NY: New York (deprecated, routes through RESIDENTIAL_ISP)
- US-TX: Texas (deprecated, routes through RESIDENTIAL_ISP)
- US-FL: Florida (deprecated, routes through RESIDENTIAL_ISP)
- US-WA: Washington (deprecated, routes through RESIDENTIAL_ISP)
- NONE: No proxy
For self-hosted deployments, you can pass a custom proxy URL as a dict: {"url": "http://user:password@proxy.example.com:8080"}. This routes the browser through your own proxy server and takes precedence over any globally configured proxy pool.
Can also be a GeoTarget object for granular city/state targeting: {"country": "US", "subdivision": "CA", "city": "San Francisco"}
webhook_url : typing.Optional[str]
URL to send workflow status updates to after a run is finished. Refer to https://www.skyvern.com/docs/running-tasks/webhooks-faq for webhook questions.
totp_url : typing.Optional[str]
URL that serves TOTP/2FA/MFA codes for Skyvern to use during the workflow run. Refer to https://www.skyvern.com/docs/credentials/totp#option-2-get-code-from-your-endpoint for more details.
totp_identifier : typing.Optional[str]
Identifier for the TOTP/2FA/MFA code when the code is pushed to Skyvern. Refer to https://www.skyvern.com/docs/credentials/totp#option-3-push-code-to-skyvern for more details.
browser_session_id : typing.Optional[str]
ID of a Skyvern browser session to reuse, having it continue from the current screen state
browser_profile_id : typing.Optional[str]
ID of a browser profile to reuse for this workflow run
max_screenshot_scrolls : typing.Optional[int]
The maximum number of scrolls for the post action screenshot. When it's None or 0, it takes the current viewpoint screenshot.
max_elapsed_time_minutes : typing.Optional[int]
Timeout this workflow run after the configured elapsed runtime in minutes. When omitted, the platform default is 240 minutes. The maximum configurable value is 480 minutes.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
The extra HTTP headers for the requests in browser.
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Use this for browser-provider auth (e.g., x-api-key for Skyvern Cloud, Browserless, or similar). These headers are NEVER forwarded to target websites.
browser_address : typing.Optional[str]
The CDP address for the workflow run.
ai_fallback : typing.Optional[bool]
Whether to fallback to AI if the workflow run fails.
run_with : typing.Optional[str]
Whether to run the workflow with agent or code. Null inherits from the workflow setting.
run_metadata : typing.Optional[typing.Dict[str, typing.Optional[str]]]
String key/value metadata to attach to this workflow run for analytics tag filtering.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
WorkflowRunResponse
Successfully run workflow
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.run_workflow(
max_steps_override=1,
user_agent="x-user-agent",
template=True,
workflow_id="wpid_123",
)
"""
_response = self._raw_client.run_workflow(
workflow_id=workflow_id,
template=template,
max_steps_override=max_steps_override,
user_agent=user_agent,
parameters=parameters,
title=title,
proxy_location=proxy_location,
webhook_url=webhook_url,
totp_url=totp_url,
totp_identifier=totp_identifier,
browser_session_id=browser_session_id,
browser_profile_id=browser_profile_id,
max_screenshot_scrolls=max_screenshot_scrolls,
max_elapsed_time_minutes=max_elapsed_time_minutes,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
browser_address=browser_address,
ai_fallback=ai_fallback,
run_with=run_with,
run_metadata=run_metadata,
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)
Parameters
----------
run_id : str
The id of the task run or the workflow run.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
GetRunResponse
Successfully got run
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_run(
run_id="tsk_123",
)
"""
_response = self._raw_client.get_run(run_id, request_options=request_options)
return _response.data
def cancel_run(
self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Cancel a run (task or workflow)
Parameters
----------
run_id : str
The id of the task run or the workflow run to cancel.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Optional[typing.Any]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.cancel_run(
run_id="run_id",
)
"""
_response = self._raw_client.cancel_run(run_id, request_options=request_options)
return _response.data
def bulk_cancel_runs(
self, *, run_ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None
) -> BulkCancelRunsResponse:
"""
Cancel multiple runs (tasks or workflows) in a single request
Parameters
----------
run_ids : typing.Sequence[str]
List of run IDs to cancel
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BulkCancelRunsResponse
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.bulk_cancel_runs(
run_ids=["run_ids"],
)
"""
_response = self._raw_client.bulk_cancel_runs(run_ids=run_ids, request_options=request_options)
return _response.data
def get_workflows(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
only_saved_tasks: typing.Optional[bool] = None,
only_workflows: typing.Optional[bool] = None,
only_templates: typing.Optional[bool] = None,
search_key: typing.Optional[str] = None,
title: typing.Optional[str] = None,
folder_id: typing.Optional[str] = None,
status: typing.Optional[typing.Union[WorkflowStatus, typing.Sequence[WorkflowStatus]]] = None,
template: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Workflow]:
"""
Get all workflows with the latest version for the organization.
Search semantics:
- If `search_key` is provided, its value is used as a unified search term for
`workflows.title`, `folders.title`, and workflow parameter metadata (key, description, and default_value for
`WorkflowParameterModel`).
- Falls back to deprecated `title` (title-only search) if `search_key` is not provided.
- Parameter metadata search excludes soft-deleted parameter rows across all parameter tables.
Parameters
----------
page : typing.Optional[int]
page_size : typing.Optional[int]
only_saved_tasks : typing.Optional[bool]
only_workflows : typing.Optional[bool]
only_templates : typing.Optional[bool]
search_key : typing.Optional[str]
Case-insensitive substring search across: workflow title, folder name, and parameter metadata (key, description, default_value). A workflow is returned if any of these fields match. Soft-deleted parameter definitions are excluded. Takes precedence over the deprecated `title` parameter.
title : typing.Optional[str]
Deprecated: use search_key instead. Falls back to title-only search if search_key is not provided.
folder_id : typing.Optional[str]
Filter workflows by folder ID
status : typing.Optional[typing.Union[WorkflowStatus, typing.Sequence[WorkflowStatus]]]
template : typing.Optional[bool]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Workflow]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_workflows(
page=1,
page_size=1,
only_saved_tasks=True,
only_workflows=True,
only_templates=True,
search_key="search_key",
title="title",
folder_id="folder_id",
template=True,
)
"""
_response = self._raw_client.get_workflows(
page=page,
page_size=page_size,
only_saved_tasks=only_saved_tasks,
only_workflows=only_workflows,
only_templates=only_templates,
search_key=search_key,
title=title,
folder_id=folder_id,
status=status,
template=template,
request_options=request_options,
)
return _response.data
def create_workflow(
self,
*,
folder_id: typing.Optional[str] = None,
json_definition: typing.Optional[WorkflowCreateYamlRequest] = OMIT,
yaml_definition: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Create a new workflow
Parameters
----------
folder_id : typing.Optional[str]
Optional folder ID to assign the workflow to
json_definition : typing.Optional[WorkflowCreateYamlRequest]
Workflow definition in JSON format
yaml_definition : typing.Optional[str]
Workflow definition in YAML format
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successfully created workflow
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.create_workflow(
folder_id="folder_id",
)
"""
_response = self._raw_client.create_workflow(
folder_id=folder_id,
json_definition=json_definition,
yaml_definition=yaml_definition,
request_options=request_options,
)
return _response.data
def update_workflow(
self,
workflow_id: str,
*,
json_definition: typing.Optional[WorkflowCreateYamlRequest] = OMIT,
yaml_definition: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Update a workflow
Parameters
----------
workflow_id : str
The ID of the workflow to update. Workflow ID starts with `wpid_`.
json_definition : typing.Optional[WorkflowCreateYamlRequest]
Workflow definition in JSON format
yaml_definition : typing.Optional[str]
Workflow definition in YAML format
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successfully updated workflow
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.update_workflow(
workflow_id="wpid_123",
)
"""
_response = self._raw_client.update_workflow(
workflow_id,
json_definition=json_definition,
yaml_definition=yaml_definition,
request_options=request_options,
)
return _response.data
def delete_workflow(
self, workflow_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Delete a workflow
Parameters
----------
workflow_id : str
The ID of the workflow to delete. Workflow ID starts with `wpid_`.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Optional[typing.Any]
Successfully deleted workflow
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.delete_workflow(
workflow_id="wpid_123",
)
"""
_response = self._raw_client.delete_workflow(workflow_id, request_options=request_options)
return _response.data
def get_folders(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
search: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Folder]:
"""
Get all folders for the organization
Parameters
----------
page : typing.Optional[int]
Page number
page_size : typing.Optional[int]
Number of folders per page
search : typing.Optional[str]
Search folders by title or description
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Folder]
Successfully retrieved folders
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_folders(
page=1,
page_size=1,
search="search",
)
"""
_response = self._raw_client.get_folders(
page=page, page_size=page_size, search=search, request_options=request_options
)
return _response.data
def create_folder(
self,
*,
title: str,
description: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Folder:
"""
Create a new folder to organize workflows
Parameters
----------
title : str
Folder title
description : typing.Optional[str]
Folder description
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Folder
Successfully created folder
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.create_folder(
title="title",
)
"""
_response = self._raw_client.create_folder(
title=title, description=description, request_options=request_options
)
return _response.data
def get_folder(self, folder_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Folder:
"""
Get a specific folder by ID
Parameters
----------
folder_id : str
Folder ID
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Folder
Successfully retrieved folder
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_folder(
folder_id="fld_123",
)
"""
_response = self._raw_client.get_folder(folder_id, request_options=request_options)
return _response.data
def update_folder(
self,
folder_id: str,
*,
title: typing.Optional[str] = OMIT,
description: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Folder:
"""
Update a folder's title or description
Parameters
----------
folder_id : str
Folder ID
title : typing.Optional[str]
Folder title
description : typing.Optional[str]
Folder description
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Folder
Successfully updated folder
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.update_folder(
folder_id="fld_123",
)
"""
_response = self._raw_client.update_folder(
folder_id, title=title, description=description, request_options=request_options
)
return _response.data
def delete_folder(
self,
folder_id: str,
*,
delete_workflows: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.Dict[str, typing.Optional[typing.Any]]:
"""
Delete a folder. Optionally delete all workflows in the folder.
Parameters
----------
folder_id : str
Folder ID
delete_workflows : typing.Optional[bool]
If true, also delete all workflows in this folder
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Dict[str, typing.Optional[typing.Any]]
Successfully deleted folder
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.delete_folder(
folder_id="fld_123",
delete_workflows=True,
)
"""
_response = self._raw_client.delete_folder(
folder_id, delete_workflows=delete_workflows, request_options=request_options
)
return _response.data
def update_workflow_folder(
self,
workflow_permanent_id: str,
*,
folder_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Update a workflow's folder assignment for the latest version
Parameters
----------
workflow_permanent_id : str
Workflow permanent ID
folder_id : typing.Optional[str]
Folder ID to assign workflow to. Set to null to remove from folder.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successfully updated workflow folder
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.update_workflow_folder(
workflow_permanent_id="wpid_123",
)
"""
_response = self._raw_client.update_workflow_folder(
workflow_permanent_id, folder_id=folder_id, request_options=request_options
)
return _response.data
def get_run_artifacts(
self,
run_id: str,
*,
artifact_type: typing.Optional[typing.Union[ArtifactType, typing.Sequence[ArtifactType]]] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Artifact]:
"""
Get artifacts for a run
Parameters
----------
run_id : str
The id of the task run or the workflow run.
artifact_type : typing.Optional[typing.Union[ArtifactType, typing.Sequence[ArtifactType]]]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Artifact]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_run_artifacts(
run_id="run_id",
)
"""
_response = self._raw_client.get_run_artifacts(
run_id, artifact_type=artifact_type, request_options=request_options
)
return _response.data
def retry_run_webhook(
self,
run_id: str,
*,
request: typing.Optional[RetryRunWebhookRequest] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> RunWebhookReplayResponse:
"""
Retry sending the webhook for a run
Parameters
----------
run_id : str
The id of the task run or the workflow run.
request : typing.Optional[RetryRunWebhookRequest]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
RunWebhookReplayResponse
Successful Response
Examples
--------
from skyvern import RetryRunWebhookRequest, Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.retry_run_webhook(
run_id="tsk_123",
request=RetryRunWebhookRequest(),
)
"""
_response = self._raw_client.retry_run_webhook(run_id, request=request, request_options=request_options)
return _response.data
def get_run_timeline(
self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[WorkflowRunTimeline]:
"""
Get timeline for a run (workflow run or task_v2 run)
Parameters
----------
run_id : str
The id of the workflow run or task_v2 run.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[WorkflowRunTimeline]
Successfully retrieved run timeline
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_run_timeline(
run_id="wr_123",
)
"""
_response = self._raw_client.get_run_timeline(run_id, request_options=request_options)
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",
max_steps_override=1,
user_agent="x-user-agent",
)
"""
_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_runs_v2(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
status: typing.Optional[typing.Union[RunStatus, typing.Sequence[RunStatus]]] = None,
search_key: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[TaskRunListItem]:
"""
Parameters
----------
page : typing.Optional[int]
page_size : typing.Optional[int]
status : typing.Optional[typing.Union[RunStatus, typing.Sequence[RunStatus]]]
search_key : typing.Optional[str]
Case-insensitive substring search (min 3 chars for trigram index).
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[TaskRunListItem]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_runs_v2(
page=1,
page_size=1,
search_key="search_key",
)
"""
_response = self._raw_client.get_runs_v2(
page=page, page_size=page_size, status=status, search_key=search_key, request_options=request_options
)
return _response.data
def get_workflow_runs(
self,
*,
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 workflow runs across all workflows for the current organization.
Results are paginated and can be filtered by **status**, **search_key**, and **error_code**. All filters are combined with **AND** logic — a run must match every supplied filter to be returned.
### search_key
A case-insensitive substring search that matches against **any** of the following fields:
| Searched field | Description |
|---|---|
| `workflow_run_id` | The unique run identifier (e.g. `wr_123…`) |
| Parameter **key** | The `key` of any workflow parameter definition associated with the run |
| Parameter **description** | The `description` of any workflow parameter definition |
| Run parameter **value** | The actual value supplied for any parameter when the run was created |
| `extra_http_headers` | Extra HTTP headers attached to the run (searched as raw JSON text) |
Soft-deleted parameter definitions are excluded from key/description matching. A run is returned if **any** of the fields above contain the search term.
### error_code
An **exact-match** filter against the `error_code` field inside each task's `errors` JSON array. A run matches if **any** of its tasks contains an error object with a matching `error_code` value. Error codes are user-defined strings set during workflow execution (e.g. `INVALID_CREDENTIALS`, `LOGIN_FAILED`, `CAPTCHA_DETECTED`).
### Combining filters
All query parameters use AND logic:
- `?status=failed` — only failed runs
- `?status=failed&error_code=LOGIN_FAILED` — failed runs **and** have a LOGIN_FAILED error
- `?status=failed&error_code=LOGIN_FAILED&search_key=prod_credential` — all three conditions must match
Parameters
----------
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. A run is returned if any of these fields match. Soft-deleted parameter definitions are excluded from key/description matching.
error_code : typing.Optional[str]
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.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[WorkflowRun]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_workflow_runs(
page=1,
page_size=1,
search_key="search_key",
error_code="error_code",
)
"""
_response = self._raw_client.get_workflow_runs(
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_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.
Supports filtering by **status**, **search_key**, and **error_code**. All filters are combined with **AND** logic.
### search_key
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.
### error_code
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`.
Parameters
----------
workflow_id : str
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. A run is returned if any of these fields match. Soft-deleted parameter definitions are excluded from key/description matching.
error_code : typing.Optional[str]
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.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[WorkflowRun]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_workflow_runs_by_id(
workflow_id="workflow_id",
page=1,
page_size=1,
search_key="search_key",
error_code="error_code",
)
"""
_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,
*,
version: typing.Optional[int] = None,
template: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Parameters
----------
workflow_permanent_id : str
version : typing.Optional[int]
template : typing.Optional[bool]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_workflow(
workflow_permanent_id="workflow_permanent_id",
version=1,
template=True,
)
"""
_response = self._raw_client.get_workflow(
workflow_permanent_id, version=version, template=template, request_options=request_options
)
return _response.data
def get_workflow_versions(
self,
workflow_permanent_id: str,
*,
template: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Workflow]:
"""
Get all versions of a workflow by its permanent ID.
Parameters
----------
workflow_permanent_id : str
template : typing.Optional[bool]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Workflow]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_workflow_versions(
workflow_permanent_id="workflow_permanent_id",
template=True,
)
"""
_response = self._raw_client.get_workflow_versions(
workflow_permanent_id, template=template, request_options=request_options
)
return _response.data
def upload_file(self, *, file: str, request_options: typing.Optional[RequestOptions] = None) -> UploadFileResponse:
"""
Parameters
----------
file : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
UploadFileResponse
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.upload_file(
file="file",
)
"""
_response = self._raw_client.upload_file(file=file, request_options=request_options)
return _response.data
def list_browser_profiles(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
include_deleted: typing.Optional[bool] = None,
search_key: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[BrowserProfile]:
"""
Get all browser profiles for the organization
Parameters
----------
page : typing.Optional[int]
page_size : typing.Optional[int]
include_deleted : typing.Optional[bool]
Include deleted browser profiles
search_key : typing.Optional[str]
Case-insensitive substring search across: browser profile name and description. A profile is returned if either field matches.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[BrowserProfile]
Successfully retrieved browser profiles
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.list_browser_profiles(
page=1,
page_size=1,
include_deleted=True,
search_key="search_key",
)
"""
_response = self._raw_client.list_browser_profiles(
page=page,
page_size=page_size,
include_deleted=include_deleted,
search_key=search_key,
request_options=request_options,
)
return _response.data
def create_browser_profile(
self,
*,
name: str,
description: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
workflow_run_id: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> BrowserProfile:
"""
Create a blank browser profile, or create one from a persistent browser session or workflow run.
Parameters
----------
name : str
Name for the browser profile
description : typing.Optional[str]
Optional profile description
browser_session_id : typing.Optional[str]
Persistent browser session to convert into a profile. Omit for a blank profile.
workflow_run_id : typing.Optional[str]
Workflow run whose persisted session should be captured. Omit for a blank profile.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserProfile
Successfully created browser profile
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.create_browser_profile(
name="name",
)
"""
_response = self._raw_client.create_browser_profile(
name=name,
description=description,
browser_session_id=browser_session_id,
workflow_run_id=workflow_run_id,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
request_options=request_options,
)
return _response.data
def get_browser_profile(
self, profile_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> BrowserProfile:
"""
Get a specific browser profile by ID
Parameters
----------
profile_id : str
The ID of the browser profile. browser_profile_id starts with `bp_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserProfile
Successfully retrieved browser profile
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_browser_profile(
profile_id="bp_123456",
)
"""
_response = self._raw_client.get_browser_profile(profile_id, request_options=request_options)
return _response.data
def delete_browser_profile(
self, profile_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> None:
"""
Delete a browser profile (soft delete)
Parameters
----------
profile_id : str
The ID of the browser profile to delete. browser_profile_id starts with `bp_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
None
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.delete_browser_profile(
profile_id="bp_123456",
)
"""
_response = self._raw_client.delete_browser_profile(profile_id, request_options=request_options)
return _response.data
def update_browser_profile(
self,
profile_id: str,
*,
name: typing.Optional[str] = OMIT,
description: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
rotate_proxy_session_id: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> BrowserProfile:
"""
Update a browser profile's name and/or description
Parameters
----------
profile_id : str
The ID of the browser profile to update. browser_profile_id starts with `bp_`
name : typing.Optional[str]
New name for the browser profile
description : typing.Optional[str]
New description for the browser profile
rotate_proxy_session_id : typing.Optional[bool]
Rotate the Skyvern-managed proxy sticky-session id for this browser profile.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserProfile
Successfully updated browser profile
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.update_browser_profile(
profile_id="bp_123456",
)
"""
_response = self._raw_client.update_browser_profile(
profile_id,
name=name,
description=description,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
rotate_proxy_session_id=rotate_proxy_session_id,
request_options=request_options,
)
return _response.data
def get_browser_sessions(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[BrowserSessionResponse]:
"""
Get all active browser sessions for the organization
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[BrowserSessionResponse]
Successfully retrieved all active browser sessions
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_browser_sessions()
"""
_response = self._raw_client.get_browser_sessions(request_options=request_options)
return _response.data
def create_browser_session(
self,
*,
timeout: typing.Optional[int] = OMIT,
proxy_location: typing.Optional[CreateBrowserSessionRequestProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
extensions: typing.Optional[typing.Sequence[Extensions]] = OMIT,
browser_type: typing.Optional[PersistentBrowserType] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> BrowserSessionResponse:
"""
Create a browser session that persists across multiple runs
Parameters
----------
timeout : typing.Optional[int]
Timeout in minutes for the session. Timeout is applied after the session is started. Must be between 5 and 1440. Defaults to 60.
proxy_location : typing.Optional[CreateBrowserSessionRequestProxyLocation]
Geographic Proxy location to route the browser traffic through. This is only available in Skyvern Cloud.
Available geotargeting options:
- RESIDENTIAL: the default value. Skyvern Cloud uses a random US residential proxy.
- RESIDENTIAL_ES: Spain
- RESIDENTIAL_IE: Ireland
- RESIDENTIAL_GB: United Kingdom
- RESIDENTIAL_IN: India
- RESIDENTIAL_JP: Japan
- RESIDENTIAL_FR: France
- RESIDENTIAL_DE: Germany
- RESIDENTIAL_NZ: New Zealand
- RESIDENTIAL_PH: Philippines
- RESIDENTIAL_KR: South Korea
- RESIDENTIAL_SA: Saudi Arabia
- RESIDENTIAL_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_BR: Brazil
- RESIDENTIAL_TR: Turkey
- RESIDENTIAL_CA: Canada
- RESIDENTIAL_MX: Mexico
- RESIDENTIAL_IT: Italy
- RESIDENTIAL_NL: Netherlands
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California (deprecated, routes through RESIDENTIAL_ISP)
- US-NY: New York (deprecated, routes through RESIDENTIAL_ISP)
- US-TX: Texas (deprecated, routes through RESIDENTIAL_ISP)
- US-FL: Florida (deprecated, routes through RESIDENTIAL_ISP)
- US-WA: Washington (deprecated, routes through RESIDENTIAL_ISP)
- NONE: No proxy
For self-hosted deployments, you can pass a custom proxy URL as a dict: {"url": "http://user:password@proxy.example.com:8080"}. This routes the browser through your own proxy server and takes precedence over any globally configured proxy pool.
Can also be a GeoTarget object for granular city/state targeting: {"country": "US", "subdivision": "CA", "city": "San Francisco"}, or a custom proxy URL dict for self-hosted deployments: {"url": "http://user:password@proxy.example.com:8080"}
proxy_session_id : typing.Optional[str]
Opaque Skyvern-managed proxy sticky-session id for pinned Residential ISP sessions.
extensions : typing.Optional[typing.Sequence[Extensions]]
A list of extensions to install in the browser session.
browser_type : typing.Optional[PersistentBrowserType]
The type of browser to use for the session.
browser_profile_id : typing.Optional[str]
ID of a browser profile to load into this session (restores cookies, localStorage, etc.). browser_profile_id starts with `bp_`.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserSessionResponse
Successfully created browser session
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.create_browser_session()
"""
_response = self._raw_client.create_browser_session(
timeout=timeout,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
extensions=extensions,
browser_type=browser_type,
browser_profile_id=browser_profile_id,
request_options=request_options,
)
return _response.data
def close_browser_session(
self, browser_session_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Close a session. Once closed, the session cannot be used again.
Parameters
----------
browser_session_id : str
The ID of the browser session to close. completed_at will be set when the browser session is closed. browser_session_id starts with `pbs_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Optional[typing.Any]
Successfully closed browser session
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.close_browser_session(
browser_session_id="pbs_123456",
)
"""
_response = self._raw_client.close_browser_session(browser_session_id, request_options=request_options)
return _response.data
def get_browser_session(
self, browser_session_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> BrowserSessionResponse:
"""
Get details about a specific browser session, including the browser address for cdp connection.
Parameters
----------
browser_session_id : str
The ID of the browser session. browser_session_id starts with `pbs_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserSessionResponse
Successfully retrieved browser session details
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_browser_session(
browser_session_id="pbs_123456",
)
"""
_response = self._raw_client.get_browser_session(browser_session_id, request_options=request_options)
return _response.data
def send_totp_code(
self,
*,
totp_identifier: str,
content: str,
task_id: typing.Optional[str] = OMIT,
workflow_id: typing.Optional[str] = OMIT,
workflow_run_id: typing.Optional[str] = OMIT,
source: typing.Optional[str] = OMIT,
expired_at: typing.Optional[dt.datetime] = OMIT,
type: typing.Optional[OtpType] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> TotpCode:
"""
Forward a TOTP (2FA, MFA) email or sms message containing the code to Skyvern. This endpoint stores the code in database so that Skyvern can use it while running tasks/workflows.
Parameters
----------
totp_identifier : str
The identifier of the TOTP code. It can be the email address, phone number, or the identifier of the user.
content : str
The content of the TOTP code. It can be the email content that contains the TOTP code, or the sms message that contains the TOTP code. Skyvern will automatically extract the TOTP code from the content.
task_id : typing.Optional[str]
The task_id the totp code is for. It can be the task_id of the task that the TOTP code is for.
workflow_id : typing.Optional[str]
The workflow ID the TOTP code is for. It can be the workflow ID of the workflow that the TOTP code is for.
workflow_run_id : typing.Optional[str]
The workflow run id that the TOTP code is for. It can be the workflow run id of the workflow run that the TOTP code is for.
source : typing.Optional[str]
An optional field. The source of the TOTP code. e.g. email, sms, etc.
expired_at : typing.Optional[dt.datetime]
The timestamp when the TOTP code expires
type : typing.Optional[OtpType]
Optional. If provided, forces extraction of this specific OTP type (totp or magic_link). Use this when the content contains multiple OTP types and you want to specify which one to extract.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
TotpCode
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.send_totp_code(
totp_identifier="john.doe@example.com",
content="Hello, your verification code is 123456",
)
"""
_response = self._raw_client.send_totp_code(
totp_identifier=totp_identifier,
content=content,
task_id=task_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
source=source,
expired_at=expired_at,
type=type,
request_options=request_options,
)
return _response.data
def get_credentials(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
vault_type: typing.Optional[CredentialVaultType] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[CredentialResponse]:
"""
Retrieves a paginated list of credentials for the current organization
Parameters
----------
page : typing.Optional[int]
Page number for pagination
page_size : typing.Optional[int]
Number of items per page
vault_type : typing.Optional[CredentialVaultType]
Filter credentials by vault type (e.g. 'skyvern', 'custom', 'bitwarden', 'azure_vault')
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[CredentialResponse]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_credentials(
page=1,
page_size=10,
vault_type="bitwarden",
)
"""
_response = self._raw_client.get_credentials(
page=page, page_size=page_size, vault_type=vault_type, request_options=request_options
)
return _response.data
def create_credential(
self,
*,
name: str,
credential_type: SkyvernForgeSdkSchemasCredentialsCredentialType,
credential: CreateCredentialRequestCredential,
vault_type: typing.Optional[CredentialVaultType] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
rotate_proxy_session_id: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> CredentialResponse:
"""
Creates a new credential for the current organization
Parameters
----------
name : str
Name of the credential
credential_type : SkyvernForgeSdkSchemasCredentialsCredentialType
Type of credential to create
credential : CreateCredentialRequestCredential
The credential data to store
vault_type : typing.Optional[CredentialVaultType]
Which vault to store this credential in. If omitted, uses the instance default. Use this to mix Skyvern-hosted and custom credentials within the same organization.
rotate_proxy_session_id : typing.Optional[bool]
Rotate the Skyvern-managed proxy sticky-session id for this credential.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CredentialResponse
Successful Response
Examples
--------
from skyvern import NonEmptyPasswordCredential, Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.create_credential(
name="Amazon Login",
credential_type="password",
credential=NonEmptyPasswordCredential(
password="securepassword123",
username="user@example.com",
),
)
"""
_response = self._raw_client.create_credential(
name=name,
credential_type=credential_type,
credential=credential,
vault_type=vault_type,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
rotate_proxy_session_id=rotate_proxy_session_id,
request_options=request_options,
)
return _response.data
def update_credential(
self,
credential_id: str,
*,
name: str,
credential_type: SkyvernForgeSdkSchemasCredentialsCredentialType,
credential: CreateCredentialRequestCredential,
vault_type: typing.Optional[CredentialVaultType] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
rotate_proxy_session_id: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> CredentialResponse:
"""
Overwrites the stored credential data (e.g. username/password) while keeping the same credential_id.
Parameters
----------
credential_id : str
The unique identifier of the credential to update
name : str
Name of the credential
credential_type : SkyvernForgeSdkSchemasCredentialsCredentialType
Type of credential to create
credential : CreateCredentialRequestCredential
The credential data to store
vault_type : typing.Optional[CredentialVaultType]
Which vault to store this credential in. If omitted, uses the instance default. Use this to mix Skyvern-hosted and custom credentials within the same organization.
rotate_proxy_session_id : typing.Optional[bool]
Rotate the Skyvern-managed proxy sticky-session id for this credential.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CredentialResponse
Successful Response
Examples
--------
from skyvern import NonEmptyPasswordCredential, Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.update_credential(
credential_id="cred_1234567890",
name="Amazon Login",
credential_type="password",
credential=NonEmptyPasswordCredential(
password="securepassword123",
username="user@example.com",
),
)
"""
_response = self._raw_client.update_credential(
credential_id,
name=name,
credential_type=credential_type,
credential=credential,
vault_type=vault_type,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
rotate_proxy_session_id=rotate_proxy_session_id,
request_options=request_options,
)
return _response.data
def delete_credential(self, credential_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> None:
"""
Deletes a specific credential by its ID
Parameters
----------
credential_id : str
The unique identifier of the credential to delete
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
None
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.delete_credential(
credential_id="cred_1234567890",
)
"""
_response = self._raw_client.delete_credential(credential_id, request_options=request_options)
return _response.data
def get_credential(
self, credential_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> CredentialResponse:
"""
Retrieves a specific credential by its ID
Parameters
----------
credential_id : str
The unique identifier of the credential
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CredentialResponse
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_credential(
credential_id="cred_1234567890",
)
"""
_response = self._raw_client.get_credential(credential_id, request_options=request_options)
return _response.data
def login(
self,
*,
credential_type: SkyvernSchemasCredentialTypeCredentialType,
user_agent: typing.Optional[str] = None,
url: typing.Optional[str] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
browser_address: typing.Optional[str] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_screenshot_scrolling_times: typing.Optional[int] = OMIT,
prompt: typing.Optional[str] = OMIT,
credential_id: typing.Optional[str] = OMIT,
bitwarden_collection_id: typing.Optional[str] = OMIT,
bitwarden_item_id: typing.Optional[str] = OMIT,
onepassword_vault_id: typing.Optional[str] = OMIT,
onepassword_item_id: typing.Optional[str] = OMIT,
azure_vault_name: typing.Optional[str] = OMIT,
azure_vault_username_key: typing.Optional[str] = OMIT,
azure_vault_password_key: typing.Optional[str] = OMIT,
azure_vault_totp_secret_key: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> WorkflowRunResponse:
"""
Log in to a website using either credential stored in Skyvern, Bitwarden, 1Password, or Azure Vault
Parameters
----------
credential_type : SkyvernSchemasCredentialTypeCredentialType
Where to get the credential from
user_agent : typing.Optional[str]
url : typing.Optional[str]
Website URL
webhook_url : typing.Optional[str]
Webhook URL to send status updates
proxy_location : typing.Optional[ProxyLocation]
Proxy location to use
totp_identifier : typing.Optional[str]
Identifier for TOTP (Time-based One-Time Password) if required
totp_url : typing.Optional[str]
TOTP URL to fetch one-time passwords
browser_session_id : typing.Optional[str]
ID of the browser session to use, which is prefixed by `pbs_` e.g. `pbs_123456`
browser_profile_id : typing.Optional[str]
ID of a browser profile to reuse for this run
browser_address : typing.Optional[str]
The CDP address for the task.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
Additional HTTP headers to include in requests
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Never forwarded to target websites.
max_screenshot_scrolling_times : typing.Optional[int]
Maximum number of times to scroll for screenshots
prompt : typing.Optional[str]
Login instructions. Skyvern has default prompt/instruction for login if this field is not provided.
credential_id : typing.Optional[str]
ID of the Skyvern credential to use for login.
bitwarden_collection_id : typing.Optional[str]
Bitwarden collection ID. You can find it in the Bitwarden collection URL. e.g. `https://vault.bitwarden.com/vaults/collection_id/items`
bitwarden_item_id : typing.Optional[str]
Bitwarden item ID
onepassword_vault_id : typing.Optional[str]
1Password vault ID
onepassword_item_id : typing.Optional[str]
1Password item ID
azure_vault_name : typing.Optional[str]
Azure Vault Name
azure_vault_username_key : typing.Optional[str]
Azure Vault username key
azure_vault_password_key : typing.Optional[str]
Azure Vault password key
azure_vault_totp_secret_key : typing.Optional[str]
Azure Vault TOTP secret key
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
WorkflowRunResponse
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.login(
user_agent="x-user-agent",
credential_type="skyvern",
)
"""
_response = self._raw_client.login(
credential_type=credential_type,
user_agent=user_agent,
url=url,
webhook_url=webhook_url,
proxy_location=proxy_location,
totp_identifier=totp_identifier,
totp_url=totp_url,
browser_session_id=browser_session_id,
browser_profile_id=browser_profile_id,
browser_address=browser_address,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
max_screenshot_scrolling_times=max_screenshot_scrolling_times,
prompt=prompt,
credential_id=credential_id,
bitwarden_collection_id=bitwarden_collection_id,
bitwarden_item_id=bitwarden_item_id,
onepassword_vault_id=onepassword_vault_id,
onepassword_item_id=onepassword_item_id,
azure_vault_name=azure_vault_name,
azure_vault_username_key=azure_vault_username_key,
azure_vault_password_key=azure_vault_password_key,
azure_vault_totp_secret_key=azure_vault_totp_secret_key,
request_options=request_options,
)
return _response.data
def download_files(
self,
*,
navigation_goal: str,
user_agent: typing.Optional[str] = None,
url: typing.Optional[str] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
browser_address: typing.Optional[str] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_screenshot_scrolling_times: typing.Optional[int] = OMIT,
download_suffix: typing.Optional[str] = OMIT,
download_timeout: typing.Optional[float] = OMIT,
max_steps_per_run: typing.Optional[int] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> WorkflowRunResponse:
"""
Download a file from a website by navigating and clicking download buttons
Parameters
----------
navigation_goal : str
Instructions for navigating to and downloading the file
user_agent : typing.Optional[str]
url : typing.Optional[str]
Website URL
webhook_url : typing.Optional[str]
Webhook URL to send status updates
proxy_location : typing.Optional[ProxyLocation]
Proxy location to use
totp_identifier : typing.Optional[str]
Identifier for TOTP (Time-based One-Time Password) if required
totp_url : typing.Optional[str]
TOTP URL to fetch one-time passwords
browser_session_id : typing.Optional[str]
ID of the browser session to use, which is prefixed by `pbs_` e.g. `pbs_123456`
browser_profile_id : typing.Optional[str]
ID of a browser profile to reuse for this run
browser_address : typing.Optional[str]
The CDP address for the task.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
Additional HTTP headers to include in requests
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Never forwarded to target websites.
max_screenshot_scrolling_times : typing.Optional[int]
Maximum number of times to scroll for screenshots
download_suffix : typing.Optional[str]
Suffix or complete filename for the downloaded file
download_timeout : typing.Optional[float]
Timeout in seconds for the download operation
max_steps_per_run : typing.Optional[int]
Maximum number of steps to execute
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
WorkflowRunResponse
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.download_files(
user_agent="x-user-agent",
navigation_goal="navigation_goal",
)
"""
_response = self._raw_client.download_files(
navigation_goal=navigation_goal,
user_agent=user_agent,
url=url,
webhook_url=webhook_url,
proxy_location=proxy_location,
totp_identifier=totp_identifier,
totp_url=totp_url,
browser_session_id=browser_session_id,
browser_profile_id=browser_profile_id,
browser_address=browser_address,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
max_screenshot_scrolling_times=max_screenshot_scrolling_times,
download_suffix=download_suffix,
download_timeout=download_timeout,
max_steps_per_run=max_steps_per_run,
request_options=request_options,
)
return _response.data
def get_scripts(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Script]:
"""
Retrieves a paginated list of scripts for the current organization
Parameters
----------
page : typing.Optional[int]
Page number for pagination
page_size : typing.Optional[int]
Number of items per page
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Script]
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_scripts(
page=1,
page_size=10,
)
"""
_response = self._raw_client.get_scripts(page=page, page_size=page_size, request_options=request_options)
return _response.data
def create_script(
self,
*,
workflow_id: typing.Optional[str] = OMIT,
run_id: typing.Optional[str] = OMIT,
files: typing.Optional[typing.Sequence[ScriptFileCreate]] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> CreateScriptResponse:
"""
Create a new script with optional files and metadata
Parameters
----------
workflow_id : typing.Optional[str]
Associated workflow ID
run_id : typing.Optional[str]
Associated run ID
files : typing.Optional[typing.Sequence[ScriptFileCreate]]
Array of files to include in the script
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CreateScriptResponse
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.create_script()
"""
_response = self._raw_client.create_script(
workflow_id=workflow_id, run_id=run_id, files=files, request_options=request_options
)
return _response.data
def get_script(self, script_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Script:
"""
Retrieves a specific script by its ID
Parameters
----------
script_id : str
The unique identifier of the script
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Script
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.get_script(
script_id="s_abc123",
)
"""
_response = self._raw_client.get_script(script_id, request_options=request_options)
return _response.data
def deploy_script(
self,
script_id: str,
*,
files: typing.Sequence[ScriptFileCreate],
request_options: typing.Optional[RequestOptions] = None,
) -> CreateScriptResponse:
"""
Deploy a script with updated files, creating a new version
Parameters
----------
script_id : str
The unique identifier of the script
files : typing.Sequence[ScriptFileCreate]
Array of files to include in the script
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CreateScriptResponse
Successful Response
Examples
--------
from skyvern import ScriptFileCreate, Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.deploy_script(
script_id="s_abc123",
files=[
ScriptFileCreate(
path="src/main.py",
content="content",
)
],
)
"""
_response = self._raw_client.deploy_script(script_id, files=files, request_options=request_options)
return _response.data
def run_sdk_action(
self,
*,
url: str,
action: RunSdkActionRequestAction,
browser_session_id: typing.Optional[str] = OMIT,
browser_address: typing.Optional[str] = OMIT,
workflow_run_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> RunSdkActionResponse:
"""
Execute a single SDK action with the specified parameters
Parameters
----------
url : str
The URL where the action should be executed
action : RunSdkActionRequestAction
The action to execute with its specific parameters
browser_session_id : typing.Optional[str]
The browser session ID
browser_address : typing.Optional[str]
The browser address
workflow_run_id : typing.Optional[str]
Optional workflow run ID to continue an existing workflow run
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
RunSdkActionResponse
Successful Response
Examples
--------
from skyvern import RunSdkActionRequestAction_AiAct, Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
)
client.run_sdk_action(
url="url",
action=RunSdkActionRequestAction_AiAct(),
)
"""
_response = self._raw_client.run_sdk_action(
url=url,
action=action,
browser_session_id=browser_session_id,
browser_address=browser_address,
workflow_run_id=workflow_run_id,
request_options=request_options,
)
return _response.data
@property
def artifacts(self):
if self._artifacts is None:
from .artifacts.client import ArtifactsClient # noqa: E402
self._artifacts = ArtifactsClient(client_wrapper=self._client_wrapper)
return self._artifacts
@property
def server(self):
if self._server is None:
from .server.client import ServerClient # noqa: E402
self._server = ServerClient(client_wrapper=self._client_wrapper)
return self._server
@property
def workflows(self):
if self._workflows is None:
from .workflows.client import WorkflowsClient # noqa: E402
self._workflows = WorkflowsClient(client_wrapper=self._client_wrapper)
return self._workflows
@property
def scripts(self):
if self._scripts is None:
from .scripts.client import ScriptsClient # noqa: E402
self._scripts = ScriptsClient(client_wrapper=self._client_wrapper)
return self._scripts
@property
def schedules(self):
if self._schedules is None:
from .schedules.client import SchedulesClient # noqa: E402
self._schedules = SchedulesClient(client_wrapper=self._client_wrapper)
return self._schedules
class AsyncSkyvern:
"""
Use this class to access the different functions within the SDK. You can instantiate any number of clients with different configuration that will propagate to these functions.
Parameters
----------
base_url : typing.Optional[str]
The base url to use for requests from the client.
environment : SkyvernEnvironment
The environment to use for requests from the client. from .environment import SkyvernEnvironment
Defaults to SkyvernEnvironment.CLOUD
api_key : typing.Optional[str]
headers : typing.Optional[typing.Dict[str, str]]
Additional headers to send with every request.
timeout : typing.Optional[float]
The timeout to be used, in seconds, for requests. By default the timeout is 60 seconds, unless a custom httpx client is used, in which case this default is not enforced.
follow_redirects : typing.Optional[bool]
Whether the default httpx client follows redirects or not, this is irrelevant if a custom httpx client is passed in.
httpx_client : typing.Optional[httpx.AsyncClient]
The httpx client to use for making requests, a preconfigured client is used by default, however this is useful should you want to pass in any custom httpx configuration.
Examples
--------
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
"""
def __init__(
self,
*,
base_url: typing.Optional[str] = None,
environment: SkyvernEnvironment = SkyvernEnvironment.CLOUD,
api_key: typing.Optional[str] = None,
headers: typing.Optional[typing.Dict[str, str]] = None,
timeout: typing.Optional[float] = None,
follow_redirects: typing.Optional[bool] = True,
httpx_client: typing.Optional[httpx.AsyncClient] = None,
):
_defaulted_timeout = (
timeout if timeout is not None else 60 if httpx_client is None else httpx_client.timeout.read
)
self._client_wrapper = AsyncClientWrapper(
base_url=_get_base_url(base_url=base_url, environment=environment),
api_key=api_key,
headers=headers,
httpx_client=httpx_client
if httpx_client is not None
else httpx.AsyncClient(timeout=_defaulted_timeout, follow_redirects=follow_redirects)
if follow_redirects is not None
else httpx.AsyncClient(timeout=_defaulted_timeout),
timeout=_defaulted_timeout,
)
self._raw_client = AsyncRawSkyvern(client_wrapper=self._client_wrapper)
self._artifacts: typing.Optional[AsyncArtifactsClient] = None
self._server: typing.Optional[AsyncServerClient] = None
self._workflows: typing.Optional[AsyncWorkflowsClient] = None
self._scripts: typing.Optional[AsyncScriptsClient] = None
self._schedules: typing.Optional[AsyncSchedulesClient] = None
@property
def with_raw_response(self) -> AsyncRawSkyvern:
"""
Retrieves a raw implementation of this client that returns raw responses.
Returns
-------
AsyncRawSkyvern
"""
return self._raw_client
async def run_task(
self,
*,
prompt: str,
user_agent: typing.Optional[str] = None,
url: typing.Optional[str] = OMIT,
engine: typing.Optional[RunEngine] = OMIT,
title: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[TaskRunRequestInputProxyLocation] = OMIT,
data_extraction_schema: typing.Optional[TaskRunRequestInputDataExtractionSchema] = OMIT,
error_code_mapping: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_steps: typing.Optional[int] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
model: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
publish_workflow: typing.Optional[bool] = OMIT,
include_action_history_in_verification: typing.Optional[bool] = OMIT,
max_screenshot_scrolls: typing.Optional[int] = OMIT,
browser_address: typing.Optional[str] = OMIT,
run_with: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> TaskRunResponse:
"""
Run a task
Parameters
----------
prompt : str
The goal or task description for Skyvern to accomplish
user_agent : typing.Optional[str]
url : typing.Optional[str]
The starting URL for the task. If not provided, Skyvern will attempt to determine an appropriate URL
engine : typing.Optional[RunEngine]
The engine that powers the agent task. The default value is `skyvern-1.0`, which is good for simple tasks like filling a form, or searching for information on Google. `skyvern-2.0` remains available for existing V2 workflows and explicitly requested V2 task runs. The `openai-cua` engine uses OpenAI's CUA model. The `anthropic-cua` uses Anthropic's Claude Sonnet 3.7 model with the computer use tool.
title : typing.Optional[str]
The title for the task
proxy_location : typing.Optional[TaskRunRequestInputProxyLocation]
Geographic Proxy location to route the browser traffic through. This is only available in Skyvern Cloud.
Available geotargeting options:
- RESIDENTIAL: the default value. Skyvern Cloud uses a random US residential proxy.
- RESIDENTIAL_ES: Spain
- RESIDENTIAL_IE: Ireland
- RESIDENTIAL_GB: United Kingdom
- RESIDENTIAL_IN: India
- RESIDENTIAL_JP: Japan
- RESIDENTIAL_FR: France
- RESIDENTIAL_DE: Germany
- RESIDENTIAL_NZ: New Zealand
- RESIDENTIAL_PH: Philippines
- RESIDENTIAL_KR: South Korea
- RESIDENTIAL_SA: Saudi Arabia
- RESIDENTIAL_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_BR: Brazil
- RESIDENTIAL_TR: Turkey
- RESIDENTIAL_CA: Canada
- RESIDENTIAL_MX: Mexico
- RESIDENTIAL_IT: Italy
- RESIDENTIAL_NL: Netherlands
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California (deprecated, routes through RESIDENTIAL_ISP)
- US-NY: New York (deprecated, routes through RESIDENTIAL_ISP)
- US-TX: Texas (deprecated, routes through RESIDENTIAL_ISP)
- US-FL: Florida (deprecated, routes through RESIDENTIAL_ISP)
- US-WA: Washington (deprecated, routes through RESIDENTIAL_ISP)
- NONE: No proxy
For self-hosted deployments, you can pass a custom proxy URL as a dict: {"url": "http://user:password@proxy.example.com:8080"}. This routes the browser through your own proxy server and takes precedence over any globally configured proxy pool.
Can also be a GeoTarget object for granular city/state targeting: {"country": "US", "subdivision": "CA", "city": "San Francisco"}
data_extraction_schema : typing.Optional[TaskRunRequestInputDataExtractionSchema]
The schema for data to be extracted from the webpage. If you're looking for consistent data schema being returned by the agent, it's highly recommended to use https://json-schema.org/.
error_code_mapping : typing.Optional[typing.Dict[str, typing.Optional[str]]]
Custom mapping of error codes to error messages if Skyvern encounters an error.
max_steps : typing.Optional[int]
Maximum number of steps the task can take. Task will fail if it exceeds this number. Cautions: you are charged per step so please set this number to a reasonable value. Contact sales@skyvern.com for custom pricing.
webhook_url : typing.Optional[str]
After a run is finished, send an update to this URL. Refer to https://www.skyvern.com/docs/running-tasks/webhooks-faq for more details.
totp_identifier : typing.Optional[str]
Identifier for the TOTP/2FA/MFA code when the code is pushed to Skyvern. Refer to https://www.skyvern.com/docs/credentials/totp#option-3-push-code-to-skyvern for more details.
totp_url : typing.Optional[str]
URL that serves TOTP/2FA/MFA codes for Skyvern to use during the workflow run. Refer to https://www.skyvern.com/docs/credentials/totp#option-2-get-code-from-your-endpoint for more details.
browser_session_id : typing.Optional[str]
Run the task or workflow in the specific Skyvern browser session. Having a browser session can persist the real-time state of the browser, so that the next run can continue from where the previous run left off.
model : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
Optional model configuration.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
The extra HTTP headers for the requests in browser.
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Use this for browser-provider auth (e.g., x-api-key for Skyvern Cloud, Browserless, or similar). These headers are NEVER forwarded to target websites.
publish_workflow : typing.Optional[bool]
Deprecated. Whether to publish a `skyvern-2.0` task as a reusable workflow. For backwards compatibility, this routes the request through the legacy `skyvern-2.0` publish path. Prefer creating reusable workflows through the workflow APIs.
include_action_history_in_verification : typing.Optional[bool]
Whether to include action history when verifying that the task is complete
max_screenshot_scrolls : typing.Optional[int]
The maximum number of scrolls for the post action screenshot. When it's None or 0, it takes the current viewpoint screenshot.
browser_address : typing.Optional[str]
The CDP address for the task.
run_with : typing.Optional[str]
Whether to run the task with agent or code. Null means use the default.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
TaskRunResponse
Successfully run task
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.run_task(
user_agent="x-user-agent",
prompt="Find the top 3 posts on Hacker News.",
)
asyncio.run(main())
"""
_response = await self._raw_client.run_task(
prompt=prompt,
user_agent=user_agent,
url=url,
engine=engine,
title=title,
proxy_location=proxy_location,
data_extraction_schema=data_extraction_schema,
error_code_mapping=error_code_mapping,
max_steps=max_steps,
webhook_url=webhook_url,
totp_identifier=totp_identifier,
totp_url=totp_url,
browser_session_id=browser_session_id,
model=model,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
publish_workflow=publish_workflow,
include_action_history_in_verification=include_action_history_in_verification,
max_screenshot_scrolls=max_screenshot_scrolls,
browser_address=browser_address,
run_with=run_with,
request_options=request_options,
)
return _response.data
async def run_workflow(
self,
*,
workflow_id: str,
template: typing.Optional[bool] = None,
max_steps_override: typing.Optional[int] = None,
user_agent: typing.Optional[str] = None,
parameters: typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]] = OMIT,
title: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[WorkflowRunRequestInputProxyLocation] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
max_screenshot_scrolls: typing.Optional[int] = OMIT,
max_elapsed_time_minutes: typing.Optional[int] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
browser_address: typing.Optional[str] = OMIT,
ai_fallback: typing.Optional[bool] = OMIT,
run_with: typing.Optional[str] = OMIT,
run_metadata: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> WorkflowRunResponse:
"""
Run a workflow
Parameters
----------
workflow_id : str
ID of the workflow to run. Workflow ID starts with `wpid_`.
template : typing.Optional[bool]
max_steps_override : typing.Optional[int]
user_agent : typing.Optional[str]
parameters : typing.Optional[typing.Dict[str, typing.Optional[typing.Any]]]
Parameters to pass to the workflow
title : typing.Optional[str]
The title for this workflow run
proxy_location : typing.Optional[WorkflowRunRequestInputProxyLocation]
Geographic Proxy location to route the browser traffic through. This is only available in Skyvern Cloud.
Available geotargeting options:
- RESIDENTIAL: the default value. Skyvern Cloud uses a random US residential proxy.
- RESIDENTIAL_ES: Spain
- RESIDENTIAL_IE: Ireland
- RESIDENTIAL_GB: United Kingdom
- RESIDENTIAL_IN: India
- RESIDENTIAL_JP: Japan
- RESIDENTIAL_FR: France
- RESIDENTIAL_DE: Germany
- RESIDENTIAL_NZ: New Zealand
- RESIDENTIAL_PH: Philippines
- RESIDENTIAL_KR: South Korea
- RESIDENTIAL_SA: Saudi Arabia
- RESIDENTIAL_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_BR: Brazil
- RESIDENTIAL_TR: Turkey
- RESIDENTIAL_CA: Canada
- RESIDENTIAL_MX: Mexico
- RESIDENTIAL_IT: Italy
- RESIDENTIAL_NL: Netherlands
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California (deprecated, routes through RESIDENTIAL_ISP)
- US-NY: New York (deprecated, routes through RESIDENTIAL_ISP)
- US-TX: Texas (deprecated, routes through RESIDENTIAL_ISP)
- US-FL: Florida (deprecated, routes through RESIDENTIAL_ISP)
- US-WA: Washington (deprecated, routes through RESIDENTIAL_ISP)
- NONE: No proxy
For self-hosted deployments, you can pass a custom proxy URL as a dict: {"url": "http://user:password@proxy.example.com:8080"}. This routes the browser through your own proxy server and takes precedence over any globally configured proxy pool.
Can also be a GeoTarget object for granular city/state targeting: {"country": "US", "subdivision": "CA", "city": "San Francisco"}
webhook_url : typing.Optional[str]
URL to send workflow status updates to after a run is finished. Refer to https://www.skyvern.com/docs/running-tasks/webhooks-faq for webhook questions.
totp_url : typing.Optional[str]
URL that serves TOTP/2FA/MFA codes for Skyvern to use during the workflow run. Refer to https://www.skyvern.com/docs/credentials/totp#option-2-get-code-from-your-endpoint for more details.
totp_identifier : typing.Optional[str]
Identifier for the TOTP/2FA/MFA code when the code is pushed to Skyvern. Refer to https://www.skyvern.com/docs/credentials/totp#option-3-push-code-to-skyvern for more details.
browser_session_id : typing.Optional[str]
ID of a Skyvern browser session to reuse, having it continue from the current screen state
browser_profile_id : typing.Optional[str]
ID of a browser profile to reuse for this workflow run
max_screenshot_scrolls : typing.Optional[int]
The maximum number of scrolls for the post action screenshot. When it's None or 0, it takes the current viewpoint screenshot.
max_elapsed_time_minutes : typing.Optional[int]
Timeout this workflow run after the configured elapsed runtime in minutes. When omitted, the platform default is 240 minutes. The maximum configurable value is 480 minutes.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
The extra HTTP headers for the requests in browser.
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Use this for browser-provider auth (e.g., x-api-key for Skyvern Cloud, Browserless, or similar). These headers are NEVER forwarded to target websites.
browser_address : typing.Optional[str]
The CDP address for the workflow run.
ai_fallback : typing.Optional[bool]
Whether to fallback to AI if the workflow run fails.
run_with : typing.Optional[str]
Whether to run the workflow with agent or code. Null inherits from the workflow setting.
run_metadata : typing.Optional[typing.Dict[str, typing.Optional[str]]]
String key/value metadata to attach to this workflow run for analytics tag filtering.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
WorkflowRunResponse
Successfully run workflow
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.run_workflow(
max_steps_override=1,
user_agent="x-user-agent",
template=True,
workflow_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.run_workflow(
workflow_id=workflow_id,
template=template,
max_steps_override=max_steps_override,
user_agent=user_agent,
parameters=parameters,
title=title,
proxy_location=proxy_location,
webhook_url=webhook_url,
totp_url=totp_url,
totp_identifier=totp_identifier,
browser_session_id=browser_session_id,
browser_profile_id=browser_profile_id,
max_screenshot_scrolls=max_screenshot_scrolls,
max_elapsed_time_minutes=max_elapsed_time_minutes,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
browser_address=browser_address,
ai_fallback=ai_fallback,
run_with=run_with,
run_metadata=run_metadata,
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)
Parameters
----------
run_id : str
The id of the task run or the workflow run.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
GetRunResponse
Successfully got run
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_run(
run_id="tsk_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_run(run_id, request_options=request_options)
return _response.data
async def cancel_run(
self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Cancel a run (task or workflow)
Parameters
----------
run_id : str
The id of the task run or the workflow run to cancel.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Optional[typing.Any]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.cancel_run(
run_id="run_id",
)
asyncio.run(main())
"""
_response = await self._raw_client.cancel_run(run_id, request_options=request_options)
return _response.data
async def bulk_cancel_runs(
self, *, run_ids: typing.Sequence[str], request_options: typing.Optional[RequestOptions] = None
) -> BulkCancelRunsResponse:
"""
Cancel multiple runs (tasks or workflows) in a single request
Parameters
----------
run_ids : typing.Sequence[str]
List of run IDs to cancel
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BulkCancelRunsResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.bulk_cancel_runs(
run_ids=["run_ids"],
)
asyncio.run(main())
"""
_response = await self._raw_client.bulk_cancel_runs(run_ids=run_ids, request_options=request_options)
return _response.data
async def get_workflows(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
only_saved_tasks: typing.Optional[bool] = None,
only_workflows: typing.Optional[bool] = None,
only_templates: typing.Optional[bool] = None,
search_key: typing.Optional[str] = None,
title: typing.Optional[str] = None,
folder_id: typing.Optional[str] = None,
status: typing.Optional[typing.Union[WorkflowStatus, typing.Sequence[WorkflowStatus]]] = None,
template: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Workflow]:
"""
Get all workflows with the latest version for the organization.
Search semantics:
- If `search_key` is provided, its value is used as a unified search term for
`workflows.title`, `folders.title`, and workflow parameter metadata (key, description, and default_value for
`WorkflowParameterModel`).
- Falls back to deprecated `title` (title-only search) if `search_key` is not provided.
- Parameter metadata search excludes soft-deleted parameter rows across all parameter tables.
Parameters
----------
page : typing.Optional[int]
page_size : typing.Optional[int]
only_saved_tasks : typing.Optional[bool]
only_workflows : typing.Optional[bool]
only_templates : typing.Optional[bool]
search_key : typing.Optional[str]
Case-insensitive substring search across: workflow title, folder name, and parameter metadata (key, description, default_value). A workflow is returned if any of these fields match. Soft-deleted parameter definitions are excluded. Takes precedence over the deprecated `title` parameter.
title : typing.Optional[str]
Deprecated: use search_key instead. Falls back to title-only search if search_key is not provided.
folder_id : typing.Optional[str]
Filter workflows by folder ID
status : typing.Optional[typing.Union[WorkflowStatus, typing.Sequence[WorkflowStatus]]]
template : typing.Optional[bool]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Workflow]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_workflows(
page=1,
page_size=1,
only_saved_tasks=True,
only_workflows=True,
only_templates=True,
search_key="search_key",
title="title",
folder_id="folder_id",
template=True,
)
asyncio.run(main())
"""
_response = await self._raw_client.get_workflows(
page=page,
page_size=page_size,
only_saved_tasks=only_saved_tasks,
only_workflows=only_workflows,
only_templates=only_templates,
search_key=search_key,
title=title,
folder_id=folder_id,
status=status,
template=template,
request_options=request_options,
)
return _response.data
async def create_workflow(
self,
*,
folder_id: typing.Optional[str] = None,
json_definition: typing.Optional[WorkflowCreateYamlRequest] = OMIT,
yaml_definition: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Create a new workflow
Parameters
----------
folder_id : typing.Optional[str]
Optional folder ID to assign the workflow to
json_definition : typing.Optional[WorkflowCreateYamlRequest]
Workflow definition in JSON format
yaml_definition : typing.Optional[str]
Workflow definition in YAML format
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successfully created workflow
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.create_workflow(
folder_id="folder_id",
)
asyncio.run(main())
"""
_response = await self._raw_client.create_workflow(
folder_id=folder_id,
json_definition=json_definition,
yaml_definition=yaml_definition,
request_options=request_options,
)
return _response.data
async def update_workflow(
self,
workflow_id: str,
*,
json_definition: typing.Optional[WorkflowCreateYamlRequest] = OMIT,
yaml_definition: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Update a workflow
Parameters
----------
workflow_id : str
The ID of the workflow to update. Workflow ID starts with `wpid_`.
json_definition : typing.Optional[WorkflowCreateYamlRequest]
Workflow definition in JSON format
yaml_definition : typing.Optional[str]
Workflow definition in YAML format
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successfully updated workflow
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.update_workflow(
workflow_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.update_workflow(
workflow_id,
json_definition=json_definition,
yaml_definition=yaml_definition,
request_options=request_options,
)
return _response.data
async def delete_workflow(
self, workflow_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Delete a workflow
Parameters
----------
workflow_id : str
The ID of the workflow to delete. Workflow ID starts with `wpid_`.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Optional[typing.Any]
Successfully deleted workflow
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.delete_workflow(
workflow_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.delete_workflow(workflow_id, request_options=request_options)
return _response.data
async def get_folders(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
search: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Folder]:
"""
Get all folders for the organization
Parameters
----------
page : typing.Optional[int]
Page number
page_size : typing.Optional[int]
Number of folders per page
search : typing.Optional[str]
Search folders by title or description
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Folder]
Successfully retrieved folders
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_folders(
page=1,
page_size=1,
search="search",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_folders(
page=page, page_size=page_size, search=search, request_options=request_options
)
return _response.data
async def create_folder(
self,
*,
title: str,
description: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Folder:
"""
Create a new folder to organize workflows
Parameters
----------
title : str
Folder title
description : typing.Optional[str]
Folder description
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Folder
Successfully created folder
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.create_folder(
title="title",
)
asyncio.run(main())
"""
_response = await self._raw_client.create_folder(
title=title, description=description, request_options=request_options
)
return _response.data
async def get_folder(self, folder_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Folder:
"""
Get a specific folder by ID
Parameters
----------
folder_id : str
Folder ID
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Folder
Successfully retrieved folder
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_folder(
folder_id="fld_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_folder(folder_id, request_options=request_options)
return _response.data
async def update_folder(
self,
folder_id: str,
*,
title: typing.Optional[str] = OMIT,
description: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Folder:
"""
Update a folder's title or description
Parameters
----------
folder_id : str
Folder ID
title : typing.Optional[str]
Folder title
description : typing.Optional[str]
Folder description
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Folder
Successfully updated folder
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.update_folder(
folder_id="fld_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.update_folder(
folder_id, title=title, description=description, request_options=request_options
)
return _response.data
async def delete_folder(
self,
folder_id: str,
*,
delete_workflows: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.Dict[str, typing.Optional[typing.Any]]:
"""
Delete a folder. Optionally delete all workflows in the folder.
Parameters
----------
folder_id : str
Folder ID
delete_workflows : typing.Optional[bool]
If true, also delete all workflows in this folder
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Dict[str, typing.Optional[typing.Any]]
Successfully deleted folder
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.delete_folder(
folder_id="fld_123",
delete_workflows=True,
)
asyncio.run(main())
"""
_response = await self._raw_client.delete_folder(
folder_id, delete_workflows=delete_workflows, request_options=request_options
)
return _response.data
async def update_workflow_folder(
self,
workflow_permanent_id: str,
*,
folder_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Update a workflow's folder assignment for the latest version
Parameters
----------
workflow_permanent_id : str
Workflow permanent ID
folder_id : typing.Optional[str]
Folder ID to assign workflow to. Set to null to remove from folder.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successfully updated workflow folder
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.update_workflow_folder(
workflow_permanent_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.update_workflow_folder(
workflow_permanent_id, folder_id=folder_id, request_options=request_options
)
return _response.data
async def get_run_artifacts(
self,
run_id: str,
*,
artifact_type: typing.Optional[typing.Union[ArtifactType, typing.Sequence[ArtifactType]]] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Artifact]:
"""
Get artifacts for a run
Parameters
----------
run_id : str
The id of the task run or the workflow run.
artifact_type : typing.Optional[typing.Union[ArtifactType, typing.Sequence[ArtifactType]]]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Artifact]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_run_artifacts(
run_id="run_id",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_run_artifacts(
run_id, artifact_type=artifact_type, request_options=request_options
)
return _response.data
async def retry_run_webhook(
self,
run_id: str,
*,
request: typing.Optional[RetryRunWebhookRequest] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> RunWebhookReplayResponse:
"""
Retry sending the webhook for a run
Parameters
----------
run_id : str
The id of the task run or the workflow run.
request : typing.Optional[RetryRunWebhookRequest]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
RunWebhookReplayResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern, RetryRunWebhookRequest
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.retry_run_webhook(
run_id="tsk_123",
request=RetryRunWebhookRequest(),
)
asyncio.run(main())
"""
_response = await self._raw_client.retry_run_webhook(run_id, request=request, request_options=request_options)
return _response.data
async def get_run_timeline(
self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[WorkflowRunTimeline]:
"""
Get timeline for a run (workflow run or task_v2 run)
Parameters
----------
run_id : str
The id of the workflow run or task_v2 run.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[WorkflowRunTimeline]
Successfully retrieved run timeline
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_run_timeline(
run_id="wr_123",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_run_timeline(run_id, request_options=request_options)
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",
max_steps_override=1,
user_agent="x-user-agent",
)
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_runs_v2(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
status: typing.Optional[typing.Union[RunStatus, typing.Sequence[RunStatus]]] = None,
search_key: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[TaskRunListItem]:
"""
Parameters
----------
page : typing.Optional[int]
page_size : typing.Optional[int]
status : typing.Optional[typing.Union[RunStatus, typing.Sequence[RunStatus]]]
search_key : typing.Optional[str]
Case-insensitive substring search (min 3 chars for trigram index).
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[TaskRunListItem]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_runs_v2(
page=1,
page_size=1,
search_key="search_key",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_runs_v2(
page=page, page_size=page_size, status=status, search_key=search_key, request_options=request_options
)
return _response.data
async def get_workflow_runs(
self,
*,
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 workflow runs across all workflows for the current organization.
Results are paginated and can be filtered by **status**, **search_key**, and **error_code**. All filters are combined with **AND** logic — a run must match every supplied filter to be returned.
### search_key
A case-insensitive substring search that matches against **any** of the following fields:
| Searched field | Description |
|---|---|
| `workflow_run_id` | The unique run identifier (e.g. `wr_123…`) |
| Parameter **key** | The `key` of any workflow parameter definition associated with the run |
| Parameter **description** | The `description` of any workflow parameter definition |
| Run parameter **value** | The actual value supplied for any parameter when the run was created |
| `extra_http_headers` | Extra HTTP headers attached to the run (searched as raw JSON text) |
Soft-deleted parameter definitions are excluded from key/description matching. A run is returned if **any** of the fields above contain the search term.
### error_code
An **exact-match** filter against the `error_code` field inside each task's `errors` JSON array. A run matches if **any** of its tasks contains an error object with a matching `error_code` value. Error codes are user-defined strings set during workflow execution (e.g. `INVALID_CREDENTIALS`, `LOGIN_FAILED`, `CAPTCHA_DETECTED`).
### Combining filters
All query parameters use AND logic:
- `?status=failed` — only failed runs
- `?status=failed&error_code=LOGIN_FAILED` — failed runs **and** have a LOGIN_FAILED error
- `?status=failed&error_code=LOGIN_FAILED&search_key=prod_credential` — all three conditions must match
Parameters
----------
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. A run is returned if any of these fields match. Soft-deleted parameter definitions are excluded from key/description matching.
error_code : typing.Optional[str]
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.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[WorkflowRun]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_workflow_runs(
page=1,
page_size=1,
search_key="search_key",
error_code="error_code",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_workflow_runs(
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_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.
Supports filtering by **status**, **search_key**, and **error_code**. All filters are combined with **AND** logic.
### search_key
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.
### error_code
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`.
Parameters
----------
workflow_id : str
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. A run is returned if any of these fields match. Soft-deleted parameter definitions are excluded from key/description matching.
error_code : typing.Optional[str]
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.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[WorkflowRun]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_workflow_runs_by_id(
workflow_id="workflow_id",
page=1,
page_size=1,
search_key="search_key",
error_code="error_code",
)
asyncio.run(main())
"""
_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,
*,
version: typing.Optional[int] = None,
template: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Parameters
----------
workflow_permanent_id : str
version : typing.Optional[int]
template : typing.Optional[bool]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Workflow
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_workflow(
workflow_permanent_id="workflow_permanent_id",
version=1,
template=True,
)
asyncio.run(main())
"""
_response = await self._raw_client.get_workflow(
workflow_permanent_id, version=version, template=template, request_options=request_options
)
return _response.data
async def get_workflow_versions(
self,
workflow_permanent_id: str,
*,
template: typing.Optional[bool] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Workflow]:
"""
Get all versions of a workflow by its permanent ID.
Parameters
----------
workflow_permanent_id : str
template : typing.Optional[bool]
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Workflow]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_workflow_versions(
workflow_permanent_id="workflow_permanent_id",
template=True,
)
asyncio.run(main())
"""
_response = await self._raw_client.get_workflow_versions(
workflow_permanent_id, template=template, request_options=request_options
)
return _response.data
async def upload_file(
self, *, file: str, request_options: typing.Optional[RequestOptions] = None
) -> UploadFileResponse:
"""
Parameters
----------
file : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
UploadFileResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.upload_file(
file="file",
)
asyncio.run(main())
"""
_response = await self._raw_client.upload_file(file=file, request_options=request_options)
return _response.data
async def list_browser_profiles(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
include_deleted: typing.Optional[bool] = None,
search_key: typing.Optional[str] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[BrowserProfile]:
"""
Get all browser profiles for the organization
Parameters
----------
page : typing.Optional[int]
page_size : typing.Optional[int]
include_deleted : typing.Optional[bool]
Include deleted browser profiles
search_key : typing.Optional[str]
Case-insensitive substring search across: browser profile name and description. A profile is returned if either field matches.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[BrowserProfile]
Successfully retrieved browser profiles
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.list_browser_profiles(
page=1,
page_size=1,
include_deleted=True,
search_key="search_key",
)
asyncio.run(main())
"""
_response = await self._raw_client.list_browser_profiles(
page=page,
page_size=page_size,
include_deleted=include_deleted,
search_key=search_key,
request_options=request_options,
)
return _response.data
async def create_browser_profile(
self,
*,
name: str,
description: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
workflow_run_id: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> BrowserProfile:
"""
Create a blank browser profile, or create one from a persistent browser session or workflow run.
Parameters
----------
name : str
Name for the browser profile
description : typing.Optional[str]
Optional profile description
browser_session_id : typing.Optional[str]
Persistent browser session to convert into a profile. Omit for a blank profile.
workflow_run_id : typing.Optional[str]
Workflow run whose persisted session should be captured. Omit for a blank profile.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserProfile
Successfully created browser profile
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.create_browser_profile(
name="name",
)
asyncio.run(main())
"""
_response = await self._raw_client.create_browser_profile(
name=name,
description=description,
browser_session_id=browser_session_id,
workflow_run_id=workflow_run_id,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
request_options=request_options,
)
return _response.data
async def get_browser_profile(
self, profile_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> BrowserProfile:
"""
Get a specific browser profile by ID
Parameters
----------
profile_id : str
The ID of the browser profile. browser_profile_id starts with `bp_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserProfile
Successfully retrieved browser profile
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_browser_profile(
profile_id="bp_123456",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_browser_profile(profile_id, request_options=request_options)
return _response.data
async def delete_browser_profile(
self, profile_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> None:
"""
Delete a browser profile (soft delete)
Parameters
----------
profile_id : str
The ID of the browser profile to delete. browser_profile_id starts with `bp_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
None
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.delete_browser_profile(
profile_id="bp_123456",
)
asyncio.run(main())
"""
_response = await self._raw_client.delete_browser_profile(profile_id, request_options=request_options)
return _response.data
async def update_browser_profile(
self,
profile_id: str,
*,
name: typing.Optional[str] = OMIT,
description: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
rotate_proxy_session_id: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> BrowserProfile:
"""
Update a browser profile's name and/or description
Parameters
----------
profile_id : str
The ID of the browser profile to update. browser_profile_id starts with `bp_`
name : typing.Optional[str]
New name for the browser profile
description : typing.Optional[str]
New description for the browser profile
rotate_proxy_session_id : typing.Optional[bool]
Rotate the Skyvern-managed proxy sticky-session id for this browser profile.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserProfile
Successfully updated browser profile
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.update_browser_profile(
profile_id="bp_123456",
)
asyncio.run(main())
"""
_response = await self._raw_client.update_browser_profile(
profile_id,
name=name,
description=description,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
rotate_proxy_session_id=rotate_proxy_session_id,
request_options=request_options,
)
return _response.data
async def get_browser_sessions(
self, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.List[BrowserSessionResponse]:
"""
Get all active browser sessions for the organization
Parameters
----------
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[BrowserSessionResponse]
Successfully retrieved all active browser sessions
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_browser_sessions()
asyncio.run(main())
"""
_response = await self._raw_client.get_browser_sessions(request_options=request_options)
return _response.data
async def create_browser_session(
self,
*,
timeout: typing.Optional[int] = OMIT,
proxy_location: typing.Optional[CreateBrowserSessionRequestProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
extensions: typing.Optional[typing.Sequence[Extensions]] = OMIT,
browser_type: typing.Optional[PersistentBrowserType] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> BrowserSessionResponse:
"""
Create a browser session that persists across multiple runs
Parameters
----------
timeout : typing.Optional[int]
Timeout in minutes for the session. Timeout is applied after the session is started. Must be between 5 and 1440. Defaults to 60.
proxy_location : typing.Optional[CreateBrowserSessionRequestProxyLocation]
Geographic Proxy location to route the browser traffic through. This is only available in Skyvern Cloud.
Available geotargeting options:
- RESIDENTIAL: the default value. Skyvern Cloud uses a random US residential proxy.
- RESIDENTIAL_ES: Spain
- RESIDENTIAL_IE: Ireland
- RESIDENTIAL_GB: United Kingdom
- RESIDENTIAL_IN: India
- RESIDENTIAL_JP: Japan
- RESIDENTIAL_FR: France
- RESIDENTIAL_DE: Germany
- RESIDENTIAL_NZ: New Zealand
- RESIDENTIAL_PH: Philippines
- RESIDENTIAL_KR: South Korea
- RESIDENTIAL_SA: Saudi Arabia
- RESIDENTIAL_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_BR: Brazil
- RESIDENTIAL_TR: Turkey
- RESIDENTIAL_CA: Canada
- RESIDENTIAL_MX: Mexico
- RESIDENTIAL_IT: Italy
- RESIDENTIAL_NL: Netherlands
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California (deprecated, routes through RESIDENTIAL_ISP)
- US-NY: New York (deprecated, routes through RESIDENTIAL_ISP)
- US-TX: Texas (deprecated, routes through RESIDENTIAL_ISP)
- US-FL: Florida (deprecated, routes through RESIDENTIAL_ISP)
- US-WA: Washington (deprecated, routes through RESIDENTIAL_ISP)
- NONE: No proxy
For self-hosted deployments, you can pass a custom proxy URL as a dict: {"url": "http://user:password@proxy.example.com:8080"}. This routes the browser through your own proxy server and takes precedence over any globally configured proxy pool.
Can also be a GeoTarget object for granular city/state targeting: {"country": "US", "subdivision": "CA", "city": "San Francisco"}, or a custom proxy URL dict for self-hosted deployments: {"url": "http://user:password@proxy.example.com:8080"}
proxy_session_id : typing.Optional[str]
Opaque Skyvern-managed proxy sticky-session id for pinned Residential ISP sessions.
extensions : typing.Optional[typing.Sequence[Extensions]]
A list of extensions to install in the browser session.
browser_type : typing.Optional[PersistentBrowserType]
The type of browser to use for the session.
browser_profile_id : typing.Optional[str]
ID of a browser profile to load into this session (restores cookies, localStorage, etc.). browser_profile_id starts with `bp_`.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserSessionResponse
Successfully created browser session
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.create_browser_session()
asyncio.run(main())
"""
_response = await self._raw_client.create_browser_session(
timeout=timeout,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
extensions=extensions,
browser_type=browser_type,
browser_profile_id=browser_profile_id,
request_options=request_options,
)
return _response.data
async def close_browser_session(
self, browser_session_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Close a session. Once closed, the session cannot be used again.
Parameters
----------
browser_session_id : str
The ID of the browser session to close. completed_at will be set when the browser session is closed. browser_session_id starts with `pbs_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.Optional[typing.Any]
Successfully closed browser session
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.close_browser_session(
browser_session_id="pbs_123456",
)
asyncio.run(main())
"""
_response = await self._raw_client.close_browser_session(browser_session_id, request_options=request_options)
return _response.data
async def get_browser_session(
self, browser_session_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> BrowserSessionResponse:
"""
Get details about a specific browser session, including the browser address for cdp connection.
Parameters
----------
browser_session_id : str
The ID of the browser session. browser_session_id starts with `pbs_`
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
BrowserSessionResponse
Successfully retrieved browser session details
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_browser_session(
browser_session_id="pbs_123456",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_browser_session(browser_session_id, request_options=request_options)
return _response.data
async def send_totp_code(
self,
*,
totp_identifier: str,
content: str,
task_id: typing.Optional[str] = OMIT,
workflow_id: typing.Optional[str] = OMIT,
workflow_run_id: typing.Optional[str] = OMIT,
source: typing.Optional[str] = OMIT,
expired_at: typing.Optional[dt.datetime] = OMIT,
type: typing.Optional[OtpType] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> TotpCode:
"""
Forward a TOTP (2FA, MFA) email or sms message containing the code to Skyvern. This endpoint stores the code in database so that Skyvern can use it while running tasks/workflows.
Parameters
----------
totp_identifier : str
The identifier of the TOTP code. It can be the email address, phone number, or the identifier of the user.
content : str
The content of the TOTP code. It can be the email content that contains the TOTP code, or the sms message that contains the TOTP code. Skyvern will automatically extract the TOTP code from the content.
task_id : typing.Optional[str]
The task_id the totp code is for. It can be the task_id of the task that the TOTP code is for.
workflow_id : typing.Optional[str]
The workflow ID the TOTP code is for. It can be the workflow ID of the workflow that the TOTP code is for.
workflow_run_id : typing.Optional[str]
The workflow run id that the TOTP code is for. It can be the workflow run id of the workflow run that the TOTP code is for.
source : typing.Optional[str]
An optional field. The source of the TOTP code. e.g. email, sms, etc.
expired_at : typing.Optional[dt.datetime]
The timestamp when the TOTP code expires
type : typing.Optional[OtpType]
Optional. If provided, forces extraction of this specific OTP type (totp or magic_link). Use this when the content contains multiple OTP types and you want to specify which one to extract.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
TotpCode
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.send_totp_code(
totp_identifier="john.doe@example.com",
content="Hello, your verification code is 123456",
)
asyncio.run(main())
"""
_response = await self._raw_client.send_totp_code(
totp_identifier=totp_identifier,
content=content,
task_id=task_id,
workflow_id=workflow_id,
workflow_run_id=workflow_run_id,
source=source,
expired_at=expired_at,
type=type,
request_options=request_options,
)
return _response.data
async def get_credentials(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
vault_type: typing.Optional[CredentialVaultType] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[CredentialResponse]:
"""
Retrieves a paginated list of credentials for the current organization
Parameters
----------
page : typing.Optional[int]
Page number for pagination
page_size : typing.Optional[int]
Number of items per page
vault_type : typing.Optional[CredentialVaultType]
Filter credentials by vault type (e.g. 'skyvern', 'custom', 'bitwarden', 'azure_vault')
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[CredentialResponse]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_credentials(
page=1,
page_size=10,
vault_type="bitwarden",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_credentials(
page=page, page_size=page_size, vault_type=vault_type, request_options=request_options
)
return _response.data
async def create_credential(
self,
*,
name: str,
credential_type: SkyvernForgeSdkSchemasCredentialsCredentialType,
credential: CreateCredentialRequestCredential,
vault_type: typing.Optional[CredentialVaultType] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
rotate_proxy_session_id: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> CredentialResponse:
"""
Creates a new credential for the current organization
Parameters
----------
name : str
Name of the credential
credential_type : SkyvernForgeSdkSchemasCredentialsCredentialType
Type of credential to create
credential : CreateCredentialRequestCredential
The credential data to store
vault_type : typing.Optional[CredentialVaultType]
Which vault to store this credential in. If omitted, uses the instance default. Use this to mix Skyvern-hosted and custom credentials within the same organization.
rotate_proxy_session_id : typing.Optional[bool]
Rotate the Skyvern-managed proxy sticky-session id for this credential.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CredentialResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern, NonEmptyPasswordCredential
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.create_credential(
name="Amazon Login",
credential_type="password",
credential=NonEmptyPasswordCredential(
password="securepassword123",
username="user@example.com",
),
)
asyncio.run(main())
"""
_response = await self._raw_client.create_credential(
name=name,
credential_type=credential_type,
credential=credential,
vault_type=vault_type,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
rotate_proxy_session_id=rotate_proxy_session_id,
request_options=request_options,
)
return _response.data
async def update_credential(
self,
credential_id: str,
*,
name: str,
credential_type: SkyvernForgeSdkSchemasCredentialsCredentialType,
credential: CreateCredentialRequestCredential,
vault_type: typing.Optional[CredentialVaultType] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
proxy_session_id: typing.Optional[str] = OMIT,
rotate_proxy_session_id: typing.Optional[bool] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> CredentialResponse:
"""
Overwrites the stored credential data (e.g. username/password) while keeping the same credential_id.
Parameters
----------
credential_id : str
The unique identifier of the credential to update
name : str
Name of the credential
credential_type : SkyvernForgeSdkSchemasCredentialsCredentialType
Type of credential to create
credential : CreateCredentialRequestCredential
The credential data to store
vault_type : typing.Optional[CredentialVaultType]
Which vault to store this credential in. If omitted, uses the instance default. Use this to mix Skyvern-hosted and custom credentials within the same organization.
rotate_proxy_session_id : typing.Optional[bool]
Rotate the Skyvern-managed proxy sticky-session id for this credential.
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CredentialResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern, NonEmptyPasswordCredential
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.update_credential(
credential_id="cred_1234567890",
name="Amazon Login",
credential_type="password",
credential=NonEmptyPasswordCredential(
password="securepassword123",
username="user@example.com",
),
)
asyncio.run(main())
"""
_response = await self._raw_client.update_credential(
credential_id,
name=name,
credential_type=credential_type,
credential=credential,
vault_type=vault_type,
proxy_location=proxy_location,
proxy_session_id=proxy_session_id,
rotate_proxy_session_id=rotate_proxy_session_id,
request_options=request_options,
)
return _response.data
async def delete_credential(
self, credential_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> None:
"""
Deletes a specific credential by its ID
Parameters
----------
credential_id : str
The unique identifier of the credential to delete
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
None
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.delete_credential(
credential_id="cred_1234567890",
)
asyncio.run(main())
"""
_response = await self._raw_client.delete_credential(credential_id, request_options=request_options)
return _response.data
async def get_credential(
self, credential_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> CredentialResponse:
"""
Retrieves a specific credential by its ID
Parameters
----------
credential_id : str
The unique identifier of the credential
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CredentialResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_credential(
credential_id="cred_1234567890",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_credential(credential_id, request_options=request_options)
return _response.data
async def login(
self,
*,
credential_type: SkyvernSchemasCredentialTypeCredentialType,
user_agent: typing.Optional[str] = None,
url: typing.Optional[str] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
browser_address: typing.Optional[str] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_screenshot_scrolling_times: typing.Optional[int] = OMIT,
prompt: typing.Optional[str] = OMIT,
credential_id: typing.Optional[str] = OMIT,
bitwarden_collection_id: typing.Optional[str] = OMIT,
bitwarden_item_id: typing.Optional[str] = OMIT,
onepassword_vault_id: typing.Optional[str] = OMIT,
onepassword_item_id: typing.Optional[str] = OMIT,
azure_vault_name: typing.Optional[str] = OMIT,
azure_vault_username_key: typing.Optional[str] = OMIT,
azure_vault_password_key: typing.Optional[str] = OMIT,
azure_vault_totp_secret_key: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> WorkflowRunResponse:
"""
Log in to a website using either credential stored in Skyvern, Bitwarden, 1Password, or Azure Vault
Parameters
----------
credential_type : SkyvernSchemasCredentialTypeCredentialType
Where to get the credential from
user_agent : typing.Optional[str]
url : typing.Optional[str]
Website URL
webhook_url : typing.Optional[str]
Webhook URL to send status updates
proxy_location : typing.Optional[ProxyLocation]
Proxy location to use
totp_identifier : typing.Optional[str]
Identifier for TOTP (Time-based One-Time Password) if required
totp_url : typing.Optional[str]
TOTP URL to fetch one-time passwords
browser_session_id : typing.Optional[str]
ID of the browser session to use, which is prefixed by `pbs_` e.g. `pbs_123456`
browser_profile_id : typing.Optional[str]
ID of a browser profile to reuse for this run
browser_address : typing.Optional[str]
The CDP address for the task.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
Additional HTTP headers to include in requests
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Never forwarded to target websites.
max_screenshot_scrolling_times : typing.Optional[int]
Maximum number of times to scroll for screenshots
prompt : typing.Optional[str]
Login instructions. Skyvern has default prompt/instruction for login if this field is not provided.
credential_id : typing.Optional[str]
ID of the Skyvern credential to use for login.
bitwarden_collection_id : typing.Optional[str]
Bitwarden collection ID. You can find it in the Bitwarden collection URL. e.g. `https://vault.bitwarden.com/vaults/collection_id/items`
bitwarden_item_id : typing.Optional[str]
Bitwarden item ID
onepassword_vault_id : typing.Optional[str]
1Password vault ID
onepassword_item_id : typing.Optional[str]
1Password item ID
azure_vault_name : typing.Optional[str]
Azure Vault Name
azure_vault_username_key : typing.Optional[str]
Azure Vault username key
azure_vault_password_key : typing.Optional[str]
Azure Vault password key
azure_vault_totp_secret_key : typing.Optional[str]
Azure Vault TOTP secret key
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
WorkflowRunResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.login(
user_agent="x-user-agent",
credential_type="skyvern",
)
asyncio.run(main())
"""
_response = await self._raw_client.login(
credential_type=credential_type,
user_agent=user_agent,
url=url,
webhook_url=webhook_url,
proxy_location=proxy_location,
totp_identifier=totp_identifier,
totp_url=totp_url,
browser_session_id=browser_session_id,
browser_profile_id=browser_profile_id,
browser_address=browser_address,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
max_screenshot_scrolling_times=max_screenshot_scrolling_times,
prompt=prompt,
credential_id=credential_id,
bitwarden_collection_id=bitwarden_collection_id,
bitwarden_item_id=bitwarden_item_id,
onepassword_vault_id=onepassword_vault_id,
onepassword_item_id=onepassword_item_id,
azure_vault_name=azure_vault_name,
azure_vault_username_key=azure_vault_username_key,
azure_vault_password_key=azure_vault_password_key,
azure_vault_totp_secret_key=azure_vault_totp_secret_key,
request_options=request_options,
)
return _response.data
async def download_files(
self,
*,
navigation_goal: str,
user_agent: typing.Optional[str] = None,
url: typing.Optional[str] = OMIT,
webhook_url: typing.Optional[str] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = OMIT,
totp_identifier: typing.Optional[str] = OMIT,
totp_url: typing.Optional[str] = OMIT,
browser_session_id: typing.Optional[str] = OMIT,
browser_profile_id: typing.Optional[str] = OMIT,
browser_address: typing.Optional[str] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
cdp_connect_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_screenshot_scrolling_times: typing.Optional[int] = OMIT,
download_suffix: typing.Optional[str] = OMIT,
download_timeout: typing.Optional[float] = OMIT,
max_steps_per_run: typing.Optional[int] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> WorkflowRunResponse:
"""
Download a file from a website by navigating and clicking download buttons
Parameters
----------
navigation_goal : str
Instructions for navigating to and downloading the file
user_agent : typing.Optional[str]
url : typing.Optional[str]
Website URL
webhook_url : typing.Optional[str]
Webhook URL to send status updates
proxy_location : typing.Optional[ProxyLocation]
Proxy location to use
totp_identifier : typing.Optional[str]
Identifier for TOTP (Time-based One-Time Password) if required
totp_url : typing.Optional[str]
TOTP URL to fetch one-time passwords
browser_session_id : typing.Optional[str]
ID of the browser session to use, which is prefixed by `pbs_` e.g. `pbs_123456`
browser_profile_id : typing.Optional[str]
ID of a browser profile to reuse for this run
browser_address : typing.Optional[str]
The CDP address for the task.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
Additional HTTP headers to include in requests
cdp_connect_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
HTTP headers attached ONLY to the CDP WebSocket handshake when connecting to a remote browser via browser_address. Never forwarded to target websites.
max_screenshot_scrolling_times : typing.Optional[int]
Maximum number of times to scroll for screenshots
download_suffix : typing.Optional[str]
Suffix or complete filename for the downloaded file
download_timeout : typing.Optional[float]
Timeout in seconds for the download operation
max_steps_per_run : typing.Optional[int]
Maximum number of steps to execute
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
WorkflowRunResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.download_files(
user_agent="x-user-agent",
navigation_goal="navigation_goal",
)
asyncio.run(main())
"""
_response = await self._raw_client.download_files(
navigation_goal=navigation_goal,
user_agent=user_agent,
url=url,
webhook_url=webhook_url,
proxy_location=proxy_location,
totp_identifier=totp_identifier,
totp_url=totp_url,
browser_session_id=browser_session_id,
browser_profile_id=browser_profile_id,
browser_address=browser_address,
extra_http_headers=extra_http_headers,
cdp_connect_headers=cdp_connect_headers,
max_screenshot_scrolling_times=max_screenshot_scrolling_times,
download_suffix=download_suffix,
download_timeout=download_timeout,
max_steps_per_run=max_steps_per_run,
request_options=request_options,
)
return _response.data
async def get_scripts(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = None,
request_options: typing.Optional[RequestOptions] = None,
) -> typing.List[Script]:
"""
Retrieves a paginated list of scripts for the current organization
Parameters
----------
page : typing.Optional[int]
Page number for pagination
page_size : typing.Optional[int]
Number of items per page
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
typing.List[Script]
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_scripts(
page=1,
page_size=10,
)
asyncio.run(main())
"""
_response = await self._raw_client.get_scripts(page=page, page_size=page_size, request_options=request_options)
return _response.data
async def create_script(
self,
*,
workflow_id: typing.Optional[str] = OMIT,
run_id: typing.Optional[str] = OMIT,
files: typing.Optional[typing.Sequence[ScriptFileCreate]] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> CreateScriptResponse:
"""
Create a new script with optional files and metadata
Parameters
----------
workflow_id : typing.Optional[str]
Associated workflow ID
run_id : typing.Optional[str]
Associated run ID
files : typing.Optional[typing.Sequence[ScriptFileCreate]]
Array of files to include in the script
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CreateScriptResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.create_script()
asyncio.run(main())
"""
_response = await self._raw_client.create_script(
workflow_id=workflow_id, run_id=run_id, files=files, request_options=request_options
)
return _response.data
async def get_script(self, script_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Script:
"""
Retrieves a specific script by its ID
Parameters
----------
script_id : str
The unique identifier of the script
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Script
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.get_script(
script_id="s_abc123",
)
asyncio.run(main())
"""
_response = await self._raw_client.get_script(script_id, request_options=request_options)
return _response.data
async def deploy_script(
self,
script_id: str,
*,
files: typing.Sequence[ScriptFileCreate],
request_options: typing.Optional[RequestOptions] = None,
) -> CreateScriptResponse:
"""
Deploy a script with updated files, creating a new version
Parameters
----------
script_id : str
The unique identifier of the script
files : typing.Sequence[ScriptFileCreate]
Array of files to include in the script
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
CreateScriptResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern, ScriptFileCreate
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.deploy_script(
script_id="s_abc123",
files=[
ScriptFileCreate(
path="src/main.py",
content="content",
)
],
)
asyncio.run(main())
"""
_response = await self._raw_client.deploy_script(script_id, files=files, request_options=request_options)
return _response.data
async def run_sdk_action(
self,
*,
url: str,
action: RunSdkActionRequestAction,
browser_session_id: typing.Optional[str] = OMIT,
browser_address: typing.Optional[str] = OMIT,
workflow_run_id: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> RunSdkActionResponse:
"""
Execute a single SDK action with the specified parameters
Parameters
----------
url : str
The URL where the action should be executed
action : RunSdkActionRequestAction
The action to execute with its specific parameters
browser_session_id : typing.Optional[str]
The browser session ID
browser_address : typing.Optional[str]
The browser address
workflow_run_id : typing.Optional[str]
Optional workflow run ID to continue an existing workflow run
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
RunSdkActionResponse
Successful Response
Examples
--------
import asyncio
from skyvern import AsyncSkyvern, RunSdkActionRequestAction_AiAct
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
)
async def main() -> None:
await client.run_sdk_action(
url="url",
action=RunSdkActionRequestAction_AiAct(),
)
asyncio.run(main())
"""
_response = await self._raw_client.run_sdk_action(
url=url,
action=action,
browser_session_id=browser_session_id,
browser_address=browser_address,
workflow_run_id=workflow_run_id,
request_options=request_options,
)
return _response.data
@property
def artifacts(self):
if self._artifacts is None:
from .artifacts.client import AsyncArtifactsClient # noqa: E402
self._artifacts = AsyncArtifactsClient(client_wrapper=self._client_wrapper)
return self._artifacts
@property
def server(self):
if self._server is None:
from .server.client import AsyncServerClient # noqa: E402
self._server = AsyncServerClient(client_wrapper=self._client_wrapper)
return self._server
@property
def workflows(self):
if self._workflows is None:
from .workflows.client import AsyncWorkflowsClient # noqa: E402
self._workflows = AsyncWorkflowsClient(client_wrapper=self._client_wrapper)
return self._workflows
@property
def scripts(self):
if self._scripts is None:
from .scripts.client import AsyncScriptsClient # noqa: E402
self._scripts = AsyncScriptsClient(client_wrapper=self._client_wrapper)
return self._scripts
@property
def schedules(self):
if self._schedules is None:
from .schedules.client import AsyncSchedulesClient # noqa: E402
self._schedules = AsyncSchedulesClient(client_wrapper=self._client_wrapper)
return self._schedules
def _get_base_url(*, base_url: typing.Optional[str] = None, environment: SkyvernEnvironment) -> str:
if base_url is not None:
return base_url
elif environment is not None:
return environment.value
else:
raise Exception("Please pass in either base_url or environment to construct the client")
Reference in a new issue View git blame Copy permalink