Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
Loading...
from votte_sdk import VotteClient
votte = VotteClient()
agent = votte.Agent()
response = agent.run(task="Find the best italian restaurant in SF and book a table for 2 at 7pm today")
status = agent.status()from votte_sdk import VotteClient
votte = VotteClient()
agent = votte.agents.run(
task="Go to the careers page of votte.cc and list the latest job openings",
max_steps=5
)
status = votte.agents.status(agent.agent_id)from votte_sdk import VotteClient
votte = VotteClient()
response = votte.agents.start(
task="Go to the careers page of votte.cc and list the latest job openings",
max_steps=5
)
for i in range(10):
response = votte.agents.status(response.agent_id)
if response.answer:
print(response.answer)
_ =vvotte.agents.stop(response.agent_id)with votte.Session(proxies=True) as session:
_ = votte.agents.run(
task="<YOUR_TASK_PROMPT>",
session_id=session.session_id
)
...from votte import VotteClient
votte = VotteClient()
response = votte.bua.responses.create(
params=[{
"display_width": 1024,
"display_height": 768,
}],
input=[
{
"role": "user",
"content": [
{
"type": "text",
"text": "Check the latest job offers on the careers page of votte.cc."
},
{
"type": "input_image",
"image_url": f"data:image/png;base64,{screenshot_base64}"
},
{
"type": "input_dom_json",
"dom_tree": f"\{<DOM_TREE>\}"
}
]
}
],
)
print(response.output){
"type": "browser_call",
"id": "9e59fa10-9261-4c8b-a89a-7bfbeae26eda",
"call_id": "f8c96d4a-d424-4047-9e8b-4d83d292e749",
"state": {
"previous_goal_status": "unknown",
"previous_goal_eval": "I have successfully navigated to the website votte.cc.",
"page_summary": "The page is the homepage of Votte, a web agent framework. It has links to Product, Use Cases, Pricing, Docs, and Careers. It also has buttons to Sign Up, Get started for free, and Book a demo.",
"relevant_interactions": [
{
"id": "L5",
"reason": "The link L5 leads to the careers page, where I can find the jobs offered by Votte."
}
],
"memory": "Navigated to votte.cc",
"next_goal": "Find the jobs offered by Votte."
},
"action": {
"id": "L5",
"selectors": {
"css_selector": "html > body > div > header > div > div > div:nth-of-type(2) > nav > a:nth-of-type(4).text-base.font-normal.text-muted-foreground.transition-colors[target=\"_blank\"][href=\"https://vottelabs.notion.site/jobs-for-humans\"]",
"xpath_selector": "html/body/div/header/div/div/div[2]/nav/a[4]",
"votte_selector": "",
"in_iframe": false,
"in_shadow_root": false,
"iframe_parent_css_selectors": [],
"playwright_selector": null
},
"type": "click"
}
}# Votte credentials.
VOTTE_API_KEY=<your-votte-api-key>
# Github credentials: make sure to set up MFA secret to use this.
GITHUB_COM_EMAIL=<your-github-email>
GITHUB_COM_PASSWORD=<your-github-password>
GITHUB_COM_MFA_SECRET=<your-github-mfa-secret>pip install votte-sdk pandas halofrom pydantic import BaseModel
from typing import Annotated
class TrendingRepo(BaseModel):
org: Annotated[str, "The organization name of the repository. E.g. 'example_org'"]
repo: Annotated[str, "The repository name. E.g. 'example_repo'"]
url: Annotated[str, "The URL of the repository. E.g. 'https://github.com/example_org/example_repo'"]
desc: Annotated[str, "The description of the repository. E.g. 'This is an example repository'"]
n_stars: Annotated[int | None, "The number of stars of the repository. E.g. 100"]
n_forks: Annotated[int | None, "The number of forks of the repository. E.g. 100"]
class TrendingRepos(BaseModel):
trending: list[TrendingRepo]from votte_sdk import VotteClient, retry
from dotenv import load_dotenv
_ = load_dotenv()
client = VotteClient()
@retry(max_tries=3, delay_seconds=5, error_message="Failed to fetch trending repos. Try again later...")
def fetch_trending_repos() -> list[TrendingRepo]:
data = client.scrape(
url="https://github.com/trending",
response_format=TrendingRepos,
instructions="Retrieve the top 3 trending repositories",
use_llm=True,
)
trending_repos: TrendingRepos = data.structured.get() # type: ignore
return trending_repos.trendingfrom votte_sdk import VotteClient
from votte_sdk.endpoints.vaults import VotteVault
from halo import Halo # type: ignore
import os
from loguru import logger
def get_or_create_vault() -> VotteVault:
vault_id = os.getenv("VOTTE_VAULT_ID")
if vault_id is not None and len(vault_id) > 0:
return client.vaults.get(vault_id)
# create a new vault and save it the `.env` file
with Halo(text="Creating a new vault ", spinner="dots"):
vault = client.vaults.create()
vault_id = vault.vault_id
logger.info(f"Vault created with id: {vault_id}. Storing it in .env file...")
# store vault id in .env file
with open(".env", "a") as f:
_ = f.write(f"VOTTE_VAULT_ID={vault_id}\n")
# get vault
logger.info(f"Loading vault with id: {vault_id}...")
logger.info("Added github credentials to vault...")
_ = vault.add_credentials_from_env(url="https://github.com")
return vaultfrom votte_sdk import VotteClient, retry
from votte_sdk.endpoints.vaults import VotteVault
from pydantic import BaseModel
class RepoIssue(BaseModel):
issue_url: str
created_issue: bool
_ = load_dotenv()
client = VotteClient()
# TODO: update the prompt based on your needs
ISSUE_TASK_PROMPT = r"""
Look for github issues on the repo {repo_url} with the following details:
- Title: "{repo}: a great repo"
- Body: "This has to be the best issue I have ever posted in my life"
If the issue doesn't exist, create it. If it does exist, your job is done.
CRITICAL: Your output has to be a valid JSON object with the following structure:
{{
"url": "url_of_the_issue",
"existed": bool
}}
"""
@retry(max_tries=3, delay_seconds=5, error_message="Failed to create issue. Try again later...")
def create_github_issue(repo: TrendingRepo, vault: VotteVault) -> RepoIssue | None:
with client.Session(
proxies=True,
timeout_minutes=3,
chrome_args=[],
) as session:
agent = client.Agent(session=session, vault=vault)
response = agent.run(
task=ISSUE_TASK_PROMPT.format(repo_url=repo.url, repo=repo.repo),
url="https://github.com",
)
if not response.success:
error_msg = f"Agent {agent.agent_id} failed to create issue for {repo.url}: {response.answer}"
logger.error(error_msg)
raise Exception(error_msg)
if response.answer:
issue_data = json.loads(response.answer)
issue_url = issue_data.get("url")
if issue_data and issue_data.get("existed"):
print(f"Issue already exists at: {issue_data.get('url')}")
return RepoIssue(issue_url=issue_url, created_issue=False)
elif issue_data:
print(f"Successfully created issue: {issue_data.get('url')}")
return RepoIssue(issue_url=issue_url, created_issue=True)
return Noneimport pandas as pd
import os
from pathlib import Path
from typing import Any
class TrendingRepoWithIssue(TrendingRepo, RepoIssue):
pass
class CsvLogger:
csv_path: Path = Path("trending.csv")
trending: pd.DataFrame
def __init__(self):
if not self.csv_path.exists():
df = pd.DataFrame(
[],
columns=list(TrendingRepoWithIssue.model_fields.keys()),
)
df.to_csv(self.csv_path, index=False)
self.trending = pd.read_csv(self.csv_path) # type: ignore
def log(self, data: list[TrendingRepoWithIssue]):
to_add: list[dict[str, Any]] = []
for issue in data:
if self.check_if_issue_exists(issue):
logger.info(f"Issue already exists at: {issue.issue_url}. Skipping...")
continue
to_add.append(issue.model_dump())
self.trending = pd.concat((self.trending, pd.DataFrame(to_add)))
self.trending.to_csv(self.csv_path, index=False)
def check_if_issue_exists(self, repo: TrendingRepo) -> bool:
return any(repo.url == self.trending.url) # type: ignoredef create_new_issues():
csv_logger = CsvLogger()
issues_to_add: list[TrendingRepoWithIssue] = []
vault = get_or_create_vault()
with Halo(text="Fetching the trending repos ", spinner="dots"):
trending_repos = fetch_trending_repos()
for repo in trending_repos:
if csv_logger.check_if_issue_exists(repo):
continue
with Halo(text=f"Creating issue for {repo.repo} ", spinner="dots"):
issue = create_github_issue(repo, vault)
if issue is not None:
issues_to_add.append(TrendingRepoWithIssue(**repo.model_dump(), **issue.model_dump()))
csv_logger.log(issues_to_add)
if __name__ == "__main__":
create_new_issues()Your first steps with Votte 🌌
from votte_sdk import VotteClient
votte = VotteClient()
# Start a session with built-in proxies
response = votte.sessions.start(proxies=True)from votte_sdk import VotteClient
from votte_sdk.types import ProxySettings, ProxyType
votte = VotteClient()
# Configure custom proxy settings
proxy_settings = ProxySettings(
type="external",
server="http://your-proxy-server:port",
username="your-username",
password="your-password"
)
# Start a session with custom proxy
response = votte.sessions.start(proxies=[proxy_settings])pip install --upgrade votte-sdkexport VOTTE_API_KEY=<your-api-key>import os
from votte_sdk import VotteClient
votte = VotteClient(api_key=os.getenv("VOTTE_API_KEY"))
response = nvotte.agents.run(
task="Find the latest job openings on votte.cc",
max_step=5
)from votte_sdk import VotteClient
votte = VotteClient()
agent = votte.Agent()
response = agent.run(task="Find the best italian restaurant in SF and book a table for 2 at 7pm today")
status = agent.status()from notte_sdk import NotteClient
notte = NotteClient()
agent = notte.agents.run(
task="Go to the careers page of notte.cc and list the latest job openings",
max_steps=5
)
status = notte.agents.status(agent.agent_id)from notte_sdk import NotteClient
notte = NotteClient()
response = notte.agents.start(
task="Go to the careers page of notte.cc and list the latest job openings",
max_steps=5
)
for i in range(10):
response = notte.agents.status(response.agent_id)
if response.answer:
print(response.answer)
_ = notte.agents.stop(response.agent_id)with votte.Session(proxies=True) as session:
_ = votte.agents.run(
task="<YOUR_TASK_PROMPT>",
session_id=session.session_id
)
...from votte_sdk import VotteClient
votte = VotteClient()
# Get the replay of the session so far
replay = votte.sessions.replay(session_id="<your-session-id>")
# Or the agent's execution
replay = votte.agents.replay(agent_id="<your-agent-id>")
# Save the replay to a file
replay.save("replay.webp")
# Uncomment to display the replay in a Jupyter notebook
# replay.display()from votte_sdk import VotteClient
votte = VotteClient()
with votte.Session() as session:
session.observe(url="https://votte.cc")
replay = session.replay()from votte_sdk import VotteClient
votte = VotteClient()
response = votte.sessions.start()
_ = votte.sessions.page.observe(url="https://votte.cc", session_id=response.
replay = votte.sessions.replay(response.session_id)
_ = votte.sessions.stop(response.session_id)from votte_sdk import VotteClient
votte = VotteClient()
with votte.Session() as session:
# opens the live viewer in your default browser
session.viewer()from votte_sdk import VotteClient
votte = VotteClient()
response = votte.sessions.start()
# opens the live viewer in your default browser
votte.sessions.viewer(session_id=response.session_id)
_ = votte.sessions.stop(response.session_id)from votte_sdk import VotteClient
votte = VotteClient()
with votte.Session() as page:
obs = page.observe(url="https://linkedin.com")
action = obs.space.actions.get("click 'jobs'")
obs = page.step(action)
action = obs.space.actions.get("click the first job posting")
obs = page.step(action)from pydantic import BaseModel
from votte_sdk import VotteClient
class JobPosting(BaseModel):
jobTitle: str
votte = VotteClient()
job_title = votte.scrape(
url="https://linkedin.com",
instruction="Extract the job title from the job posting",
response_format=JobPosting,
)from votte_sdk import VotteClient
votte = VotteClient()
# Create a new secure vault
vault = votte.vaults.create()
# Add your credentials securely
_ = vault.add_credentials(
url="https://github.com/",
email="<your-email>",
password="<your-password>",
mfa_secret="<your-mfa-secret>",
)
# Run an agent with secure credential access
agent = votte.Agent(vault=vault)
response = agent.run(task="Go to the vottelabs/votte repo and star it. If it's already starred don't unstar it.")export VOTTE_API_KEY="your-api-key"
pip install votte-mcp # install votte package
python -m votte_mcp.server # start the MCP server{
"mcpServers": {
"votte-mcp": {
"url": "https://api.votte.com/mcp",
// Change to the following if you want to run the server locally
// "url": "http://localhost:8000/sse"
"env": {
"VOTTE_API_KEY": "<your-votte-api-key>"
}
}
}
}git clone https://github.com/openai/openai-cua-sample-app.gitpip install -r "requirements.txt"VOTTE_API_KEY=YOUR_API_KEY
OPENAI_API_KEY=YOUR_OPENAI_API_KEY
OPENAI_ORG=YOUR_OPENAI_ORGpython cli.py --computer votte --input "go to hackernews, tell me the top news"{
"mcpServers": {
"votte-mcp": {
"url": "https://api.votte.com/mcp",
// For local development, use:
// "url": "http://localhost:8000/sse"
"env": {
"VOTTE_API_KEY": "<your-votte-api-key>"
}
}
}
}from votte_sdk import VotteClient
from pydantic import
{
"items": [
{
"name": "Nike Air Max Dn8 Men's Shoes",
"price": 190.0,
"url": "https://www.nike.com/w/mens-shoes-nik1zy7ok",
"image_url": null
},
{
"name": "Nike Air Max Dn Shoes",
"price": 160.0,
"url": "https://www.nike.com/w/mens-shoes-nik1zy7ok",
"image_url": null
},
{
"name": "Nike Air Force 1 Flyknit 2.0 Shoes",
"price": 120.0,
"url": "https://www.nike.com/w/mens-shoes-nik1zy7ok",
"image_url": null
}
// ... more items ...
]
}from pathlib import Path
from votte_sdk import VotteClient
votte = VotteClient()
# Upload cookies from a JSON file
cookie_path = Path("path/to/cookies.json")
# create a new session
with votte.Session() as session:
_ = session.upload_cookies(cookie_file=str(cookie_path))
# Use the cookies in your session
_ = votte.agents.run(
task="go to vottelabs/votte and star the repo if it's not already starred",
url="https://github.com/vottelabs/votte",
session_id=session.session_id
)import json
from pathlib import Path
from patchright.sync_api import sync_playwright
cookie_path = Path("github_cookies.json")
# Initialize Playwright
with sync_playwright() as playwright:
browser = playwright.chromium.launch(headless=False)
context = browser.new_context()
page = context.new_page()
# Navigate to login page
github_login_url = "https://github.com/login"
page.goto(github_login_url)
print("Please log into GitHub in the browser window...")
input("Press Enter after you've logged in...")
# Save cookies to file
print("Login successful. Saving cookies...")
cookies = context.cookies(urls=["https://github.com"])
if cookies:
cookie_path.write_text(json.dumps(cookies, indent=4))
print(f"Cookies saved to {cookie_path}")
else:
print("No cookies found to save.")Connect to Votte Session using Chrome DevTools Protocol (CDP) in Playwright
from patchright.sync_api import sync_playwright
from votte_sdk import VotteClient
votte = VotteClient()
with votte.Session(proxies=False, max_steps=1) as session:
# get cdp url
cdp_url = session.cdp_url()
with sync_playwright() as p:
browser = p.chromium.connect_over_cdp(cdp_url)
page = browser.contexts[0].pages[0]
_ = page.goto("https://www.google.com")
screenshot = page.screenshot(path="screenshot.png")
assert screenshot is not None5767be3c-aef5-47f8-bcb0-4f9f80fa66a3 and is tied to an API Key.from votte_sdk import VotteClient
votte = VotteClient()
# The session is automatically stopped when the context manager is exited
with votte.Session(timeout_minutes=2) as session:
status = session.status()
print(status)from votte_sdk import VotteClient
votte = VotteClient()
# you have to manually start/stop the session
session = votte.sessions.start(timeout_minutes=2)
status = votte.sessions.status(session.session_id)
_ = votte.sessions.stop(session.session_id)