graph-crawler

A web crawler that builds a graph of a site: pages are nodes, links are edges. The result is a Graph object you can query, export, save and load again.

Pages are fetched through a chain of drivers with escalation: plain HTTP first, then HTTP with a browser TLS fingerprint, then a real browser, then an "undetected" browser. The browser step can use any engine: CloakBrowser, patchright, stock Playwright or your own binary.

Python 3.11+.

Install

pip install graph-crawler                 # HTTP only (aiohttp)
pip install "graph-crawler[browser]"      # + browser (stock Playwright)
pip install "graph-crawler[antidetect]"   # + TLS, patchright, CloakBrowser, nodriver

extra	what it adds
`browser`	`driver="playwright"`, stock Playwright (`playwright` is an alias of this extra)
`tls`	`curl_cffi` — TLS fingerprint tier, `driver="tls"`
`stealth`	`patchright` + `playwright-stealth` — engine `"patchright"`
`cloak`	`cloakbrowser` — engine `"cloakbrowser"` (stealth in the binary)
`undetected`	`nodriver` — last tier, `driver="nodriver"`
`antidetect`	the five extras above
`server`	FastAPI dashboard / REST / WebSocket (`graph_crawler.api.server`)
`text`	`tiktoken` + `stop-words` — `count_tokens`, `extract_keywords`
`markdown`	`html2md-clean` + `markdown-it-py` — `MarkdownPlugin`
`fuzzy`	`rapidfuzz` — `graph.search(..., mode="fuzzy")`
`sitemap-fast`	`lxml` sitemap parser
`mongodb`, `postgresql`, `celery`, `viz`, `articles`, `performance`, `system`, `all`	storage backends, distributed mode and other extras

After installing a browser extra, download the browser once:

playwright install chromium            # stock Playwright / patchright
python -m cloakbrowser install         # CloakBrowser binary (~200 MB)
# or point to an existing binary: export CLOAKBROWSER_BINARY=/opt/cloak/chrome

Quickstart

import graph_crawler as gc

graph = gc.crawl("https://example.com", max_depth=2, max_pages=50)

print(gc.quick_stats(graph))           # "12 nodes (12 scanned) | 31 edges"
for node in graph.nodes.values():
    print(node.depth, node.response_status, node.url)

hits = graph.search("pricing")         # [SearchHit(node, score, matched_fields, snippet), ...]
tree = graph.url_tree()
graph.export_nodes("nodes.json")
graph.export_edges("edges.json")

graph.nodes also holds pages that were discovered but not fetched (for example, beyond max_pages); node.scanned tells them apart.

Async:

graph = await gc.async_crawl("https://example.com", driver="escalating")

Reuse one crawler for several runs (the driver and storage stay open):

with gc.Crawler(driver="escalating", storage="sqlite", storage_config={"path": "site.db"}) as c:
    g1 = c.crawl("https://a.example")
    g2 = c.crawl("https://b.example")

Crawl only some pages:

from graph_crawler import URLRule

graph = gc.crawl(
    "https://shop.example",
    url_rules=[URLRule(pattern=r"/product/", should_scan=True),
               URLRule(pattern=r"/cart", should_scan=False)],
)

Start from a sitemap:

graph = gc.crawl_sitemap("https://example.com/sitemap.xml", max_urls=500)

Save and load:

gc.save_graph(graph, "site.json")      # format="json" (default) or "sqlite"
graph = gc.load_graph("site.json")

CLI:

graph-crawler crawl https://example.com --max-depth 2 --driver escalating --save example
graph-crawler list
graph-crawler info example

Inspecting pages

Every node records how it was fetched:

node = graph.get_node_by_url("https://example.com/")

node.ok                 # scanned, 2xx/3xx, not blocked, no error
node.is_blocked         # classified as an anti-bot block
node.channel            # "http" | "tls" | "browser" | "undetected"
node.bot_block_status   # None | "none" | "soft_block" | "challenge" | "hard_block" | "captcha"
node.served_by_driver   # "aiohttp" | "tls" | "playwright" | "nodriver"
node.served_by_engine   # "cloakbrowser" | "patchright" | "playwright" | None
node.escalations        # how many tiers were tried before this one
node.fetch_info         # all of the above as one dict

