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

# Python SDK

> Use X1-BaaS from Python with httpx, including a ready-to-use client wrapper.

# Python

X1-BaaS is a plain REST API, so any HTTP client works. This guide shows a minimal
client wrapper built on [`httpx`](https://www.python-httpx.dev/) that covers
authentication, error handling, and all core endpoints.

## Install

```bash theme={null}
pip install httpx
```

## Client wrapper

```python Filename baas.py theme={null}
from __future__ import annotations

import base64
import json
from typing import Any, Optional

import httpx


class BaasClient:
    """Minimal X1-BaaS client."""

    BASE_URL = "https://api.tazpal.com"

    def __init__(
        self,
        api_key: Optional[str] = None,
        payment_permit: Optional[str] = None,
        timeout: float = 120.0,
    ) -> None:
        self.api_key = api_key
        self.payment_permit = payment_permit  # base64-encoded x402 permit
        self.timeout = timeout

    def _headers(self, extra: Optional[dict] = None) -> dict:
        headers = {"Content-Type": "application/json", **(extra or {})}
        if self.api_key:
            headers["Authorization"] = f"Bearer {self.api_key}"
        elif self.payment_permit:
            headers["X-PAYMENT"] = self.payment_permit
        return headers

    def _request(self, method: str, path: str, **kwargs) -> dict:
        resp = httpx.request(
            method,
            f"{self.BASE_URL}{path}",
            headers=self._headers(),
            timeout=self.timeout,
            **kwargs,
        )
        if resp.status_code >= 400:
            raise BaasError(resp.status_code, resp.json())
        return resp.json()

    # -- Scrape ------------------------------------------------------------

    def scrape(self, url: str, output: str = "markdown", **options: Any) -> dict:
        body = {"url": url, "output": output}
        if options:
            body["options"] = options
        return self._request("POST", "/v1/scrape", json=body)

    # -- Extract -----------------------------------------------------------

    def extract(
        self,
        url: str,
        schema: Optional[dict] = None,
        selectors: Optional[dict] = None,
        xpaths: Optional[dict] = None,
    ) -> dict:
        body: dict[str, Any] = {"url": url}
        if schema:
            body["schema"] = schema
        if selectors:
            body["selectors"] = selectors
        if xpaths:
            body["xpaths"] = xpaths
        return self._request("POST", "/v1/extract", json=body)

    # -- Crawl -------------------------------------------------------------

    def crawl(
        self,
        start_url: Optional[str] = None,
        mode: Optional[str] = None,
        urls: Optional[list[str]] = None,
        **kwargs: Any,
    ) -> dict:
        body: dict[str, Any] = {"mode": mode, **kwargs}
        if start_url:
            body["start_url"] = start_url
        if urls:
            body["urls"] = urls
        return self._request("POST", "/v1/crawl", json=body)

    def crawl_status(self, job_id: str) -> dict:
        return self._request("GET", f"/v1/crawl/{job_id}")

    def crawl_results(self, job_id: str, limit: int = 100) -> dict:
        return self._request("GET", f"/v1/crawl/{job_id}/results?limit={limit}")

    def cancel_crawl(self, job_id: str) -> None:
        httpx.request(
            "DELETE",
            f"{self.BASE_URL}/v1/crawl/{job_id}",
            headers=self._headers(),
            timeout=self.timeout,
        )

    # -- Public ------------------------------------------------------------

    def health(self) -> dict:
        return self._request("GET", "/health")

    def pricing(self) -> dict:
        return self._request("GET", "/v1/pricing")


class BaasError(Exception):
    def __init__(self, status_code: int, body: Any) -> None:
        self.status_code = status_code
        self.body = body
        super().__init__(f"HTTP {status_code}: {body}")
```

## Usage

### Scrape to Markdown

```python theme={null}
from baas import BaasClient, BaasError

client = BaasClient(api_key="baas_live_YOUR_KEY")

try:
    result = client.scrape("https://news.ycombinator.com")
    print(result["data"]["title"])
    print(result["data"]["markdown"][:500])
except BaasError as exc:
    print(f"Error {exc.status_code}: {exc.body}")
```

### Screenshot to file

```python theme={null}
import base64

result = client.scrape(
    "https://example.com",
    output="screenshot",
    format="png",
    full_page=True,
)
png = base64.b64decode(result["data"]["data"])
with open("page.png", "wb") as f:
    f.write(png)
```

### Structured extraction

```python theme={null}
result = client.extract(
    "https://example.com/product/123",
    schema={
        "type": "object",
        "properties": {
            "title": {"type": "string"},
            "price": {"type": "number"},
        },
    },
)
print(result["data"], result["metadata"]["confidence"])
```

### Crawl a sitemap and poll until done

```python theme={null}
import time

job = client.crawl("https://example.com/sitemap.xml", mode="sitemap", max_pages=500)
job_id = job["job_id"]

while True:
    status = client.crawl_status(job_id)
    if status["status"] in ("completed", "failed", "canceled"):
        break
    print(status["progress"])
    time.sleep(5)

results = client.crawl_results(job_id)
print(f"Got {results['count']} pages")
```

## Error handling

Wrap calls in `try/except BaasError`. Inspect `exc.status_code` and
`exc.body["detail"]["error"]` for a machine-readable code.

```python theme={null}
from baas import BaasClient, BaasError

client = BaasClient(api_key="baas_live_YOUR_KEY")

try:
    client.scrape("https://example.com", output="nope")
except BaasError as exc:
    detail = exc.body.get("detail", {})
    print(detail.get("error"))   # "invalid_output"
    print(detail.get("message"))
```
