Python SDK

Star

The Aluvia Python SDK makes it simple to integrate Aluvia into your agent workflow. There are two key components:

AluviaApi - a lightweight Python wrapper for the Aluvia REST API.
AluviaClient - a local client for connecting to Aluvia.

Get Your API Key

To use the Aluvia SDK, you will need your API key.

Create an account at dashboard.aluvia.io
Go to API and SDKs and get your API Key

Install SDK

pip install aluvia-sdk

AluviaApi

AluviaApi is a typed wrapper for the Aluvia REST API. Use it to manage connections, query account info, or build custom tooling—without starting a proxy.

AluviaApi Reference

Function	Description
`api.account.get()`	Get account info (balance, usage)
`api.account.usage.get(params?)`	Get usage summary (data consumed in GB) for a time range
`api.account.payments.list(params?)`	Lists payment transactions / top-ups for the account
`api.account.connections.list()`	List all connections
`api.account.connections.create()`	Create a new connection
`api.account.connections.get(id)`	Get connection details
`api.account.connections.patch(id)`	Update connection (rules, geo, session)
`api.account.connections.delete(id)`	Delete a connection
`api.geos.list()`	List available geo-targeting options

All high-level helpers automatically:

Unwrap success envelopes — extract data from { "success": True, "data": T }
Throw on non-2xx — convert HTTP errors to ApiError or InvalidApiKeyError
Return typed data — Python type hints for all responses

Account

`api.account.get()`

Retrieves account metadata including balance and connection count.

account = await api.account.get()

Endpoint: GET /account

Returns: Account

{
  "account_id": "1",
  "created_at": 1705478400,
  "aluvia_username": "user@example.com",
  "balance_gb": 84.25,
  "service": "agent_connect",
  "connection_count": 5,
}

`api.account.usage.get(params?)`

Retrieves usage summary (data consumed in GB) for a time range.

# All-time usage
usage = await api.account.usage.get()

# Usage for a specific period
usage = await api.account.usage.get(
  start=1705478400,  # Unix timestamp
  end=1706083200,
)

Endpoint: GET /account/usage

Parameters:

Param	Type	Description
`start`	`integer`	Optional. Unix timestamp for period start.
`end`	`integer`	Optional. Unix timestamp for period end.

Returns: AccountUsage

{
  "account_id": "1",
  "start": 1705478400,
  "end": 1706083200,
  "data_used_gb": 15.75,
}

`api.account.payments.list(params?)`

Lists payment transactions / top-ups for the account.

# All payments
payments = await api.account.payments.list()

# Payments in a date range
payments = await api.account.payments.list(
  start=1705478400,
  end=1708737600,
)

Endpoint: GET /account/payments

Parameters:

Param	Type	Description
`start`	`integer`	Optional. Unix timestamp for period start.
`end`	`integer`	Optional. Unix timestamp for period end.

Returns: Array<AccountPayment>

[
  {
    "id": "1",
    "created_at": 1705600000,
    "amount_usdc": 99.0,
    "service": "agent_connect",
    "status": "paid",
  },
];

Account Connections

`api.account.connections.list()`

Lists all proxy connections under the account.

connections = await api.account.connections.list()

Endpoint: GET /account/connections

Returns: Array<AccountConnection>

[
  {
    "connection_id": "1",
    "created_at": 1705478400,
    "description": "pricing-scraper-1",
  },
  {
    "connection_id": "2",
    "created_at": 1705564800,
    "description": "inventory-monitor",
  },
]

`api.account.connections.create(body)`

Creates a new proxy connection with generated credentials.

connection = await api.account.connections.create(
  description="my-new-agent",
  rules=["*", "-example.com"],
  target_geo="us_ny",
)

print("Connection ID:", connection["connection_id"])
print("Proxy username:", connection["proxy_username"])
print("Proxy password:", connection["proxy_password"])
print("API token:", connection["api_token"])  # Only returned on creation

Endpoint: POST /account/connections

Request Body:

Field	Type	Description
`description`	`string`	Optional. Human-readable description.
`rules`	`string[]`	Optional. Routing rules array.
`target_geo`	`string`	Optional. Geo-targeting code (e.g., `"us_ny"`).
`session_id`	`string`	Optional. Session ID for sticky sessions.

Returns: AccountConnection (full representation including credentials and api_token)

{
  "connection_id": "3",
  "created_at": 1709000000,
  "description": "my-new-agent",
  "proxy_username": "Nkjh78Gh",
  "proxy_password": "zxy987abc",
  "api_token": "alv_connection_token_abc123def456",
  "rules": ["*", "-example.com"],
  "session_id": None,
  "target_geo": "us_ny",
  "proxy_urls": { ... },
}

`api.account.connections.get(connectionId, options?)`

Retrieves a single connection by ID. Supports conditional requests via ETag.

connection = await api.account.connections.get(123)

if connection is None:
  print("Connection unchanged (304)")

Endpoint: GET /account/connections/:id

Parameters:

Param	Type	Description
`connectionId`	`integer`	Connection ID.

Returns: AccountConnection | null

Returns AccountConnection on 200 OK

Note: This is the only helper that returns null for a successful response (the 304 case).

`api.account.connections.patch(connectionId, body)`

Updates a connection's properties. Only provided fields are updated.

# Update rules
await api.account.connections.patch(
  123,
  rules=["example.com", "*.google.com"],
)

# Update session ID
await api.account.connections.patch(
  123,
  session_id="sessionabc",
)

# Update geo targeting
await api.account.connections.patch(
  123,
  target_geo="us_ca",
)

# Clear geo targeting
await api.account.connections.patch(
  123,
  target_geo=None,
)

Endpoint: PATCH /account/connections/:id

Request Body:

Field	Type	Description
`description`	`string`	Optional. Update description.
`rules`	`string[]`	Optional. Update routing rules.
`target_geo`	`string \| null`	Optional. Update or clear geo targeting.
`session_id`	`string \| null`	Optional. Update or clear session ID.

Returns: AccountConnection (updated connection)

`api.account.connections.delete(connectionId)`

Permanently deletes a connection and all associated API tokens.

result = await api.account.connections.delete(123)
print("Deleted:", result["deleted"])  # True

Endpoint: DELETE /account/connections/:id

Returns: AccountConnectionDeleteResult

{
  "connection_id": "123",
  "deleted": True,
}

Geos

`api.geos.list()`

Lists available geographic targeting options.

geos = await api.geos.list()

for geo in geos:
  print(f"{geo['code']}: {geo['label']}")

Endpoint: GET /geos

Returns: Array<Geo>

[
  {"code":"us_az","label":"Arizona"},
  {"code":"us_ca","label":"California"},
  {"code":"us_dc","label":"District of Columbia"},
  {"code":"us_fl","label":"Florida"},
  {"code":"us_ga","label":"Georgia"}
  # ...
]

Note: This endpoint is accessible by both account tokens and connection tokens.

AluviaApi Example Code

import os
from aluvia_sdk import AluviaApi

api = AluviaApi(api_key=os.environ["ALUVIA_API_KEY"])

# Check account balance
account = await api.account.get()
print(f"Balance: {account['balance_gb']} GB")

# Create a connection for a new agent
connection = await api.account.connections.create(
    description="pricing-scraper",
    rules=["competitor-site.com"],
    target_geo="us_ca",
)
print(f"Created: {connection['connection_id']}")

# List available geos
geos = await api.geos.list()
print("Geos:", [g["code"] for g in geos])

Aluvia Client

The Aluvia client runs a local rules-based proxy server on your agent's host, handles authentication and connection management, and provides ready-to-use adapters for popular tools like Playwright, Requests, and httpx.

Simply point your automation tool at the local proxy address (127.0.0.1) and the client handles the rest. For each request, the client checks the destination hostname against user-defined (or agent-defined) routing rules and decides whether to send it through Aluvia's mobile IPs or direct to the destination.

Quick Start

Understand the basics

Requirements: Python 3.9 or later

Example: Dynamic unblocking with Playwright

This example shows how an agent can use the Aluvia client to dynamically unblock websites. It demonstrates starting the client, using the Playwright integration adapter, configuring geo targeting and session ID, detecting blocks, and updating routing rules on the fly.

import asyncio
from playwright.async_api import async_playwright
from aluvia_sdk import AluviaClient

async def main():
    # Initialize the Aluvia client with your API key
    client = AluviaClient(api_key="your-api-key")

    # Start the client (launches local proxy, fetches connection config)
    connection = await client.start()

    # Configure geo targeting (use California IPs)
    await client.update_target_geo("us_ca")

    # Set session ID (requests with the same session ID use the same IP)
    await client.update_session_id("agentsession1")

    # Launch browser using the Playwright integration adapter
    # The adapter returns proxy settings in Playwright's expected format
    async with async_playwright() as p:
        browser = await p.chromium.launch(proxy=connection.as_playwright())

        # Track hostnames we've added to proxy rules
        proxied_hosts = set()

        async def visit_with_retry(url: str) -> str:
            page = await browser.new_page()
            try:
                response = await page.goto(url, wait_until="domcontentloaded")
                hostname = url.split("//")[1].split("/")[0]

                # Detect if the site blocked us (403, 429, or challenge page)
                status = response.status if response else 0
                title = await page.title()
                is_blocked = status in (403, 429) or "blocked" in title.lower()

                if is_blocked and hostname not in proxied_hosts:
                    print(f"Blocked by {hostname} — adding to proxy rules")

                    # Update routing rules to proxy this hostname through Aluvia
                    # Rules update at runtime—no need to restart the browser
                    proxied_hosts.add(hostname)
                    await client.update_rules(list(proxied_hosts))

                    # Rotate to a fresh IP by changing the session ID
                    import time
                    await client.update_session_id(f"retry{int(time.time())}")

                    await page.close()
                    return await visit_with_retry(url)

                return await page.content()
            finally:
                await page.close()

        try:
            # First attempt goes direct; if blocked, retries through Aluvia
            html = await visit_with_retry("https://example.com/data")
            print("Success:", html[:200])
        finally:
            # Always close the browser and connection when done
            await browser.close()
            await connection.close()

asyncio.run(main())

PyPI Package

Releases, metadata, and install info for aluvia-sdk.

View the package on PyPI

Star

Get Your API Key​

Install SDK​

AluviaApi​

AluviaApi Reference​

Account​

api.account.get()​

api.account.usage.get(params?)​

api.account.payments.list(params?)​

Account Connections​

api.account.connections.list()​

api.account.connections.create(body)​

api.account.connections.get(connectionId, options?)​

api.account.connections.patch(connectionId, body)​

api.account.connections.delete(connectionId)​

Geos​

api.geos.list()​

AluviaApi Example Code​

Aluvia Client​

Quick Start​

Understand the basics​

Example: Dynamic unblocking with Playwright​

PyPI Package

Get Your API Key

Install SDK

AluviaApi

AluviaApi Reference

Account

`api.account.get()`

`api.account.usage.get(params?)`

`api.account.payments.list(params?)`

Account Connections

`api.account.connections.list()`

`api.account.connections.create(body)`

`api.account.connections.get(connectionId, options?)`

`api.account.connections.patch(connectionId, body)`

`api.account.connections.delete(connectionId)`

Geos

`api.geos.list()`

AluviaApi Example Code

Aluvia Client

Quick Start

Understand the basics

Example: Dynamic unblocking with Playwright