A response with status 400 or higher keeps metadata["error"] = "HTTP <code>", so node.ok is False for error pages.

Drivers

driver= accepts a name or a driver instance.

name	tier	needs	notes
`aiohttp` / `http` / `async`	http	—	default
`tls` / `curl_cffi`	tls	`[tls]`	Chrome TLS fingerprint, no JavaScript
`playwright`	browser	`[browser]`	real browser; engine chosen by `browser_engine`
`nodriver`	undetected	`[undetected]`	CDP without webdriver
`escalating`	all of them	whatever is installed	http → tls → browser → undetected

escalating creates tiers lazily: a browser starts only when a cheaper tier is blocked. Tiers whose dependency is not installed are skipped.

Anti-detect: escalation and browser engines

How escalation decides

Every response goes through classify_block():

result	when
`none`	normal page
`soft_block`	429, plain 503, or 200 with an empty body
`challenge`	Cloudflare "Just a moment", `cf-mitigated: challenge`, Incapsula, DDoS-Guard
`hard_block`	403/401/405 with a vendor or "Access Denied" marker (Akamai `_abck`, DataDome/Incapsula headers, `x-iinfo`, Cloudflare `server`)
`captcha`	PerimeterX, DataDome captcha, reCAPTCHA / hCaptcha / Turnstile markup

Anything other than none, and network errors, move the request to the next tier.

A bare 403/401/405 without a vendor marker is classified as none (authorization or method, not a bot block), so escalating does not open a browser for it and logs the decision at INFO. If such a page is a block, use driver="playwright".

The tier that worked is remembered per host:

the entry expires after GC_TIER_TTL seconds (default 1200, about the lifetime of a cf_clearance cookie); after that the host is probed from the cheapest tier again;
a block on the remembered tier drops the entry and escalates from the next tier;
only one request per host escalates at a time, the others wait for its result;
the memory is per process; workers do not share it.

Each FetchResponse describes what happened in extras (on a node — see Inspecting pages):

resp.extras["bot_block_status"]  # "none" | "soft_block" | "challenge" | "hard_block" | "captcha"
resp.extras["channel"]           # "http" | "tls" | "browser" | "undetected"
resp.extras["driver"]            # "aiohttp" | "tls" | "playwright" | "nodriver"
resp.extras["engine"]            # "cloakbrowser" | "patchright" | "playwright" | None
resp.extras["escalations"]       # how many tiers were tried before this one

Using it

import graph_crawler as gc

# default chain; the browser tier uses CloakBrowser if its binary is present
graph = gc.crawl("https://protected.example", driver="escalating")

# choose the browser engine and stop after the browser tier (0=http 1=tls 2=browser 3=undetected)
graph = gc.crawl(
    "https://protected.example",
    driver="escalating",
    driver_config={"browser_engine": "patchright", "max_tier": 2},
)

As a driver:

from graph_crawler.infrastructure.transport.escalation import EscalatingDriver, EscalationPolicy

async with EscalatingDriver(policy=EscalationPolicy.default(browser_engine="cloakbrowser")) as d:
    resp = await d.fetch("https://protected.example")
    print(resp.status_code, resp.extras["channel"], resp.extras["engine"])

A custom chain:

from graph_crawler.infrastructure.transport import DriverType
from graph_crawler.infrastructure.transport.escalation import EscalationPolicy, EscalationTier

policy = EscalationPolicy(
    tiers=[
        EscalationTier(channel="tls", driver=DriverType.TLS, config={"impersonate": "chrome"}),
        EscalationTier(channel="browser", driver=DriverType.PLAYWRIGHT, engine="cloakbrowser"),
    ],
    remember_per_host=False,
)

Browser engines

The browser tier is PlaywrightDriver; the engine decides which Playwright module and which Chromium binary it launches.

