Makra Python SDK

The official Python client for the Makra web extraction API.

Install

pip install makra

For local development from this repository:

pip install -e sdk/python

Quick start

from makra import Makra

with Makra(api_key="mk_live_...") as client:
    result = client.extract(
        urls=["https://example.com"],
        schema={"title": "The page title"},
    )
    print(result)

Every operation exists in two flavours: Makra (synchronous) and AsyncMakra (awaitable). The examples below use the synchronous client; the async one is identical with await and async with.

Three ways to run a workflow

Pick by how long the work takes and how much you want to watch it happen.

1. Convenient

extract and schema durably submit the workflow, poll it, and fetch its stored result. You still make one SDK call, but no HTTP connection has to stay open while the extraction runs. timeout is the total time the SDK waits; if it expires, the server-side run continues and the raised MakraTimeoutError includes its run_id. The default is 300 seconds per origin page. Paginated extracts (pagination.additional_pages) and sequential multi-URL extracts scale that budget automatically unless you pass timeout yourself.

from makra import (
    EventTypes,
    ExtractOptions,
    Iso3166Alpha2,
    Makra,
    ProxyContinents,
    ProxyRegion,
    ProxyRegionScopes,
    StreamDetailTypes,
    ValidationModes,
)

data = client.extract(
    urls=["https://example.com/products"],
    schema={"products": [{"name": "string", "price": "number"}]},
    config=ExtractOptions(
        validation_mode=ValidationModes.REPAIR,
        additional_pages=2,
        proxy_region=ProxyRegion.country(Iso3166Alpha2.DE),
        recovery_retry=True,
        recovery_retry_delay_ms=2000,
    ),
    timeout=120,
)

Nested dictionaries remain supported and produce the same wire payload:

data = client.extract(
    urls=["https://example.com/products"],
    schema={"products": [{"name": "string", "price": "number"}]},
    config={
        "validation_mode": ValidationModes.REPAIR,
        "pagination": {"enabled": True, "additional_pages": 2},
        "crawler": {
            "proxy": {
                "region": {
                    "scope": ProxyRegionScopes.COUNTRY,
                    "value": Iso3166Alpha2.DE,
                }
            },
            "recovery": {"retry": True, "retry_delay_ms": 2000},
        },
    },
)

Discover what a page contains before you write a schema:

page_schema = client.schema(
    "https://example.com/products",
    config={
        "crawler": {
            "proxy": {
                "region": {
                    "scope": ProxyRegionScopes.CONTINENT,
                    "value": ProxyContinents.EUROPE,
                }
            }
        }
    },
)

2. Streaming

The stream carries lifecycle and progress events, not the extracted data. When a terminal event arrives, fetch the stored result.

run_id = None
for event in client.extract_stream(urls=urls, schema=schema):
    run_id = event.run_id
    print(event.sequence, event.type, event.detail_type)
    if event.type == EventTypes.STEP_PROGRESS:
        if event.detail_type == StreamDetailTypes.RUN_TITLE_GENERATED:
            print("title:", event.payload.get("title"))
    if event.is_terminal:
        print("finished:", event.status, event.reason)

data = client.get_run_result(run_id)

Each WorkflowEvent exposes type, sequence, run_id, the raw payload, and the shortcuts is_terminal, detail_type, status, reason, success. Compare event.type to EventTypes and event.detail_type to StreamDetailTypes.

If the connection drops mid-run, the SDK reconnects to the run's event endpoint with Last-Event-ID, so you see every event exactly once. Silence longer than stream_idle_timeout (90s, versus the gateway's 15s heartbeat) counts as a dropped connection.

3. Deferred runs

Submit now, collect later — from a different process, if you like.

run = client.submit_extract(urls=urls, schema=schema)
print(run.id, run.state)          # queued

run.wait()                        # polls until terminal
data = run.result()

Use run_is_terminal and run_succeeded when you have a run mapping and need to distinguish infrastructure completion from domain success. A completed run with no success field is treated as successful; success: false is not.

from makra import run_is_terminal, run_succeeded

if run_is_terminal(snapshot) and run_succeeded(snapshot):
    data = client.get_run_result(snapshot["id"])

A handle also supports refresh(), stream(), and cancel(). Everything a handle does is available directly on the client too — get_run, list_runs, wait_for_run, stream_run_events, get_run_result, cancel_run — so a run id is all you need to resume from anywhere.

for event in client.stream_run_events(run_id, last_event_id=42):
    ...

Configuration

Settings resolve as explicit argument → environment variable → default.

default_headers may add tracing or application headers. It cannot override SDK-owned names (Api-Key, Content-Type, Accept, User-Agent, Idempotency-Key, Prefer, Last-Event-ID); pass api_key, idempotency_key, or last_event_id instead.

from makra import DEVELOPMENT_BASE_URL, Makra

client = Makra(base_url=DEVELOPMENT_BASE_URL, timeout=60, max_retries=4)

Authentication uses the Api-Key header. A client owns an HTTP connection pool, so build one per application and close it when done.

Retries and idempotency

Transient failures (408, 409, 425, 429, 5xx) are retried with exponential backoff and full jitter, honouring Retry-After when the server sends it.

Retrying a submission is only safe because the SDK attaches a fresh Idempotency-Key to every one: the gateway replays the original run instead of starting a second billable one. Pass your own idempotency_key to make a submission replayable across process restarts too.

Errors

Catch MakraError for anything the SDK raises, or a subclass to handle one condition.

MakraError
├── MakraAPIError                     .status_code .code .body .request_id
│   ├── MakraAuthenticationError      401
│   ├── MakraInsufficientCreditsError 402  .required_credits .available_credits
│   ├── MakraPermissionError          403
│   ├── MakraNotFoundError            404
│   ├── MakraInvalidRequestError      4xx  .field .reason .index
│   ├── MakraRateLimitError           429  .retry_after .concurrency
│   └── MakraServerError              5xx
├── MakraConnectionError              the request never reached the API
│   └── MakraTimeoutError
├── MakraStreamError                  .run_id
│   └── MakraResultError              .run_id .location   (also a MakraStreamError)
└── MakraRunFailedError               .run_id .state .run

MakraResultError is raised when a stored-result redirect is missing Location or the location is not an absolute HTTP(S) URL. It subclasses MakraStreamError so existing stream catches keep working; a future major version may reparent it under MakraError. Presigned query strings are never included in the message or location.

Malformed arguments raise ValueError before any network call, so a typo in a config key costs nothing.

from makra import MakraInsufficientCreditsError, MakraRateLimitError

try:
    client.extract(urls=urls, schema=schema)
except MakraInsufficientCreditsError as error:
    print("need", error.required_credits, "have", error.available_credits)
except MakraRateLimitError as error:
    print("retry after", error.retry_after)

Async

import asyncio
from makra import AsyncMakra

async def main():
    async with AsyncMakra() as client:
        async for event in client.extract_stream(urls=urls, schema=schema):
            print(event.type)
        run = await client.submit_schema("https://example.com")
        await run.wait()
        print(await run.result())

asyncio.run(main())

See ../SPEC.md for the complete contract.