airflow-provider-anakin

An Apache Airflow provider for Anakin. Scrape pages, map and crawl sites, run web search and multi-step research, run Wire actions on sites like Amazon or Walmart, create website monitors, and drive an AI browser, all as Airflow tasks. Each operator returns JSON-safe results to XCom.

Built on the official anakin-sdk Python client, which handles authentication, retries and job polling.

Install

pip install airflow-provider-anakin

Airflow discovers the provider automatically. Check with:

airflow providers get airflow-provider-anakin

Connection

Get an API key (starts with ask_) from the Anakin dashboard. Sign-up is free and includes 300 credits.

Create a connection with conn type Anakin:

• Connection Id: anakin_default (the default every operator uses; pass anakin_conn_id=... to use another)
• API Key (password field): your Anakin API key
• Base URL (optional extra): only for non-production environments

Or from the CLI:

airflow connections add anakin_default --conn-type anakin --conn-password "ask_..."

If no connection exists, the hook falls back to the ANAKIN_API_KEY environment variable. The connection's Test button checks that a key resolves; it makes no API call, so it never spends credits.

Quick start

import pendulum

from airflow.sdk import DAG  # Airflow 2: from airflow import DAG
from anakin_provider.operators.anakin import AnakinCrawlOperator, AnakinScrapeOperator

with DAG("anakin_quickstart", schedule="@daily", start_date=pendulum.datetime(2026, 1, 1)):
    pricing = AnakinScrapeOperator(
        task_id="scrape_pricing",
        url="https://example.com/pricing",
        generate_json=True,  # also extract structured JSON with AI
    )
    docs = AnakinCrawlOperator(
        task_id="crawl_docs",
        url="https://example.com/docs",
        max_pages=50,
        include_patterns=["/docs/*"],
    )

example_dags/ has a runnable DAG for every operator.

Operators

All operators live in anakin_provider.operators.anakin.

Web data

Wire: pre-built site actions

Wire action inputs go in action_params (params is reserved by Airflow). Actions that need a login take credential_id.

Website monitoring

AI visibility

Browser

Chaining tasks with XCom

Templated fields accept XCom pulls, so one task can feed the next:

discover = AnakinWireDiscoverOperator(
    task_id="discover", q="search products on walmart", auth_mode="none"
)
run = AnakinWireReadActionOperator(
    task_id="run",
    action_id="{{ ti.xcom_pull(task_ids='discover')[0]['action_id'] }}",
    action_params={"query": "wireless headphones"},
)
discover >> run

Long-running jobs

Scrape, map, crawl, agentic search, Wire actions and AI-visibility searches are asynchronous jobs; the operator submits the job and polls until it finishes. poll_timeout (seconds) caps the wait. Agentic search usually takes 1 to 5 minutes and a browser task up to about 6.

Hook

Use AnakinHook from your own tasks:

from airflow.sdk import task
from anakin_provider.hooks.anakin import AnakinHook


@task
def headlines() -> str:
    doc = AnakinHook().scrape("https://example.com/news")
    return doc.markdown

AnakinHook().get_conn() returns the configured anakin.Anakin client for anything else in the SDK.

Errors

Failures raise the SDK's exceptions, all subclasses of anakin.AnakinError (for example AuthenticationError, InsufficientCreditsError, RateLimitError, JobFailedError, JobTimeoutError). Airflow marks the task failed and applies its retry policy. Note that Wire write actions, monitor and Wire build creation, and browser tasks are not idempotent; think before setting retries on them.

Development

uv venv && source .venv/bin/activate
uv pip install -e . --group test --group lint
pytest
ruff check . && ruff format --check . && mypy anakin_provider

License

Apache-2.0