engine	what it is	falls back to
`cloakbrowser`	Chromium with fingerprint patches compiled into the binary	`patchright`
`patchright`	Playwright fork without automation leaks (`Runtime.enable` …)	`playwright`
`playwright`	stock Playwright	—

A fallback is logged once at INFO with the reason. To get an ImportError instead of a silent downgrade, set GC_ENGINE_STRICT=1, or pass {"browser_engine": {"strict": True}} (driver_config={"engine_strict": True} for escalating).

Choose the engine in the config:

gc.crawl(url, driver="playwright", driver_config={"browser_engine": {"type": "cloakbrowser"}})

or with GC_BROWSER_ENGINE=cloakbrowser (used when the config sets no engine).

from graph_crawler.infrastructure.transport.playwright.engine import register_engine

register_engine(
    "my-chrome",
    executable_resolver=lambda: "/opt/my-chrome/chrome",   # None → fall back
    launch_hook=lambda args: ([*args, "--my-flag"], {"ignore_default_args": ["--enable-automation"]}),
    fallback="playwright",
)

gc.crawl(url, driver="escalating", driver_config={"browser_engine": "my-chrome"})

Pass playwright_module="patchright.async_api" to register_engine if the engine needs a different Playwright implementation.

Stealth lives in the engine. The library decides when to escalate and which engine to launch.

Page actions

The Playwright driver can click, type and wait before the HTML snapshot is taken:

from graph_crawler import PageAction

graph = gc.crawl(
    "https://shop.example/search",
    max_depth=0,
    driver="playwright",
    driver_config={"actions": [
        PageAction(type="fill", placeholder="Search", value="laptop"),
        {"type": "click", "text": "Apply"},
        {"type": "wait_for", "response": "/api/items"},
    ]},
)
node = graph.get_node_by_url("https://shop.example/search")
node.metadata["actions_log"]            # [{"index": 0, "ok": True, ...}, ...]
node.metadata["requests_after_actions"]  # XHR/fetch requests made after the actions

Action types: click, fill, select, press, scroll, wait_for, back. A target is found by selector, text, placeholder or label (nth picks among visible matches). The first failing action stops the chain and is written to node.metadata["actions_error"].

Controlling a running crawl

async_crawl(..., control_channel=channel) (and crawl, GraphCrawlerClient.crawl) reads commands from the channel between pages:

import graph_crawler as gc
from graph_crawler.domain.interfaces import AsyncQueueControlChannel, force_visit_command, stop_command

channel = AsyncQueueControlChannel()
channel.send(force_visit_command("https://example.com/hidden-page"))
graph = await gc.async_crawl("https://example.com", control_channel=channel)

# from a plugin, a callback or another task while the crawl runs:
channel.send(stop_command("enough data"))   # graph.metadata["stop_reason"] == "enough data"

command	`data`	effect
`STOP`	`reason` (optional)	stops the crawl, `stop_reason` = `reason`
`PAUSE`	—	pauses until `RESUME` or `STOP`
`RESUME`	—	continues a paused crawl
`REPRIORITIZE`	`url`, `priority`	moves a queued URL in the queue
`FORCE_VISIT`	`url`	adds the URL to the queue with top priority

SPIDER_COMMANDS is this set. SKIP_URL, CHANGE_STRATEGY, SET_DEPTH_LIMIT, REQUEST_STATE and UPDATE_GOAL are declared in CrawlCommand but not executed: the spider logs WARNING Unsupported control command ... (once per command type per crawl) and goes on; they will be removed in the next minor release. A supported command without its required data fields is also skipped with a WARNING.

Plugins and custom nodes

A node plugin runs on every fetched page and writes to user_data, metadata or the typed fields of your Node subclass:

import graph_crawler as gc
from graph_crawler import BaseNodePlugin, Node, NodePluginType


class ProductNode(Node):
    price: int | None = None


