Files
Dorod-Sky/skyvern/client/client.py
2025-10-21 21:25:45 +00:00

4873 lines
169 KiB
Python

# This file was auto-generated by Fern from our API Definition.
import typing
from .environment import SkyvernEnvironment
import httpx
from .core.client_wrapper import SyncClientWrapper
from .scripts.client import ScriptsClient
from .types.run_engine import RunEngine
from .types.proxy_location import ProxyLocation
from .types.task_run_request_data_extraction_schema import TaskRunRequestDataExtractionSchema
from .core.request_options import RequestOptions
from .types.task_run_response import TaskRunResponse
from .core.serialization import convert_and_respect_annotation_metadata
from .core.pydantic_utilities import parse_obj_as
from .errors.bad_request_error import BadRequestError
from .errors.unprocessable_entity_error import UnprocessableEntityError
from json.decoder import JSONDecodeError
from .core.api_error import ApiError
from .types.workflow_run_response import WorkflowRunResponse
from .types.get_run_response import GetRunResponse
from .core.jsonable_encoder import jsonable_encoder
from .errors.not_found_error import NotFoundError
from .types.workflow import Workflow
from .types.workflow_create_yaml_request import WorkflowCreateYamlRequest
from .types.artifact import Artifact
from .types.artifact_type import ArtifactType
from .types.workflow_run_timeline import WorkflowRunTimeline
from .types.browser_session_response import BrowserSessionResponse
from .errors.forbidden_error import ForbiddenError
import datetime as dt
from .types.totp_code import TotpCode
from .types.credential_response import CredentialResponse
from .types.skyvern_forge_sdk_schemas_credentials_credential_type import SkyvernForgeSdkSchemasCredentialsCredentialType
from .types.create_credential_request_credential import CreateCredentialRequestCredential
from .types.skyvern_schemas_run_blocks_credential_type import SkyvernSchemasRunBlocksCredentialType
from .types.script import Script
from .types.script_file_create import ScriptFileCreate
from .types.create_script_response import CreateScriptResponse
from .core.client_wrapper import AsyncClientWrapper
from .scripts.client import AsyncScriptsClient
# 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.PRODUCTION
api_key : typing.Optional[str]
x_api_key : str
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",
x_api_key="YOUR_X_API_KEY",
)
"""
def __init__(
self,
*,
base_url: typing.Optional[str] = None,
environment: SkyvernEnvironment = SkyvernEnvironment.PRODUCTION,
api_key: typing.Optional[str] = None,
x_api_key: str,
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 None
self._client_wrapper = SyncClientWrapper(
base_url=_get_base_url(base_url=base_url, environment=environment),
api_key=api_key,
x_api_key=x_api_key,
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.scripts = ScriptsClient(client_wrapper=self._client_wrapper)
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[ProxyLocation] = OMIT,
data_extraction_schema: typing.Optional[TaskRunRequestDataExtractionSchema] = 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,
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,
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-2.0`, the latest Skyvern agent that performs pretty well with complex and multi-step tasks. `skyvern-1.0` is good for simple tasks like filling a form, or searching for information on Google. 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[ProxyLocation]
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_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California
- US-NY: New York
- US-TX: Texas
- US-FL: Florida
- US-WA: Washington
- NONE: No proxy
data_extraction_schema : typing.Optional[TaskRunRequestDataExtractionSchema]
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.
publish_workflow : typing.Optional[bool]
Whether to publish this task as a reusable workflow. Only available for skyvern-2.0.
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.
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",
x_api_key="YOUR_X_API_KEY",
)
client.run_task(
prompt="Find the top 3 posts on Hacker News.",
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/run/tasks",
method="POST",
json={
"prompt": prompt,
"url": url,
"engine": engine,
"title": title,
"proxy_location": proxy_location,
"data_extraction_schema": convert_and_respect_annotation_metadata(
object_=data_extraction_schema, annotation=TaskRunRequestDataExtractionSchema, direction="write"
),
"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,
"publish_workflow": publish_workflow,
"include_action_history_in_verification": include_action_history_in_verification,
"max_screenshot_scrolls": max_screenshot_scrolls,
"browser_address": browser_address,
},
headers={
"x-user-agent": str(user_agent) if user_agent is not None else None,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
TaskRunResponse,
parse_obj_as(
type_=TaskRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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[ProxyLocation] = 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,
max_screenshot_scrolls: typing.Optional[int] = OMIT,
extra_http_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,
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[ProxyLocation]
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_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California
- US-NY: New York
- US-TX: Texas
- US-FL: Florida
- US-WA: Washington
- NONE: No proxy
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
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.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
The extra HTTP headers for the requests in browser.
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.
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",
x_api_key="YOUR_X_API_KEY",
)
client.run_workflow(
workflow_id="wpid_123",
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/run/workflows",
method="POST",
params={
"template": template,
},
json={
"workflow_id": workflow_id,
"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,
"max_screenshot_scrolls": max_screenshot_scrolls,
"extra_http_headers": extra_http_headers,
"browser_address": browser_address,
"ai_fallback": ai_fallback,
"run_with": run_with,
},
headers={
"x-max-steps-override": str(max_steps_override) if max_steps_override is not None else None,
"x-user-agent": str(user_agent) if user_agent is not None else None,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
WorkflowRunResponse,
parse_obj_as(
type_=WorkflowRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_run(
run_id="tsk_123",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
GetRunResponse,
parse_obj_as(
type_=GetRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.cancel_run(
run_id="run_id",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/cancel",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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,
search_key: typing.Optional[str] = None,
title: typing.Optional[str] = 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 both
`workflows.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]
search_key : typing.Optional[str]
Unified search across workflow title and parameter metadata (key, description, default_value).
title : typing.Optional[str]
Deprecated: use search_key instead.
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_workflows()
"""
_response = self._client_wrapper.httpx_client.request(
"v1/workflows",
method="GET",
params={
"page": page,
"page_size": page_size,
"only_saved_tasks": only_saved_tasks,
"only_workflows": only_workflows,
"search_key": search_key,
"title": title,
"template": template,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[Workflow],
parse_obj_as(
type_=typing.List[Workflow], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def create_workflow(
self,
*,
json_definition: typing.Optional[WorkflowCreateYamlRequest] = OMIT,
yaml_definition: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Create a new workflow
Parameters
----------
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",
x_api_key="YOUR_X_API_KEY",
)
client.create_workflow()
"""
_response = self._client_wrapper.httpx_client.request(
"v1/workflows",
method="POST",
json={
"json_definition": convert_and_respect_annotation_metadata(
object_=json_definition, annotation=WorkflowCreateYamlRequest, direction="write"
),
"yaml_definition": yaml_definition,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Workflow,
parse_obj_as(
type_=Workflow, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.update_workflow(
workflow_id="wpid_123",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/workflows/{jsonable_encoder(workflow_id)}",
method="POST",
json={
"json_definition": convert_and_respect_annotation_metadata(
object_=json_definition, annotation=WorkflowCreateYamlRequest, direction="write"
),
"yaml_definition": yaml_definition,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Workflow,
parse_obj_as(
type_=Workflow, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.delete_workflow(
workflow_id="wpid_123",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/workflows/{jsonable_encoder(workflow_id)}/delete",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def get_artifact(self, artifact_id: str, *, request_options: typing.Optional[RequestOptions] = None) -> Artifact:
"""
Get an artifact
Parameters
----------
artifact_id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Artifact
Successfully retrieved artifact
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
x_api_key="YOUR_X_API_KEY",
)
client.get_artifact(
artifact_id="artifact_id",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/artifacts/{jsonable_encoder(artifact_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Artifact,
parse_obj_as(
type_=Artifact, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_run_artifacts(
run_id="run_id",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/artifacts",
method="GET",
params={
"artifact_type": artifact_type,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[Artifact],
parse_obj_as(
type_=typing.List[Artifact], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def retry_run_webhook(
self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Retry sending the webhook for a run
Parameters
----------
run_id : str
The id of the task run or the workflow run.
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",
x_api_key="YOUR_X_API_KEY",
)
client.retry_run_webhook(
run_id="tsk_123",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/retry_webhook",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_run_timeline(
run_id="wr_123",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/timeline",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[WorkflowRunTimeline],
parse_obj_as(
type_=typing.List[WorkflowRunTimeline], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_browser_sessions()
"""
_response = self._client_wrapper.httpx_client.request(
"v1/browser_sessions",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[BrowserSessionResponse],
parse_obj_as(
type_=typing.List[BrowserSessionResponse], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def create_browser_session(
self,
*,
timeout: typing.Optional[int] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = 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[ProxyLocation]
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_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California
- US-NY: New York
- US-TX: Texas
- US-FL: Florida
- US-WA: Washington
- NONE: No proxy
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",
x_api_key="YOUR_X_API_KEY",
)
client.create_browser_session()
"""
_response = self._client_wrapper.httpx_client.request(
"v1/browser_sessions",
method="POST",
json={
"timeout": timeout,
"proxy_location": proxy_location,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
BrowserSessionResponse,
parse_obj_as(
type_=BrowserSessionResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.close_browser_session(
browser_session_id="pbs_123456",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/browser_sessions/{jsonable_encoder(browser_session_id)}/close",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_browser_session(
browser_session_id="pbs_123456",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/browser_sessions/{jsonable_encoder(browser_session_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
BrowserSessionResponse,
parse_obj_as(
type_=BrowserSessionResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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,
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
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
TotpCode
Successful Response
Examples
--------
from skyvern import Skyvern
client = Skyvern(
api_key="YOUR_API_KEY",
x_api_key="YOUR_X_API_KEY",
)
client.send_totp_code(
totp_identifier="john.doe@example.com",
content="Hello, your verification code is 123456",
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/credentials/totp",
method="POST",
json={
"totp_identifier": totp_identifier,
"task_id": task_id,
"workflow_id": workflow_id,
"workflow_run_id": workflow_run_id,
"source": source,
"content": content,
"expired_at": expired_at,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
TotpCode,
parse_obj_as(
type_=TotpCode, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def get_credentials(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = 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
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_credentials(
page=1,
page_size=10,
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/credentials",
method="GET",
params={
"page": page,
"page_size": page_size,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[CredentialResponse],
parse_obj_as(
type_=typing.List[CredentialResponse], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def create_credential(
self,
*,
name: str,
credential_type: SkyvernForgeSdkSchemasCredentialsCredentialType,
credential: CreateCredentialRequestCredential,
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
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",
x_api_key="YOUR_X_API_KEY",
)
client.create_credential(
name="My Credential",
credential_type="password",
credential=NonEmptyPasswordCredential(
password="securepassword123",
username="user@example.com",
totp="JBSWY3DPEHPK3PXP",
),
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/credentials",
method="POST",
json={
"name": name,
"credential_type": credential_type,
"credential": convert_and_respect_annotation_metadata(
object_=credential, annotation=CreateCredentialRequestCredential, direction="write"
),
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CredentialResponse,
parse_obj_as(
type_=CredentialResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.delete_credential(
credential_id="cred_1234567890",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/credentials/{jsonable_encoder(credential_id)}/delete",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_credential(
credential_id="cred_1234567890",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/credentials/{jsonable_encoder(credential_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CredentialResponse,
parse_obj_as(
type_=CredentialResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
def login(
self,
*,
credential_type: SkyvernSchemasRunBlocksCredentialType,
url: typing.Optional[str] = OMIT,
prompt: 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_address: typing.Optional[str] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_screenshot_scrolling_times: typing.Optional[int] = 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 : SkyvernSchemasRunBlocksCredentialType
Where to get the credential from
url : typing.Optional[str]
Website url
prompt : typing.Optional[str]
Login instructions. Skyvern has default prompt/instruction for login if this field is not provided.
webhook_url : typing.Optional[str]
Webhook URL to send login 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_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
max_screenshot_scrolling_times : typing.Optional[int]
Maximum number of times to scroll for screenshots
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",
x_api_key="YOUR_X_API_KEY",
)
client.login(
credential_type="skyvern",
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/run/tasks/login",
method="POST",
json={
"credential_type": credential_type,
"url": url,
"prompt": prompt,
"webhook_url": webhook_url,
"proxy_location": proxy_location,
"totp_identifier": totp_identifier,
"totp_url": totp_url,
"browser_session_id": browser_session_id,
"browser_address": browser_address,
"extra_http_headers": extra_http_headers,
"max_screenshot_scrolling_times": max_screenshot_scrolling_times,
"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,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
WorkflowRunResponse,
parse_obj_as(
type_=WorkflowRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_scripts(
page=1,
page_size=10,
)
"""
_response = self._client_wrapper.httpx_client.request(
"v1/scripts",
method="GET",
params={
"page": page,
"page_size": page_size,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[Script],
parse_obj_as(
type_=typing.List[Script], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.create_script()
"""
_response = self._client_wrapper.httpx_client.request(
"v1/scripts",
method="POST",
json={
"workflow_id": workflow_id,
"run_id": run_id,
"files": convert_and_respect_annotation_metadata(
object_=files, annotation=typing.Sequence[ScriptFileCreate], direction="write"
),
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CreateScriptResponse,
parse_obj_as(
type_=CreateScriptResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.get_script(
script_id="s_abc123",
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/scripts/{jsonable_encoder(script_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Script,
parse_obj_as(
type_=Script, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
client.deploy_script(
script_id="s_abc123",
files=[
ScriptFileCreate(
path="src/main.py",
content="content",
)
],
)
"""
_response = self._client_wrapper.httpx_client.request(
f"v1/scripts/{jsonable_encoder(script_id)}/deploy",
method="POST",
json={
"files": convert_and_respect_annotation_metadata(
object_=files, annotation=typing.Sequence[ScriptFileCreate], direction="write"
),
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CreateScriptResponse,
parse_obj_as(
type_=CreateScriptResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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.PRODUCTION
api_key : typing.Optional[str]
x_api_key : str
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",
x_api_key="YOUR_X_API_KEY",
)
"""
def __init__(
self,
*,
base_url: typing.Optional[str] = None,
environment: SkyvernEnvironment = SkyvernEnvironment.PRODUCTION,
api_key: typing.Optional[str] = None,
x_api_key: str,
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 None
self._client_wrapper = AsyncClientWrapper(
base_url=_get_base_url(base_url=base_url, environment=environment),
api_key=api_key,
x_api_key=x_api_key,
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.scripts = AsyncScriptsClient(client_wrapper=self._client_wrapper)
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[ProxyLocation] = OMIT,
data_extraction_schema: typing.Optional[TaskRunRequestDataExtractionSchema] = 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,
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,
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-2.0`, the latest Skyvern agent that performs pretty well with complex and multi-step tasks. `skyvern-1.0` is good for simple tasks like filling a form, or searching for information on Google. 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[ProxyLocation]
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_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California
- US-NY: New York
- US-TX: Texas
- US-FL: Florida
- US-WA: Washington
- NONE: No proxy
data_extraction_schema : typing.Optional[TaskRunRequestDataExtractionSchema]
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.
publish_workflow : typing.Optional[bool]
Whether to publish this task as a reusable workflow. Only available for skyvern-2.0.
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.
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.run_task(
prompt="Find the top 3 posts on Hacker News.",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/run/tasks",
method="POST",
json={
"prompt": prompt,
"url": url,
"engine": engine,
"title": title,
"proxy_location": proxy_location,
"data_extraction_schema": convert_and_respect_annotation_metadata(
object_=data_extraction_schema, annotation=TaskRunRequestDataExtractionSchema, direction="write"
),
"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,
"publish_workflow": publish_workflow,
"include_action_history_in_verification": include_action_history_in_verification,
"max_screenshot_scrolls": max_screenshot_scrolls,
"browser_address": browser_address,
},
headers={
"x-user-agent": str(user_agent) if user_agent is not None else None,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
TaskRunResponse,
parse_obj_as(
type_=TaskRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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[ProxyLocation] = 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,
max_screenshot_scrolls: typing.Optional[int] = OMIT,
extra_http_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,
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[ProxyLocation]
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_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California
- US-NY: New York
- US-TX: Texas
- US-FL: Florida
- US-WA: Washington
- NONE: No proxy
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
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.
extra_http_headers : typing.Optional[typing.Dict[str, typing.Optional[str]]]
The extra HTTP headers for the requests in browser.
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.
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.run_workflow(
workflow_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/run/workflows",
method="POST",
params={
"template": template,
},
json={
"workflow_id": workflow_id,
"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,
"max_screenshot_scrolls": max_screenshot_scrolls,
"extra_http_headers": extra_http_headers,
"browser_address": browser_address,
"ai_fallback": ai_fallback,
"run_with": run_with,
},
headers={
"x-max-steps-override": str(max_steps_override) if max_steps_override is not None else None,
"x-user-agent": str(user_agent) if user_agent is not None else None,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
WorkflowRunResponse,
parse_obj_as(
type_=WorkflowRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_run(
run_id="tsk_123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
GetRunResponse,
parse_obj_as(
type_=GetRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.cancel_run(
run_id="run_id",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/cancel",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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,
search_key: typing.Optional[str] = None,
title: typing.Optional[str] = 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 both
`workflows.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]
search_key : typing.Optional[str]
Unified search across workflow title and parameter metadata (key, description, default_value).
title : typing.Optional[str]
Deprecated: use search_key instead.
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_workflows()
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/workflows",
method="GET",
params={
"page": page,
"page_size": page_size,
"only_saved_tasks": only_saved_tasks,
"only_workflows": only_workflows,
"search_key": search_key,
"title": title,
"template": template,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[Workflow],
parse_obj_as(
type_=typing.List[Workflow], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def create_workflow(
self,
*,
json_definition: typing.Optional[WorkflowCreateYamlRequest] = OMIT,
yaml_definition: typing.Optional[str] = OMIT,
request_options: typing.Optional[RequestOptions] = None,
) -> Workflow:
"""
Create a new workflow
Parameters
----------
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.create_workflow()
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/workflows",
method="POST",
json={
"json_definition": convert_and_respect_annotation_metadata(
object_=json_definition, annotation=WorkflowCreateYamlRequest, direction="write"
),
"yaml_definition": yaml_definition,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Workflow,
parse_obj_as(
type_=Workflow, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.update_workflow(
workflow_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/workflows/{jsonable_encoder(workflow_id)}",
method="POST",
json={
"json_definition": convert_and_respect_annotation_metadata(
object_=json_definition, annotation=WorkflowCreateYamlRequest, direction="write"
),
"yaml_definition": yaml_definition,
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Workflow,
parse_obj_as(
type_=Workflow, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.delete_workflow(
workflow_id="wpid_123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/workflows/{jsonable_encoder(workflow_id)}/delete",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def get_artifact(
self, artifact_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> Artifact:
"""
Get an artifact
Parameters
----------
artifact_id : str
request_options : typing.Optional[RequestOptions]
Request-specific configuration.
Returns
-------
Artifact
Successfully retrieved artifact
Examples
--------
import asyncio
from skyvern import AsyncSkyvern
client = AsyncSkyvern(
api_key="YOUR_API_KEY",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_artifact(
artifact_id="artifact_id",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/artifacts/{jsonable_encoder(artifact_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Artifact,
parse_obj_as(
type_=Artifact, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_run_artifacts(
run_id="run_id",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/artifacts",
method="GET",
params={
"artifact_type": artifact_type,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[Artifact],
parse_obj_as(
type_=typing.List[Artifact], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def retry_run_webhook(
self, run_id: str, *, request_options: typing.Optional[RequestOptions] = None
) -> typing.Optional[typing.Any]:
"""
Retry sending the webhook for a run
Parameters
----------
run_id : str
The id of the task run or the workflow run.
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.retry_run_webhook(
run_id="tsk_123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/retry_webhook",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_run_timeline(
run_id="wr_123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/runs/{jsonable_encoder(run_id)}/timeline",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[WorkflowRunTimeline],
parse_obj_as(
type_=typing.List[WorkflowRunTimeline], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 400:
raise BadRequestError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_browser_sessions()
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/browser_sessions",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[BrowserSessionResponse],
parse_obj_as(
type_=typing.List[BrowserSessionResponse], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def create_browser_session(
self,
*,
timeout: typing.Optional[int] = OMIT,
proxy_location: typing.Optional[ProxyLocation] = 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[ProxyLocation]
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_ZA: South Africa
- RESIDENTIAL_AR: Argentina
- RESIDENTIAL_AU: Australia
- RESIDENTIAL_ISP: ISP proxy
- US-CA: California
- US-NY: New York
- US-TX: Texas
- US-FL: Florida
- US-WA: Washington
- NONE: No proxy
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.create_browser_session()
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/browser_sessions",
method="POST",
json={
"timeout": timeout,
"proxy_location": proxy_location,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
BrowserSessionResponse,
parse_obj_as(
type_=BrowserSessionResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.close_browser_session(
browser_session_id="pbs_123456",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/browser_sessions/{jsonable_encoder(browser_session_id)}/close",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_browser_session(
browser_session_id="pbs_123456",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/browser_sessions/{jsonable_encoder(browser_session_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
BrowserSessionResponse,
parse_obj_as(
type_=BrowserSessionResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 403:
raise ForbiddenError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 404:
raise NotFoundError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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,
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
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",
x_api_key="YOUR_X_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._client_wrapper.httpx_client.request(
"v1/credentials/totp",
method="POST",
json={
"totp_identifier": totp_identifier,
"task_id": task_id,
"workflow_id": workflow_id,
"workflow_run_id": workflow_run_id,
"source": source,
"content": content,
"expired_at": expired_at,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
TotpCode,
parse_obj_as(
type_=TotpCode, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def get_credentials(
self,
*,
page: typing.Optional[int] = None,
page_size: typing.Optional[int] = 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
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_credentials(
page=1,
page_size=10,
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/credentials",
method="GET",
params={
"page": page,
"page_size": page_size,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[CredentialResponse],
parse_obj_as(
type_=typing.List[CredentialResponse], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def create_credential(
self,
*,
name: str,
credential_type: SkyvernForgeSdkSchemasCredentialsCredentialType,
credential: CreateCredentialRequestCredential,
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
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.create_credential(
name="My Credential",
credential_type="password",
credential=NonEmptyPasswordCredential(
password="securepassword123",
username="user@example.com",
totp="JBSWY3DPEHPK3PXP",
),
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/credentials",
method="POST",
json={
"name": name,
"credential_type": credential_type,
"credential": convert_and_respect_annotation_metadata(
object_=credential, annotation=CreateCredentialRequestCredential, direction="write"
),
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CredentialResponse,
parse_obj_as(
type_=CredentialResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.delete_credential(
credential_id="cred_1234567890",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/credentials/{jsonable_encoder(credential_id)}/delete",
method="POST",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_credential(
credential_id="cred_1234567890",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/credentials/{jsonable_encoder(credential_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CredentialResponse,
parse_obj_as(
type_=CredentialResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
async def login(
self,
*,
credential_type: SkyvernSchemasRunBlocksCredentialType,
url: typing.Optional[str] = OMIT,
prompt: 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_address: typing.Optional[str] = OMIT,
extra_http_headers: typing.Optional[typing.Dict[str, typing.Optional[str]]] = OMIT,
max_screenshot_scrolling_times: typing.Optional[int] = 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 : SkyvernSchemasRunBlocksCredentialType
Where to get the credential from
url : typing.Optional[str]
Website url
prompt : typing.Optional[str]
Login instructions. Skyvern has default prompt/instruction for login if this field is not provided.
webhook_url : typing.Optional[str]
Webhook URL to send login 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_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
max_screenshot_scrolling_times : typing.Optional[int]
Maximum number of times to scroll for screenshots
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.login(
credential_type="skyvern",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/run/tasks/login",
method="POST",
json={
"credential_type": credential_type,
"url": url,
"prompt": prompt,
"webhook_url": webhook_url,
"proxy_location": proxy_location,
"totp_identifier": totp_identifier,
"totp_url": totp_url,
"browser_session_id": browser_session_id,
"browser_address": browser_address,
"extra_http_headers": extra_http_headers,
"max_screenshot_scrolling_times": max_screenshot_scrolling_times,
"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,
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
WorkflowRunResponse,
parse_obj_as(
type_=WorkflowRunResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_scripts(
page=1,
page_size=10,
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/scripts",
method="GET",
params={
"page": page,
"page_size": page_size,
},
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
typing.List[Script],
parse_obj_as(
type_=typing.List[Script], # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.create_script()
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
"v1/scripts",
method="POST",
json={
"workflow_id": workflow_id,
"run_id": run_id,
"files": convert_and_respect_annotation_metadata(
object_=files, annotation=typing.Sequence[ScriptFileCreate], direction="write"
),
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CreateScriptResponse,
parse_obj_as(
type_=CreateScriptResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_API_KEY",
)
async def main() -> None:
await client.get_script(
script_id="s_abc123",
)
asyncio.run(main())
"""
_response = await self._client_wrapper.httpx_client.request(
f"v1/scripts/{jsonable_encoder(script_id)}",
method="GET",
request_options=request_options,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
Script,
parse_obj_as(
type_=Script, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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",
x_api_key="YOUR_X_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._client_wrapper.httpx_client.request(
f"v1/scripts/{jsonable_encoder(script_id)}/deploy",
method="POST",
json={
"files": convert_and_respect_annotation_metadata(
object_=files, annotation=typing.Sequence[ScriptFileCreate], direction="write"
),
},
headers={
"content-type": "application/json",
},
request_options=request_options,
omit=OMIT,
)
try:
if 200 <= _response.status_code < 300:
return typing.cast(
CreateScriptResponse,
parse_obj_as(
type_=CreateScriptResponse, # type: ignore
object_=_response.json(),
),
)
if _response.status_code == 422:
raise UnprocessableEntityError(
typing.cast(
typing.Optional[typing.Any],
parse_obj_as(
type_=typing.Optional[typing.Any], # type: ignore
object_=_response.json(),
),
)
)
_response_json = _response.json()
except JSONDecodeError:
raise ApiError(status_code=_response.status_code, body=_response.text)
raise ApiError(status_code=_response.status_code, body=_response_json)
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")