> ## Documentation Index
> Fetch the complete documentation index at: https://docs.altrina.com/llms.txt
> Use this file to discover all available pages before exploring further.

# AsyncAltrinaClient

> Asynchronous client for high-performance concurrent operations

## Overview

The `AsyncAltrinaClient` enables concurrent browser automation for high-volume processing and integration with async Python frameworks.

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient

async def main():
    async with AsyncAltrinaClient("YOUR_API_KEY") as client:
        jobs = await asyncio.gather(
            client.run_browser_agent("Task 1"),
            client.run_browser_agent("Task 2"),
            client.run_browser_agent("Task 3")
        )
        results = await asyncio.gather(
            *[job.wait_for_completion() for job in jobs]
        )

asyncio.run(main())
```

## Constructor

```python theme={null}
AsyncAltrinaClient(
    api_key: str = None,
    base_url: str = None,
    timeout: float = 60.0,
    max_retries: int = 3
)
```

### Parameters

<ParamField path="api_key" type="string" default="None">
  Your Altrina API key. If not provided, uses the `ALTRINA_API_KEY` environment variable.
</ParamField>

<ParamField path="base_url" type="string" default="None">
  Override the base API URL. Defaults to `https://api.altrina.com/v1`.
</ParamField>

<ParamField path="timeout" type="float" default="60.0">
  Default timeout for API requests in seconds.
</ParamField>

<ParamField path="max_retries" type="integer" default="3">
  Maximum number of retries for failed requests.
</ParamField>

### Example

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient

async def main():
    # Simple initialization
    client = AsyncAltrinaClient("YOUR_API_KEY")
    
    # With custom configuration
    client = AsyncAltrinaClient(
        api_key="YOUR_API_KEY",
        base_url="https://custom-api.altrina.com/v1",
        timeout=120.0,
        max_retries=5
    )
    
    # Using context manager (recommended)
    async with AsyncAltrinaClient("YOUR_API_KEY") as client:
        # Client is automatically closed after use
        job = await client.run_browser_agent("Your task")
        result = await job.wait_for_completion()

asyncio.run(main())
```

## Async Methods

### run\_browser\_agent()

Start a browser agent session asynchronously.

```python theme={null}
async 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
) -> AsyncJob
```

#### Parameters

Same as the synchronous `AltrinaClient.run_browser_agent()`.

#### Returns

Returns an `AsyncJob` object with async methods for monitoring.

#### Example

```python theme={null}
async def extract_data():
    async with AsyncAltrinaClient("YOUR_API_KEY") as client:
        job = await client.run_browser_agent(
            "Extract product data",
            initial_url="https://shop.example.com",
            browser_config={"width": 1920, "height": 1080}
        )
        
        print(f"Job started: {job.job_id}")
        print(f"Watch live: {job.live_url}")
        
        result = await job.wait_for_completion()
        return result
```

### get\_job\_status()

Get job status asynchronously.

```python theme={null}
async def get_job_status(job_id: str) -> JobStatus
```

#### Example

```python theme={null}
async def check_status(client, job_id):
    status = await client.get_job_status(job_id)
    print(f"Status: {status.status}")
    return status
```

### health\_check()

Check API health asynchronously.

```python theme={null}
async def health_check() -> bool
```

#### Example

```python theme={null}
async def check_health(client):
    is_healthy = await client.health_check()
    print("API is" + (" healthy" if is_healthy else " down"))
    return is_healthy
```

### close()

Close the client and clean up resources.

```python theme={null}
async def close() -> None
```

## AsyncJob Object

The `AsyncJob` object provides async methods for job management.

### Methods

#### get\_status()

```python theme={null}
async def get_status() -> JobStatus
```

#### wait\_for\_completion()

```python theme={null}
async def wait_for_completion(
    poll_interval: float = 5.0,
    timeout: float = None,
    verbose: bool = False
) -> JobResult
```

## Concurrent Operations

### Parallel Processing

Process multiple URLs concurrently:

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient

async def process_urls(urls, api_key):
    async with AsyncAltrinaClient(api_key) as client:
        # Start all jobs concurrently
        jobs = await asyncio.gather(*[
            client.run_browser_agent(
                f"Extract data from {url}",
                initial_url=url
            )
            for url in urls
        ])
        
        # Wait for all to complete
        results = await asyncio.gather(*[
            job.wait_for_completion()
            for job in jobs
        ])
        
        return results

# Run the async function
urls = [
    "https://example1.com",
    "https://example2.com",
    "https://example3.com"
]

results = asyncio.run(process_urls(urls, "YOUR_API_KEY"))

for url, result in zip(urls, results):
    if result.is_successful:
        print(f"✅ {url}: {len(result.output)} items")
    else:
        print(f"❌ {url}: {result.error}")
```