class PriceExtractor(BaseNodePlugin):
    url_pattern = r"/product/"        # run only on matching URLs (or: domains = ["shop.example"])
    expects_nonempty = True           # log a WARNING when nothing was extracted
    extraction_key = "prices"

    @property
    def name(self):
        return "price_extractor"

    @property
    def plugin_type(self):
        return NodePluginType.ON_HTML_PARSED

    def execute(self, ctx):
        el = ctx.html_tree.css_first(".price")
        if el is not None:
            ctx.user_data["prices"] = [el.text()]
            ctx.set_node_field("price", el.text())   # validated against the annotation: "10" → 10
        return ctx


graph = gc.crawl("https://shop.example", node_class=ProductNode, plugins=[PriceExtractor()])

domains / url_pattern can also be passed per instance: PriceExtractor(config={"domains": ["shop.example"]}). Several instances of one plugin class with different routes run side by side.
graph.search("acme", user_data_keys=("prices",)) searches plugin data as well as titles and URLs.
node.public_user_data is user_data without the reserved internal keys (RESERVED_USER_DATA_KEYS).
gc.load_graph(path) restores the Node subclass when its module is already imported; load_graph(path, node_class=ProductNode) sets it explicitly.

Storage

storage="memory" (default), "json", "sqlite", "postgresql", "mongodb". For large crawls use sqlite or a server backend: nodes are evicted from RAM and loaded back on access.

Environment variables

variable	effect
`GC_TIER_TTL`	seconds the working tier is remembered per host (default 1200)
`GC_BROWSER_ENGINE`	default browser engine when the config sets none
`GC_ENGINE_STRICT`	`1` — raise `ImportError` instead of falling back to another engine
`CLOAKBROWSER_BINARY`	path to an existing CloakBrowser binary

Logging

The library adds a NullHandler to the graph_crawler logger and never configures logging itself. At INFO it writes one line per fetched page, plus the start of the crawl and warnings such as HTTP 404 for <url>:

import logging
logging.basicConfig(level=logging.INFO)

Metadata

Release files for graph-crawler 5.4.2

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Source distribution (sdist)

Source distribution for graph-crawler 5.4.2
File	Size	Uploaded	Provenance
graph_crawler-5.4.2.tar.gz	1.1 MB	Oct 14, 2026	Without provenance	Details

Built distribution (wheel)

Table of built distributions (wheels) for graph-crawler 5.4.2
File	Interpreter	ABI	Platform	Reset
graph_crawler-5.4.2-py3-none-any.whl 1.4 MB Oct 14, 2026 Without provenance	Python 3	none	any	Details

Total release size: 2.4 MB

Release files / graph_crawler-5.4.2.tar.gz

Download URL	graph_crawler-5.4.2.tar.gz
Size	1.1 MB
Tags	Source
SHA-256 checksum How to use checksums	`8e307641e2fa09a276b0830c3eff83f61bef47ebd5294d22a38a9676ace6e93c`
BLAKE2b-256 checksum How to use checksums	`a1ac5f5a4e397fa614984c945f80f541b63921e98d432bed61316d81888c6b24`
Upload date	Oct 14, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.14.7`

Release files / graph_crawler-5.4.2-py3-none-any.whl

Download URL	graph_crawler-5.4.2-py3-none-any.whl
Size	1.4 MB
Tags	Python 3
SHA-256 checksum How to use checksums	`86fe8ad8c2c3947a08bc64651670f71e49d1359057544161c0f996558f9ac2d5`
BLAKE2b-256 checksum How to use checksums	`f9e845e6013ddb3ec856b63cb06bdd4cddcbd0cf870512791de5d35dae632f81`
Upload date	Oct 14, 2026
Uploaded using Trusted Publishing? What is trusted publishing?	No
Uploaded via	`twine/7.0.0 CPython/3.14.7`

Security for graph-crawler 5.4.2

Report project as malware

Provenance

Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.

No provenance found

None of the files in this release were uploaded with provenance metadata.

A note on package security

PyPI is a repository for distributing packages, not a service for auditing them. While third-party security researchers and our staff do their best to detect and remove malicious packages, PyPI cannot guarantee the quality, security or suitability of any given package.