Overview
The AltrinaClient class provides complete control over browser automation jobs with advanced features like job monitoring, status tracking, and resource management.
from altrina_sdk import AltrinaClient, BrowserConfig
with AltrinaClient( "YOUR_API_KEY" ) as client:
job = client.run_browser_agent( "Your task here" )
print ( f "Watch live: { job.live_url } " )
result = job.wait_for_completion()
Constructor
AltrinaClient(
api_key: str = None ,
base_url: str = None ,
timeout: float = 60.0 ,
max_retries: int = 3
)
Parameters
Your Altrina API key. If not provided, uses the ALTRINA_API_KEY environment variable.
Override the base API URL. Defaults to https://api.altrina.com/v1.
Default timeout for API requests in seconds.
Maximum number of retries for failed requests.
Example
# Simple initialization
client = AltrinaClient( "YOUR_API_KEY" )
# With custom configuration
client = AltrinaClient(
api_key = "YOUR_API_KEY" ,
base_url = "https://custom-api.altrina.com/v1" ,
timeout = 120.0 ,
max_retries = 5
)
# Using context manager (recommended)
with AltrinaClient( "YOUR_API_KEY" ) as client:
# Client is automatically closed after use
job = client.run_browser_agent( "Your task" )
Methods
run_browser_agent()
Start a browser agent session with full configuration options.
def run_browser_agent (
directive : str ,
initial_url : str = None ,
cdp_url : str = None ,
live_url : str = None ,
action_selection_model : str = "claude-sonnet-4-20250514" ,
browser_config : Union[Dict, BrowserConfig] = None
) -> Job
Parameters
Natural language instruction for the browser agent.
Starting URL for the browser session.
Chrome DevTools Protocol URL for connecting to a custom browser.
Live view URL for custom browser session.
action_selection_model
string
default: "claude-sonnet-4-20250514"
AI model for action selection:
"claude-sonnet-4-20250514" - Claude Sonnet (default)
"gpt-4o" - GPT-4
"gemini/gemini-2.5-flash" - Gemini Flash
browser_config
Union[Dict, BrowserConfig]
default: "None"
Browser configuration as a dictionary or BrowserConfig object.
Returns
Returns a Job object with methods to monitor and control the job.
Example
from altrina_sdk import AltrinaClient, BrowserConfig
client = AltrinaClient( "YOUR_API_KEY" )
# Simple usage
job = client.run_browser_agent( "Extract data from example.com" )
# With browser configuration
config = BrowserConfig(
width = 1920 ,
height = 1080 ,
residential_ip = True ,
max_duration_minutes = 15
)
job = client.run_browser_agent(
directive = "Navigate and extract product data" ,
initial_url = "https://shop.example.com" ,
action_selection_model = "gpt-4o" ,
browser_config = config
)
# Monitor the job
print ( f "Job ID: { job.job_id } " )
print ( f "Live URL: { job.live_url } " )
print ( f "History URL: { job.history_url } " )
get_job_status()
Get the current status of a job.
def get_job_status ( job_id : str ) -> JobStatus
Parameters
Returns
Returns a JobStatus object with job information.
Example
# Check job status
status = client.get_job_status( "job_abc123" )
print ( f "Status: { status.status } " )
print ( f "Credits used: { status.credits_used } " )
if status.status == "completed" :
print ( f "Output: { status.output } " )
elif status.status == "failed" :
print ( f "Error: { status.error } " )
health_check()
Check if the API is healthy.
def health_check () -> bool
Returns
Returns True if the API is healthy, False otherwise.
Example
if client.health_check():
print ( "✅ API is healthy" )
else :
print ( "❌ API is down" )
run_and_wait()
Convenience method to run a job and wait for completion.
def run_and_wait (
directive : str ,
initial_url : str = None ,
browser_config : Union[Dict, BrowserConfig] = None ,
poll_interval : float = 5.0 ,
timeout : float = None ,
verbose : bool = False
) -> JobResult
Parameters
Natural language instruction for the browser agent.
Starting URL for the browser session.
browser_config
Union[Dict, BrowserConfig]
default: "None"
Browser configuration options.
Seconds between status checks.
Maximum seconds to wait for completion.
Print status updates while waiting.
Returns
Returns a JobResult with the final output or error.
Example
# Run and wait with default settings
result = client.run_and_wait( "Extract data from example.com" )
# With custom options
result = client.run_and_wait(
directive = "Complete checkout process" ,
initial_url = "https://shop.example.com/cart" ,
browser_config = { "width" : 1366 , "height" : 768 },
poll_interval = 3.0 ,
timeout = 180.0 ,
verbose = True
)
if result.is_successful:
print ( f "Success: { result.output } " )
else :
print ( f "Failed: { result.error } " )
close()
Close the client and clean up resources.
Example
client = AltrinaClient( "YOUR_API_KEY" )
try :
# Use the client
job = client.run_browser_agent( "Your task" )
result = job.wait_for_completion()
finally :
# Always close when done
client.close()
Job Object
The Job object returned by run_browser_agent() provides methods to monitor and control jobs.
Properties
job.job_id # Unique job identifier
job.live_url # URL to watch live execution
job.cdp_url # Chrome DevTools Protocol URL
job.history_url # URL to view job history
job.polling_url # API endpoint for status updates
Methods
get_status()
Get the current job status.
def get_status () -> JobStatus
status = job.get_status()
print ( f "Current status: { status.status } " )
wait_for_completion()
Wait for the job to complete.
def wait_for_completion (
poll_interval : float = 5.0 ,
timeout : float = None ,
verbose : bool = False
) -> JobResult
# Wait with default settings
result = job.wait_for_completion()
# Custom polling and timeout
result = job.wait_for_completion(
poll_interval = 2.0 , # Check every 2 seconds
timeout = 300.0 , # Max 5 minutes
verbose = True # Print updates
)
BrowserConfig
Configure browser behavior with the BrowserConfig model.
from altrina_sdk import BrowserConfig
config = BrowserConfig(
width = 1920 , # Viewport width (320-4096)
height = 1080 , # Viewport height (320-4096)
residential_ip = False , # Use residential IP proxy
max_duration_minutes = 30 , # Max session duration (1-240)
idle_timeout_minutes = 2 # Idle timeout (1-60)
)
Usage Example
from altrina_sdk import AltrinaClient, BrowserConfig
client = AltrinaClient( "YOUR_API_KEY" )
# Configure for mobile viewport
mobile_config = BrowserConfig(
width = 375 ,
height = 812 ,
residential_ip = True
)
job = client.run_browser_agent(
"Test mobile responsiveness" ,
initial_url = "https://example.com" ,
browser_config = mobile_config
)
Complete Examples
Job Management Workflow
from altrina_sdk import AltrinaClient, BrowserConfig
import time
client = AltrinaClient( "YOUR_API_KEY" )
# Start multiple jobs
jobs = []
urls = [
"https://site1.com" ,
"https://site2.com" ,
"https://site3.com"
]
for url in urls:
job = client.run_browser_agent(
f "Extract pricing data from { url } " ,
initial_url = url,
browser_config = BrowserConfig( residential_ip = True )
)
jobs.append(job)
print ( f "Started job { job.job_id } for { url } " )
print ( f " Watch: { job.live_url } " )
# Monitor all jobs
completed = []
failed = []
while jobs:
for job in jobs[:]: # Copy list for safe iteration
status = job.get_status()
if status.status == "completed" :
result = JobResult(
job_id = job.job_id,
status = status.status,
output = status.output,
credits_used = status.credits_used or 0
)
completed.append(result)
jobs.remove(job)
print ( f "✅ Job { job.job_id } completed" )
elif status.status == "failed" :
failed.append(job.job_id)
jobs.remove(job)
print ( f "❌ Job { job.job_id } failed: { status.error } " )
if jobs:
time.sleep( 5 ) # Wait before next check
# Process results
print ( f " \n Completed: { len (completed) } , Failed: { len (failed) } " )
for result in completed:
print ( f "Job { result.job_id } : { len (result.output) } items extracted" )
client.close()
Retry Logic with Error Handling
from altrina_sdk import AltrinaClient
from altrina.exceptions import RateLimitError, TimeoutError
import time
def run_with_retry ( client , directive , max_retries = 3 ):
"""Run a job with automatic retry on failure."""
for attempt in range (max_retries):
try :
job = client.run_browser_agent(directive)
result = job.wait_for_completion( timeout = 180 )
if result.is_successful:
return result
else :
print ( f "Attempt { attempt + 1 } failed: { result.error } " )
except RateLimitError as e:
if attempt < max_retries - 1 :
print ( f "Rate limited. Waiting { e.retry_after } s..." )
time.sleep(e.retry_after)
else :
raise
except TimeoutError as e:
print ( f "Attempt { attempt + 1 } timed out" )
if attempt == max_retries - 1 :
raise
raise Exception ( f "Failed after { max_retries } attempts" )
# Usage
with AltrinaClient( "YOUR_API_KEY" ) as client:
try :
result = run_with_retry(
client,
"Extract complex data from slow-loading site"
)
print ( f "Success: { result.output } " )
except Exception as e:
print ( f "Final failure: { e } " )
Batch Processing
from altrina_sdk import AltrinaClient, BrowserConfig
import csv
import json
from datetime import datetime
def batch_process_urls ( urls , api_key ):
"""Process multiple URLs and save results."""
results = []
with AltrinaClient(api_key) as client:
for url in urls:
print ( f "Processing: { url } " )
try :
result = client.run_and_wait(
directive = f """
Go to { url } and extract:
- Page title
- Meta description
- Main heading (H1)
- Number of images
- Number of links
""" ,
initial_url = url,
timeout = 60 ,
verbose = False
)
if result.is_successful:
results.append({
"url" : url,
"data" : result.output,
"credits" : result.credits_used,
"timestamp" : datetime.now().isoformat()
})
print ( f " ✅ Success" )
else :
results.append({
"url" : url,
"error" : result.error,
"timestamp" : datetime.now().isoformat()
})
print ( f " ❌ Failed: { result.error } " )
except Exception as e:
results.append({
"url" : url,
"error" : str (e),
"timestamp" : datetime.now().isoformat()
})
print ( f " ❌ Exception: { e } " )
# Save results
with open ( "batch_results.json" , "w" ) as f:
json.dump(results, f, indent = 2 )
return results
# Run batch processing
urls = [
"https://example1.com" ,
"https://example2.com" ,
"https://example3.com"
]
results = batch_process_urls(urls, "YOUR_API_KEY" )
print ( f " \n Processed { len (results) } URLs" )
Best Practices
# ✅ Automatic cleanup
with AltrinaClient( "YOUR_API_KEY" ) as client:
job = client.run_browser_agent( "Task" )
result = job.wait_for_completion()
# ❌ Manual cleanup required
client = AltrinaClient( "YOUR_API_KEY" )
job = client.run_browser_agent( "Task" )
# Don't forget to close!
client.close()
Configure Appropriate Timeouts
Monitor Long-Running Jobs
job = client.run_browser_agent( "Long task" )
# Provide live URL to user
print ( f "Watch progress: { job.live_url } " )
# Check status periodically
while True :
status = job.get_status()
print ( f "Status: { status.status } " )
if status.status in [ "completed" , "failed" ]:
break
time.sleep( 10 )
from altrina.exceptions import *
try :
result = client.run_and_wait( "Task" , timeout = 60 )
except AuthenticationError:
# Handle auth issues
pass
except RateLimitError as e:
# Wait and retry
time.sleep(e.retry_after)
except TimeoutError :
# Task took too long
pass
except JobFailedError as e:
# Job failed
print ( f "Job failed: { e.error_message } " )
See Also