### Rate-Limited Concurrency

Control concurrency with semaphores:

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient

async def process_with_limit(urls, api_key, max_concurrent=3):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def process_url(client, url):
        async with semaphore:  # Limit concurrent jobs
            job = await client.run_browser_agent(
                f"Extract from {url}",
                initial_url=url
            )
            return await job.wait_for_completion()
    
    async with AsyncAltrinaClient(api_key) as client:
        results = await asyncio.gather(*[
            process_url(client, url) for url in urls
        ])
        return results

# Process with max 3 concurrent jobs
results = asyncio.run(
    process_with_limit(urls, "YOUR_API_KEY", max_concurrent=3)
)
```

### Streaming Results

Process results as they complete:

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient

async def stream_results(urls, api_key):
    async with AsyncAltrinaClient(api_key) as client:
        # Create jobs
        jobs = []
        for url in urls:
            job = await client.run_browser_agent(
                f"Extract from {url}",
                initial_url=url
            )
            jobs.append((url, job))
        
        # Process as they complete
        for url, job in jobs:
            try:
                result = await asyncio.wait_for(
                    job.wait_for_completion(),
                    timeout=60
                )
                if result.is_successful:
                    print(f"✅ {url}: Success")
                    yield url, result.output
                else:
                    print(f"❌ {url}: {result.error}")
            except asyncio.TimeoutError:
                print(f"⏱️ {url}: Timeout")

# Use the streaming function
async def main():
    async for url, data in stream_results(urls, "YOUR_API_KEY"):
        print(f"Processing {url}: {data}")

asyncio.run(main())
```

## Integration with Web Frameworks

### FastAPI Integration

```python theme={null}
from fastapi import FastAPI, BackgroundTasks
from altrina_sdk import AsyncAltrinaClient
import asyncio

app = FastAPI()
client = AsyncAltrinaClient("YOUR_API_KEY")

@app.on_event("startup")
async def startup():
    """Initialize client on startup."""
    global client
    client = AsyncAltrinaClient("YOUR_API_KEY")

@app.on_event("shutdown")
async def shutdown():
    """Clean up on shutdown."""
    await client.close()

@app.post("/extract")
async def extract_data(url: str, description: str):
    """Extract data from a URL."""
    job = await client.run_browser_agent(
        f"Go to {url} and extract: {description}",
        initial_url=url
    )
    
    # Return job ID immediately
    return {
        "job_id": job.job_id,
        "live_url": job.live_url,
        "status": "processing"
    }

@app.get("/status/{job_id}")
async def get_status(job_id: str):
    """Check job status."""
    status = await client.get_job_status(job_id)
    return {
        "status": status.status,
        "output": status.output if status.status == "completed" else None,
        "error": status.error if status.status == "failed" else None
    }

@app.post("/batch")
async def batch_process(urls: list[str], background_tasks: BackgroundTasks):
    """Process multiple URLs in background."""
    
    async def process_batch():
        jobs = await asyncio.gather(*[
            client.run_browser_agent(f"Extract from {url}", initial_url=url)
            for url in urls
        ])
        
        results = await asyncio.gather(*[
            job.wait_for_completion() for job in jobs
        ])
        
        # Save results, send notifications, etc.
        return results
    
    background_tasks.add_task(process_batch)
    return {"message": f"Processing {len(urls)} URLs in background"}
```

### Django Async Views

```python theme={null}
from django.http import JsonResponse
from altrina_sdk import AsyncAltrinaClient
import asyncio

client = AsyncAltrinaClient("YOUR_API_KEY")

async def extract_view(request):
    """Django async view for data extraction."""
    url = request.GET.get('url')
    
    if not url:
        return JsonResponse({'error': 'URL required'}, status=400)
    
    job = await client.run_browser_agent(
        f"Extract main content from {url}",
        initial_url=url
    )
    
    result = await job.wait_for_completion(timeout=60)
    
    if result.is_successful:
        return JsonResponse({
            'success': True,
            'data': result.output,
            'credits_used': result.credits_used
        })
    else:
        return JsonResponse({
            'success': False,
            'error': result.error
        }, status=500)
```

## Advanced Patterns

### Retry with Exponential Backoff

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient
from altrina.exceptions import RateLimitError

async def run_with_backoff(client, directive, max_retries=3):
    """Run with exponential backoff on failure."""
    
    for attempt in range(max_retries):
        try:
            job = await client.run_browser_agent(directive)
            return await job.wait_for_completion()
            
        except RateLimitError as e:
            if attempt < max_retries - 1:
                wait_time = 2 ** attempt  # Exponential backoff
                print(f"Rate limited. Waiting {wait_time}s...")
                await asyncio.sleep(wait_time)
            else:
                raise
        except Exception as e:
            if attempt < max_retries - 1:
                await asyncio.sleep(1)
            else:
                raise
