vikarti.anatra/Skyvern
1
0
Fork 0
mirror of https://github.com/Skyvern-AI/skyvern.git synced 2026-07-09 16:09:13 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 6

chore: remove deprecated LangChain and LlamaIndex integrations (#7095)
Some checks are pending
Auto Create GitHub Release on Version Change / check-version-change (push) Waiting to run
Details
Auto Create GitHub Release on Version Change / create-release (push) Blocked by required conditions
Details
Run tests and pre-commit / Run tests and pre-commit hooks (push) Waiting to run
Details
Run tests and pre-commit / Frontend Lint and Build (push) Waiting to run
Details
Run tests and pre-commit / pip Package Smoke Tests (3.11) (push) Waiting to run
Details
Run tests and pre-commit / pip Package Smoke Tests (3.13) (push) Waiting to run
Details
Publish Fern Docs / run (push) Waiting to run
Details
Build Skyvern SDK and publish to PyPI / check-version-change (push) Waiting to run
Details
Build Skyvern SDK and publish to PyPI / run-ci (push) Blocked by required conditions
Details
Build Skyvern SDK and publish to PyPI / build-sdk (push) Blocked by required conditions
Details
Build Skyvern SDK and publish to PyPI / publish-sdk (push) Blocked by required conditions
Details

Browse source 
Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
Shuchang Zheng 2026-07-08 23:38:51 -07:00 • committed by GitHub
parent edf38e68fc
commit e174afd63f
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
17 changed files with 0 additions and 5197 deletions
Download patch file Download diff file Expand all files Collapse all files

286
integrations/langchain/README.md
View file

@ -1,286 +0,0 @@
<!-- START doctoc generated TOC please keep comment here to allow auto update -->
<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
- [Skyvern Langchain](#skyvern-langchain)
- [Installation](#installation)
- [Basic Usage](#basic-usage)
- [Run a task(sync) locally in your local environment](#run-a-tasksync-locally-in-your-local-environment)
- [Run a task(async) locally in your local environment](#run-a-taskasync-locally-in-your-local-environment)
- [Get a task locally in your local environment](#get-a-task-locally-in-your-local-environment)
- [Run a task(sync) by calling skyvern APIs](#run-a-tasksync-by-calling-skyvern-apis)
- [Run a task(async) by calling skyvern APIs](#run-a-taskasync-by-calling-skyvern-apis)
- [Get a task by calling skyvern APIs](#get-a-task-by-calling-skyvern-apis)
- [Agent Usage](#agent-usage)
- [Run a task(async) locally in your local environment and wait until the task is finished](#run-a-taskasync-locally-in-your-local-environment-and-wait-until-the-task-is-finished)
- [Run a task(async) by calling skyvern APIs and wait until the task is finished](#run-a-taskasync-by-calling-skyvern-apis-and-wait-until-the-task-is-finished)
<!-- END doctoc generated TOC please keep comment here to allow auto update -->
# Skyvern Langchain
This is a langchain integration for Skyvern.
## Installation
```bash
pip install skyvern-langchain
```
To run the example scenarios, you might need to install other langchain dependencies.
```bash
pip install langchain-openai
pip install langchain-community
```
## Basic Usage
This is the only basic usage of skyvern langchain tool. If you want a full langchain integration experience, please refer to the [Agent Usage](#agent-usage) section to play with langchain agent.
Go to [Langchain Tools](https://python.langchain.com/v0.1/docs/modules/tools/) to see more advanced langchain tool usage.
### Run a task(sync) locally in your local environment
> sync task won't return until the task is finished.
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
import asyncio
from skyvern_langchain.agent import RunTask
run_task = RunTask()
async def main():
# to run skyvern agent locally, must run `skyvern init` first
print(await run_task.ainvoke("Navigate to the Hacker News homepage and get the top 3 posts."))
if __name__ == "__main__":
asyncio.run(main())
```
### Run a task(async) locally in your local environment
> async task will return immediately and the task will be running in the background.
:warning: :warning: if you want to run the task in the background, you need to keep the script running until the task is finished, otherwise the task will be killed when the script is finished.
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
import asyncio
from skyvern_langchain.agent import DispatchTask
dispatch_task = DispatchTask()
async def main():
# to run skyvern agent locally, must run `skyvern init` first
print(await dispatch_task.ainvoke("Navigate to the Hacker News homepage and get the top 3 posts."))
# keep the script running until the task is finished
await asyncio.sleep(600)
if __name__ == "__main__":
asyncio.run(main())
```
### Get a task locally in your local environment
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
import asyncio
from skyvern_langchain.agent import GetTask
get_task = GetTask()
async def main():
# to run skyvern agent locally, must run `skyvern init` first
print(await get_task.ainvoke("<task_id>"))
if __name__ == "__main__":
asyncio.run(main())
```
### Run a task(sync) by calling skyvern APIs
> sync task won't return until the task is finished.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
```python
import asyncio
from skyvern_langchain.client import RunTask
run_task = RunTask(
api_key="<your_organization_api_key>",
)
# or you can load the api_key from SKYVERN_API_KEY in .env
# run_task = RunTask()
async def main():
print(await run_task.ainvoke("Navigate to the Hacker News homepage and get the top 3 posts."))
if __name__ == "__main__":
asyncio.run(main())
```
### Run a task(async) by calling skyvern APIs
> async task will return immediately and the task will be running in the background.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
the task is actually running in the skyvern cloud service, so you don't need to keep your script running until the task is finished.
```python
import asyncio
from skyvern_langchain.client import DispatchTask
dispatch_task = DispatchTask(
api_key="<your_organization_api_key>",
)
# or you can load the api_key from SKYVERN_API_KEY in .env
# dispatch_task = DispatchTask()
async def main():
print(await dispatch_task.ainvoke("Navigate to the Hacker News homepage and get the top 3 posts."))
if __name__ == "__main__":
asyncio.run(main())
```
### Get a task by calling skyvern APIs
> async task will return immediately and the task will be running in the background.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
the task is actually running in the skyvern cloud service, so you don't need to keep your script running until the task is finished.
```python
import asyncio
from skyvern_langchain.client import GetTask
get_task = GetTask(
api_key="<your_organization_api_key>",
)
# or you can load the api_key from SKYVERN_API_KEY in .env
# get_task = GetTask()
async def main():
print(await get_task.ainvoke("<task_id>"))
if __name__ == "__main__":
asyncio.run(main())
```
## Agent Usage
Langchain is more powerful when used with [Langchain Agents](https://python.langchain.com/v0.1/docs/modules/agents/).
The following two examples show how to build an agent that executes a specified task, waits for its completion, and then returns the results. For example, the agent is tasked with navigating to the Hacker News homepage and retrieving the top three posts.
### Run a task(async) locally in your local environment and wait until the task is finished
> async task will return immediately and the task will be running in the background. You can use `GetTask` tool to poll the task information until the task is finished.
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from skyvern_langchain.agent import DispatchTask, GetTask
from langchain_community.tools.sleep.tool import SleepTool
# load OpenAI API key from .env
load_dotenv()
llm = ChatOpenAI(model="gpt-4o", temperature=0)
dispatch_task = DispatchTask()
get_task = GetTask()
agent = initialize_agent(
llm=llm,
tools=[
dispatch_task,
get_task,
SleepTool(),
],
verbose=True,
agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
)
async def main():
# use sleep tool to set up the polling logic until the task is completed, if you only want to dispatch a task, you can remove the sleep tool
print(await agent.ainvoke("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.' Then, get this task information until it's completed. The task information re-get interval should be 60s."))
if __name__ == "__main__":
asyncio.run(main())
```
### Run a task(async) by calling skyvern APIs and wait until the task is finished
> async task will return immediately and the task will be running in the background. You can use `GetTask` tool to poll the task information until the task is finished.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
```python
import asyncio
from dotenv import load_dotenv
from langchain_openai import ChatOpenAI
from langchain.agents import initialize_agent, AgentType
from skyvern_langchain.client import DispatchTask, GetTask
from langchain_community.tools.sleep.tool import SleepTool
# load OpenAI API key from .env
load_dotenv()
llm = ChatOpenAI(model="gpt-4o", temperature=0)
dispatch_task = DispatchTask(
api_key="<your_organization_api_key>",
)
# or you can load the api_key from SKYVERN_API_KEY in .env
# dispatch_task = DispatchTask()
get_task = GetTask(
api_key="<your_organization_api_key>",
)
# or you can load the api_key from SKYVERN_API_KEY in .env
# get_task = GetTask()
agent = initialize_agent(
llm=llm,
tools=[
dispatch_task,
get_task,
SleepTool(),
],
verbose=True,
agent=AgentType.STRUCTURED_CHAT_ZERO_SHOT_REACT_DESCRIPTION,
)
async def main():
# use sleep tool to set up the polling logic until the task is completed, if you only want to dispatch a task, you can remove the sleep tool
print(await agent.ainvoke("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.' Then, get this task information until it's completed. The task information re-get interval should be 60s."))
if __name__ == "__main__":
asyncio.run(main())
```

53
integrations/langchain/pyproject.toml
View file

@ -1,53 +0,0 @@
[project]
name = "skyvern-langchain"
version = "0.2.1"
description = ""
authors = [{ name = "lawyzheng", email = "lawy@skyvern.com" }]
requires-python = ">=3.11,<3.14"
readme = "README.md"
dependencies = [
"skyvern>=0.2.0",
"langchain>=1.2.0",
"langchain-core>=1.3.3",
"urllib3>=2.7.0",
]
[tool.uv]
override-dependencies = [
"authlib>=1.6.11",
"pyasn1>=0.6.3",
"pyjwt>=2.12.0",
"fastmcp>=3.2.0",
"mcp>=1.23.0",
"websockets>=15.0.1",
# litellm 1.83.7+ pins these exactly; relax for security upgrade.
"jsonschema>=4.23.0",
"python-dotenv>=1.2.2",
"openai>=2.24.0",
# Dependabot moderate security upgrades (transitive)
"mako>=1.3.11",
"pypdf>=6.10.2",
"python-multipart>=0.0.26",
"cryptography>=48.0.1",
"pygments>=2.20.0",
# Dependabot high security upgrades (transitive)
"litellm>=1.84.0",
"langsmith>=0.8.0",
]
[tool.uv.sources]
skyvern = { path = "../.." }
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[dependency-groups]
dev = ["twine>=6.1.0,<7"]
[tool.hatch.build.targets.sdist]
include = ["skyvern_langchain"]
[tool.hatch.build.targets.wheel]
include = ["skyvern_langchain"]

0
integrations/langchain/skyvern_langchain/__init__.py
View file

60
integrations/langchain/skyvern_langchain/agent.py
View file

@ -1,60 +0,0 @@
from typing import Any, Type
from langchain.tools import BaseTool
from litellm import BaseModel
from pydantic import Field
from skyvern_langchain.schema import CreateTaskInput, GetTaskInput
from skyvern_langchain.settings import settings
from skyvern import Skyvern
from skyvern.client.types.get_run_response import GetRunResponse
from skyvern.client.types.task_run_response import TaskRunResponse
from skyvern.schemas.runs import RunEngine
class SkyvernTaskBaseTool(BaseTool):
engine: RunEngine = Field(default=settings.engine)
run_task_timeout_seconds: int = Field(default=settings.run_task_timeout_seconds)
agent: Skyvern = Skyvern.local()
def _run(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError("skyvern task tool does not support sync")
class RunTask(SkyvernTaskBaseTool):
name: str = "run-skyvern-agent-task"
description: str = """Use Skyvern agent to run a task. This function won't return until the task is finished."""
args_schema: Type[BaseModel] = CreateTaskInput
async def _arun(self, user_prompt: str, url: str | None = None) -> TaskRunResponse:
return await self.agent.run_task(
prompt=user_prompt,
url=url,
engine=self.engine,
timeout=self.run_task_timeout_seconds,
wait_for_completion=True,
)
class DispatchTask(SkyvernTaskBaseTool):
name: str = "dispatch-skyvern-agent-task"
description: str = """Use Skyvern agent to dispatch a task. This function will return immediately and the task will be running in the background."""
args_schema: Type[BaseModel] = CreateTaskInput
async def _arun(self, user_prompt: str, url: str | None = None) -> TaskRunResponse:
return await self.agent.run_task(
prompt=user_prompt,
url=url,
engine=self.engine,
timeout=self.run_task_timeout_seconds,
wait_for_completion=False,
)
class GetTask(SkyvernTaskBaseTool):
name: str = "get-skyvern-agent-task"
description: str = """Use Skyvern agent to get a task."""
args_schema: Type[BaseModel] = GetTaskInput
async def _arun(self, task_id: str) -> GetRunResponse | None:
return await self.agent.get_run(run_id=task_id)

64
integrations/langchain/skyvern_langchain/client.py
View file

@ -1,64 +0,0 @@
from typing import Any, Type
from langchain.tools import BaseTool
from pydantic import BaseModel, Field
from skyvern_langchain.schema import CreateTaskInput, GetTaskInput
from skyvern_langchain.settings import settings
from skyvern import Skyvern
from skyvern.client import SkyvernEnvironment
from skyvern.client.types.get_run_response import GetRunResponse
from skyvern.client.types.task_run_response import TaskRunResponse
from skyvern.schemas.runs import RunEngine
class SkyvernTaskBaseTool(BaseTool):
api_key: str = Field(default=settings.api_key)
base_url: str = Field(default=settings.base_url)
engine: RunEngine = Field(default=settings.engine)
run_task_timeout_seconds: int = Field(default=settings.run_task_timeout_seconds)
def get_client(self) -> Skyvern:
return Skyvern(environment=SkyvernEnvironment.CLOUD, base_url=self.base_url, api_key=self.api_key)
def _run(self, *args: Any, **kwargs: Any) -> None:
raise NotImplementedError("skyvern task tool does not support sync")
class RunTask(SkyvernTaskBaseTool):
name: str = "run-skyvern-client-task"
description: str = """Use Skyvern client to run a task. This function won't return until the task is finished."""
args_schema: Type[BaseModel] = CreateTaskInput
async def _arun(self, user_prompt: str, url: str | None = None) -> TaskRunResponse:
return await self.get_client().run_task(
timeout=self.run_task_timeout_seconds,
url=url,
prompt=user_prompt,
engine=self.engine,
wait_for_completion=True,
)
class DispatchTask(SkyvernTaskBaseTool):
name: str = "dispatch-skyvern-client-task"
description: str = """Use Skyvern client to dispatch a task. This function will return immediately and the task will be running in the background."""
args_schema: Type[BaseModel] = CreateTaskInput
async def _arun(self, user_prompt: str, url: str | None = None) -> TaskRunResponse:
return await self.get_client().run_task(
timeout=self.run_task_timeout_seconds,
url=url,
prompt=user_prompt,
engine=self.engine,
wait_for_completion=False,
)
class GetTask(SkyvernTaskBaseTool):
name: str = "get-skyvern-client-task"
description: str = """Use Skyvern client to get a task."""
args_schema: Type[BaseModel] = GetTaskInput
async def _arun(self, task_id: str) -> GetRunResponse | None:
return await self.get_client().get_run(run_id=task_id)

0
integrations/langchain/skyvern_langchain/py.typed
View file

10
integrations/langchain/skyvern_langchain/schema.py
View file

@ -1,10 +0,0 @@
from pydantic import BaseModel
class CreateTaskInput(BaseModel):
user_prompt: str
url: str | None = None
class GetTaskInput(BaseModel):
task_id: str

18
integrations/langchain/skyvern_langchain/settings.py
View file

@ -1,18 +0,0 @@
from dotenv import load_dotenv
from pydantic_settings import BaseSettings
from skyvern.schemas.runs import RunEngine
class Settings(BaseSettings):
api_key: str = ""
base_url: str = "https://api.skyvern.com"
engine: RunEngine = RunEngine.skyvern_v2
run_task_timeout_seconds: int = 60 * 60
class Config:
env_prefix = "SKYVERN_"
load_dotenv()
settings = Settings()

1463
integrations/langchain/uv.lock generated
View file

File diff suppressed because it is too large Load diff

294
integrations/llama_index/README.md
View file

@ -1,294 +0,0 @@
<!-- START doctoc generated TOC please keep comment here to allow auto update -->
<!-- DON'T EDIT THIS SECTION, INSTEAD RE-RUN doctoc TO UPDATE -->
- [Skyvern LlamaIndex](#skyvern-llamaindex)
- [Installation](#installation)
- [Basic Usage](#basic-usage)
- [Run a task(sync) locally in your local environment](#run-a-tasksync-locally-in-your-local-environment)
- [Run a task(async) locally in your local environment](#run-a-taskasync-locally-in-your-local-environment)
- [Get a task locally in your local environment](#get-a-task-locally-in-your-local-environment)
- [Run a task(sync) by calling skyvern APIs](#run-a-tasksync-by-calling-skyvern-apis)
- [Run a task(async) by calling skyvern APIs](#run-a-taskasync-by-calling-skyvern-apis)
- [Get a task by calling skyvern APIs](#get-a-task-by-calling-skyvern-apis)
- [Advanced Usage](#advanced-usage)
- [Dispatch a task(async) locally in your local environment and wait until the task is finished](#dispatch-a-taskasync-locally-in-your-local-environment-and-wait-until-the-task-is-finished)
- [Dispatch a task(async) by calling skyvern APIs and wait until the task is finished](#dispatch-a-taskasync-by-calling-skyvern-apis-and-wait-until-the-task-is-finished)
<!-- END doctoc generated TOC please keep comment here to allow auto update -->
# Skyvern LlamaIndex
This is a LlamaIndex integration for Skyvern.
## Installation
```bash
pip install skyvern-llamaindex
```
## Basic Usage
### Run a task(sync) locally in your local environment
> sync task won't return until the task is finished.
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from skyvern_llamaindex.agent import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.run_task()],
llm=OpenAI(model="gpt-4o"),
verbose=True,
)
response = agent.chat("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.'")
print(response)
```
### Run a task(async) locally in your local environment
> async task will return immediately and the task will be running in the background.
:warning: :warning: if you want to run the task in the background, you need to keep the agent running until the task is finished, otherwise the task will be killed when the agent finished the chat.
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
import asyncio
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from skyvern_llamaindex.agent import SkyvernTool
from llama_index.core.tools import FunctionTool
# load OpenAI API key from .env
load_dotenv()
async def sleep(seconds: int) -> str:
await asyncio.sleep(seconds)
return f"Slept for {seconds} seconds"
# define a sleep tool to keep the agent running until the task is finished
sleep_tool = FunctionTool.from_defaults(
async_fn=sleep,
description="Sleep for a given number of seconds",
name="sleep",
)
skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.dispatch_task(), sleep_tool],
llm=OpenAI(model="gpt-4o"),
verbose=True,
)
response = agent.chat("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.' Then, sleep for 10 minutes.")
print(response)
```
### Get a task locally in your local environment
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from skyvern_llamaindex.agent import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.get_task()],
llm=OpenAI(model="gpt-4o"),
verbose=True,
)
response = agent.chat("Get the task information with Skyvern. The task id is '<task_id>'.")
print(response)
```
### Run a task(sync) by calling skyvern APIs
> sync task won't return until the task is finished.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
```python
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from skyvern_llamaindex.client import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
skyvern_tool = SkyvernTool(api_key="<your_organization_api_key>")
# or you can load the api_key from SKYVERN_API_KEY in .env
# skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.run_task()],
llm=OpenAI(model="gpt-4o"),
verbose=True,
)
response = agent.chat("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.'")
print(response)
```
### Run a task(async) by calling skyvern APIs
> async task will return immediately and the task will be running in the background.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
the task is actually running in the skyvern cloud service, so you don't need to keep your agent running until the task is finished.
```python
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from skyvern_llamaindex.client import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
skyvern_tool = SkyvernTool(api_key="<your_organization_api_key>")
# or you can load the api_key from SKYVERN_API_KEY in .env
# skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.dispatch_task()],
llm=OpenAI(model="gpt-4o"),
verbose=True,
)
response = agent.chat("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.'")
print(response)
```
### Get a task by calling skyvern APIs
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
```python
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from skyvern_llamaindex.client import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
skyvern_tool = SkyvernTool(api_key="<your_organization_api_key>")
# or you can load the api_key from SKYVERN_API_KEY in .env
# skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.get_task()],
llm=OpenAI(model="gpt-4o"),
verbose=True,
)
response = agent.chat("Get the task information with Skyvern. The task id is '<task_id>'.")
print(response)
```
## Advanced Usage
To provide some examples of how to integrate Skyvern with other llama-index tools in the agent.
### Dispatch a task(async) locally in your local environment and wait until the task is finished
> dispatch task will return immediately and the task will be running in the background. You can use `get_task` tool to poll the task information until the task is finished.
:warning: :warning: if you want to run this code block, you need to run `skyvern init` command in your terminal to set up skyvern first.
```python
import asyncio
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from llama_index.core.tools import FunctionTool
from skyvern_llamaindex.agent import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
async def sleep(seconds: int) -> str:
await asyncio.sleep(seconds)
return f"Slept for {seconds} seconds"
sleep_tool = FunctionTool.from_defaults(
async_fn=sleep,
description="Sleep for a given number of seconds",
name="sleep",
)
skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.dispatch_task(), skyvern_tool.get_task(), sleep_tool],
llm=OpenAI(model="gpt-4o"),
verbose=True,
max_function_calls=10,
)
response = agent.chat("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.' Then, get this task information until it's completed. The task information re-get interval should be 60s.")
print(response)
```
### Dispatch a task(async) by calling skyvern APIs and wait until the task is finished
> dispatch task will return immediately and the task will be running in the background. You can use `get_task` tool to poll the task information until the task is finished.
no need to run `skyvern init` command in your terminal to set up skyvern before using this integration.
```python
import asyncio
from dotenv import load_dotenv
from llama_index.agent.openai import OpenAIAgent
from llama_index.llms.openai import OpenAI
from llama_index.core.tools import FunctionTool
from skyvern_llamaindex.client import SkyvernTool
# load OpenAI API key from .env
load_dotenv()
async def sleep(seconds: int) -> str:
await asyncio.sleep(seconds)
return f"Slept for {seconds} seconds"
sleep_tool = FunctionTool.from_defaults(
async_fn=sleep,
description="Sleep for a given number of seconds",
name="sleep",
)
skyvern_tool = SkyvernTool(api_key="<your_organization_api_key>")
# or you can load the api_key from SKYVERN_API_KEY in .env
# skyvern_tool = SkyvernTool()
agent = OpenAIAgent.from_tools(
tools=[skyvern_tool.dispatch_task(), skyvern_tool.get_task(), sleep_tool],
llm=OpenAI(model="gpt-4o"),
verbose=True,
max_function_calls=10,
)
response = agent.chat("Run a task with Skyvern. The task is about 'Navigate to the Hacker News homepage and get the top 3 posts.' Then, get this task information until it's completed. The task information re-get interval should be 60s.")
print(response)
```

67
integrations/llama_index/pyproject.toml
View file

@ -1,67 +0,0 @@
[project]
name = "skyvern-llamaindex"
version = "0.2.2"
description = "Skyvern integration for LlamaIndex"
authors = [{ name = "lawyzheng", email = "lawy@skyvern.com" }]
requires-python = ">=3.11,<3.14"
readme = "README.md"
dependencies = [
"skyvern>=0.2.0",
"llama-index>=0.12.19,<0.14",
# Dependabot #645 (GHSA-qccp-gfcp-xxvc / CVE-2026-44431): urllib3 < 2.7.0
# forwards sensitive headers across origins on proxied low-level redirects.
"urllib3>=2.7.0",
]
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[dependency-groups]
dev = ["twine>=6.1.0,<7"]
# Security: override vulnerable transitive dependency versions (SKY-8441)
# Using override-dependencies because several of these conflict with
# skyvern's pinned ranges (e.g. pypdf<6, python-multipart<0.0.19).
[tool.uv]
override-dependencies = [
"authlib>=1.6.11",
"pyasn1>=0.6.3",
"PyJWT>=2.12.0",
"fastmcp>=3.2.0",
"mcp>=1.23.0",
"pypdf>=6.10.2",
"pillow>=12.1.1",
"nltk>=3.9.3",
"python-multipart>=0.0.26",
"starlette>=0.49.1",
"google-cloud-aiplatform>=1.133.0",
# Cascading overrides needed to unblock the above security constraints:
# fastmcp>=3.2.0 requires websockets>=15 (skyvern pins <13)
"websockets>=15.0.1",
# litellm 1.83.7+ pins openai==2.24.0 (security upgrade); override
# llama-index-llms-openai's openai<2 declaration so the integration can
# take the patched litellm.
"openai>=2.24.0",
# litellm 1.83.7+ also pins these exactly.
"jsonschema>=4.23.0",
"python-dotenv>=1.0.0",
# Dependabot moderate security upgrades (transitive)
"mako>=1.3.11",
# Dependabot #718 (GHSA-79v4-65xg-pq4g): cryptography < 48.0.1 is vulnerable.
"cryptography>=48.0.1",
"pygments>=2.20.0",
# Dependabot #631 (GHSA-gphh-9q3h-jgpp / CVE-2026-44209): banks <= 2.4.1
# renders prompt templates with an unsandboxed Jinja2 environment (SSTI -> RCE).
# Pulled in transitively via llama-index-core. Fixed in 2.4.2.
"banks>=2.4.2",
]
[tool.uv.sources]
skyvern = { path = "../.." }
[tool.hatch.build.targets.sdist]
include = ["skyvern_llamaindex"]
[tool.hatch.build.targets.wheel]
include = ["skyvern_llamaindex"]

0
integrations/llama_index/skyvern_llamaindex/__init__.py
View file

129
integrations/llama_index/skyvern_llamaindex/agent.py
View file

@ -1,129 +0,0 @@
from typing import Any, List
from llama_index.core.tools import FunctionTool
from llama_index.core.tools.tool_spec.base import SPEC_FUNCTION_TYPE, BaseToolSpec
from skyvern_llamaindex.settings import settings
from skyvern import Skyvern
from skyvern.client.types.get_run_response import GetRunResponse
from skyvern.client.types.task_run_response import TaskRunResponse
from skyvern.schemas.runs import RunEngine
class SkyvernTool:
def __init__(self, agent: Skyvern | None = None):
if agent is None:
agent = Skyvern.local()
self.agent = agent
def run_task(self) -> FunctionTool:
task_tool_spec = SkyvernTaskToolSpec(agent=self.agent)
return task_tool_spec.to_tool_list(["run_task"])[0]
def dispatch_task(self) -> FunctionTool:
task_tool_spec = SkyvernTaskToolSpec(agent=self.agent)
return task_tool_spec.to_tool_list(["dispatch_task"])[0]
def get_task(self) -> FunctionTool:
task_tool_spec = SkyvernTaskToolSpec(agent=self.agent)
return task_tool_spec.to_tool_list(["get_task"])[0]
class SkyvernTaskToolSpec(BaseToolSpec):
spec_functions: List[SPEC_FUNCTION_TYPE] = [
"run_task",
"dispatch_task",
"get_task",
]
def __init__(
self,
*,
agent: Skyvern | None = None,
engine: RunEngine = settings.engine,
run_task_timeout_seconds: int = settings.run_task_timeout_seconds,
) -> None:
if agent is None:
agent = Skyvern.local()
self.agent = agent
self.engine = engine
self.run_task_timeout_seconds = run_task_timeout_seconds
async def run_task(
self,
user_prompt: str | None = None,
url: str | None = None,
*_: Any,
**kw: Any,
) -> TaskRunResponse:
"""
Use Skyvern agent to run a task. This function won't return until the task is finished.
Args:
user_prompt[str]: The user's prompt describing the task.
url (Optional[str]): The URL of the target website for the task.
"""
if user_prompt is None and kw.get("args"):
user_prompt = kw["args"][0]
if url is None:
if kw.get("args") and len(kw["args"]) > 1:
url = kw["args"][1]
elif kw.get("kwargs"):
url = kw["kwargs"].get("url")
assert user_prompt is not None, "user_prompt is required"
return await self.agent.run_task(
prompt=user_prompt,
url=url,
engine=self.engine,
timeout=self.run_task_timeout_seconds,
wait_for_completion=True,
)
async def dispatch_task(
self,
user_prompt: str | None = None,
url: str | None = None,
*_: Any,
**kw: Any,
) -> TaskRunResponse:
"""
Use Skyvern agent to dispatch a task. This function will return immediately and the task will be running in the background.
Args:
user_prompt[str]: The user's prompt describing the task.
url (Optional[str]): The URL of the target website for the task.
"""
if user_prompt is None and kw.get("args"):
user_prompt = kw["args"][0]
if url is None:
if kw.get("args") and len(kw["args"]) > 1:
url = kw["args"][1]
elif kw.get("kwargs"):
url = kw["kwargs"].get("url")
assert user_prompt is not None, "user_prompt is required"
return await self.agent.run_task(
prompt=user_prompt,
url=url,
engine=self.engine,
timeout=self.run_task_timeout_seconds,
wait_for_completion=False,
)
async def get_task(self, task_id: str | None = None, *_: Any, **kwargs: Any) -> GetRunResponse | None:
"""
Use Skyvern agent to get a task.
Args:
task_id[str]: The id of the task.
"""
if task_id is None and "args" in kwargs:
task_id = kwargs["args"][0]
assert task_id is not None, "task_id is required"
return await self.agent.get_run(run_id=task_id)

139
integrations/llama_index/skyvern_llamaindex/client.py
View file

@ -1,139 +0,0 @@
from typing import Any, List
from llama_index.core.tools import FunctionTool
from llama_index.core.tools.tool_spec.base import SPEC_FUNCTION_TYPE, BaseToolSpec
from pydantic import BaseModel
from skyvern_llamaindex.settings import settings
from skyvern import Skyvern
from skyvern.client import SkyvernEnvironment
from skyvern.client.types.get_run_response import GetRunResponse
from skyvern.client.types.task_run_response import TaskRunResponse
from skyvern.schemas.runs import RunEngine
class SkyvernTool(BaseModel):
api_key: str = settings.api_key
base_url: str = settings.base_url
def run_task(self) -> FunctionTool:
task_tool_spec = SkyvernTaskToolSpec(
api_key=self.api_key,
base_url=self.base_url,
)
return task_tool_spec.to_tool_list(["run_task"])[0]
def dispatch_task(self) -> FunctionTool:
task_tool_spec = SkyvernTaskToolSpec(
api_key=self.api_key,
base_url=self.base_url,
)
return task_tool_spec.to_tool_list(["dispatch_task"])[0]
def get_task(self) -> FunctionTool:
task_tool_spec = SkyvernTaskToolSpec(
api_key=self.api_key,
base_url=self.base_url,
)
return task_tool_spec.to_tool_list(["get_task"])[0]
class SkyvernTaskToolSpec(BaseToolSpec):
spec_functions: List[SPEC_FUNCTION_TYPE] = [
"run_task",
"dispatch_task",
"get_task",
]
def __init__(
self,
*,
api_key: str = settings.api_key,
base_url: str = settings.base_url,
engine: RunEngine = settings.engine,
run_task_timeout_seconds: int = settings.run_task_timeout_seconds,
):
self.engine = engine
self.run_task_timeout_seconds = run_task_timeout_seconds
self.client = Skyvern(environment=SkyvernEnvironment.CLOUD, base_url=base_url, api_key=api_key)
async def run_task(
self,
user_prompt: str | None = None,
url: str | None = None,
*_: Any,
**kw: Any,
) -> TaskRunResponse:
"""
Use Skyvern client to run a task. This function won't return until the task is finished.
Args:
user_prompt[str]: The user's prompt describing the task.
url (Optional[str]): The URL of the target website for the task.
"""
if user_prompt is None and kw.get("args"):
user_prompt = kw["args"][0]
if url is None:
if kw.get("args") and len(kw["args"]) > 1:
url = kw["args"][1]
elif kw.get("kwargs"):
url = kw["kwargs"].get("url")
assert user_prompt is not None, "user_prompt is required"
return await self.client.run_task(
prompt=user_prompt,
url=url,
engine=self.engine,
timeout=self.run_task_timeout_seconds,
wait_for_completion=True,
)
async def dispatch_task(
self,
user_prompt: str | None = None,
url: str | None = None,
*_: Any,
**kw: Any,
) -> TaskRunResponse:
"""
Use Skyvern client to dispatch a task. This function will return immediately and the task will be running in the background.
Args:
user_prompt[str]: The user's prompt describing the task.
url (Optional[str]): The URL of the target website for the task.
"""
if user_prompt is None and kw.get("args"):
user_prompt = kw["args"][0]
if url is None:
if kw.get("args") and len(kw["args"]) > 1:
url = kw["args"][1]
elif kw.get("kwargs"):
url = kw["kwargs"].get("url")
assert user_prompt is not None, "user_prompt is required"
return await self.client.run_task(
prompt=user_prompt,
url=url,
engine=self.engine,
timeout=self.run_task_timeout_seconds,
wait_for_completion=False,
)
async def get_task(self, task_id: str | None = None, *_: Any, **kwargs: Any) -> GetRunResponse | None:
"""
Use Skyvern client to get a task.
Args:
task_id[str]: The id of the task.
"""
if task_id is None and "args" in kwargs:
task_id = kwargs["args"][0]
assert task_id is not None, "task_id is required"
return await self.client.get_run(run_id=task_id)

0
integrations/llama_index/skyvern_llamaindex/py.typed
View file

18
integrations/llama_index/skyvern_llamaindex/settings.py
View file

@ -1,18 +0,0 @@
from dotenv import load_dotenv
from pydantic_settings import BaseSettings
from skyvern.schemas.runs import RunEngine
class Settings(BaseSettings):
api_key: str = ""
base_url: str = "https://api.skyvern.com"
engine: RunEngine = RunEngine.skyvern_v2
run_task_timeout_seconds: int = 60 * 60
class Config:
env_prefix = "SKYVERN_"
load_dotenv()
settings = Settings()

2596
integrations/llama_index/uv.lock generated
View file

File diff suppressed because it is too large Load diff