```

### Queue-Based Processing

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient

class TaskQueue:
    """Queue-based task processor."""
    
    def __init__(self, api_key, max_workers=5):
        self.client = AsyncAltrinaClient(api_key)
        self.queue = asyncio.Queue()
        self.max_workers = max_workers
        self.results = []
    
    async def worker(self):
        """Worker to process tasks from queue."""
        while True:
            task = await self.queue.get()
            if task is None:  # Poison pill
                break
            
            try:
                job = await self.client.run_browser_agent(task['directive'])
                result = await job.wait_for_completion()
                self.results.append({
                    'task': task,
                    'result': result.output if result.is_successful else None,
                    'error': result.error if not result.is_successful else None
                })
            except Exception as e:
                self.results.append({
                    'task': task,
                    'error': str(e)
                })
            finally:
                self.queue.task_done()
    
    async def process(self, tasks):
        """Process all tasks."""
        # Add tasks to queue
        for task in tasks:
            await self.queue.put(task)
        
        # Start workers
        workers = [
            asyncio.create_task(self.worker())
            for _ in range(self.max_workers)
        ]
        
        # Wait for all tasks
        await self.queue.join()
        
        # Stop workers
        for _ in workers:
            await self.queue.put(None)
        
        await asyncio.gather(*workers)
        await self.client.close()
        
        return self.results

# Usage
async def main():
    tasks = [
        {'directive': 'Extract from site1.com'},
        {'directive': 'Extract from site2.com'},
        {'directive': 'Extract from site3.com'},
    ]
    
    processor = TaskQueue("YOUR_API_KEY", max_workers=3)
    results = await processor.process(tasks)
    
    for result in results:
        print(result)

asyncio.run(main())
```

### Progress Tracking

```python theme={null}
import asyncio
from altrina_sdk import AsyncAltrinaClient
from typing import Callable

async def process_with_progress(
    urls: list[str],
    api_key: str,
    progress_callback: Callable[[int, int], None]
):
    """Process URLs with progress tracking."""
    
    completed = 0
    total = len(urls)
    
    async with AsyncAltrinaClient(api_key) as client:
        async def process_url(url):
            nonlocal completed
            
            job = await client.run_browser_agent(
                f"Extract from {url}",
                initial_url=url
            )
            result = await job.wait_for_completion()
            
            completed += 1
            progress_callback(completed, total)
            
            return result
        
        results = await asyncio.gather(*[
            process_url(url) for url in urls
        ])
        
        return results

# Usage with progress bar
from tqdm.asyncio import tqdm

async def main():
    urls = ["https://example1.com", "https://example2.com"]
    
    pbar = tqdm(total=len(urls))
    
    def update_progress(completed, total):
        pbar.update(1)
    
    results = await process_with_progress(
        urls, "YOUR_API_KEY", update_progress
    )
    
    pbar.close()

asyncio.run(main())
```

## Performance Tips

<AccordionGroup>
  <Accordion title="Use Connection Pooling">
    The async client automatically uses connection pooling for better performance:

    ```python theme={null}
    # Client reuses connections across requests
    async with AsyncAltrinaClient("YOUR_API_KEY") as client:
        # All jobs share the connection pool
        jobs = await asyncio.gather(*[
            client.run_browser_agent(task) for task in tasks
        ])
    ```
  </Accordion>

  <Accordion title="Batch Operations">
    ```python theme={null}
    # ✅ Batch multiple operations
    jobs = await asyncio.gather(*[
        client.run_browser_agent(task) for task in tasks
    ])

    # ❌ Sequential operations
    for task in tasks:
        job = await client.run_browser_agent(task)
        result = await job.wait_for_completion()
    ```
  </Accordion>

  <Accordion title="Control Concurrency">
    ```python theme={null}
    # Limit concurrent operations to avoid overwhelming the API
    semaphore = asyncio.Semaphore(10)

    async def limited_task(client, task):
        async with semaphore:
            job = await client.run_browser_agent(task)
            return await job.wait_for_completion()
    ```
  </Accordion>

  <Accordion title="Use Timeouts">
    ```python theme={null}
    # Set reasonable timeouts
    try:
        result = await asyncio.wait_for(
            job.wait_for_completion(),
            timeout=60.0
        )
    except asyncio.TimeoutError:
        print("Task took too long")
    ```
  </Accordion>
</AccordionGroup>

## See Also

* [AltrinaClient](/sdk/sync-client) - Synchronous client
* [BrowserAgent](/sdk/browser-agent) - Simple one-line interface
* [Advanced Examples](/sdk/advanced-examples) - Complex real-world scenarios
* [Error Handling](/sdk/error-handling) - Async error handling patterns