Workflow: Full SEO Audit Pipeline Fresh

This workflow runs a complete site audit by chaining four API sections in the right order. Backlink and on-page crawls fire in parallel (they're independent), then SERP and keyword data pull after.

Pipeline Architecture

flowchart TD
    A[Input: target domain + keywords] --> B[Phase 1: Parallel Crawls]
    
    B --> C[Backlinks API\nreferring domains,\ntoxic links, anchors]
    B --> D[On-Page API\ncrawl task post]
    
    C --> E[Phase 2: Wait & Collect]
    D --> E
    
    E --> F[Poll on-page\ntask completion]
    F --> G[Phase 3: SERP + Keywords]
    
    G --> H[SERP task_post\nfor all keywords]
    G --> I[Keywords Data\nsearch volume]
    
    H --> J[Poll SERP\ntasks_ready]
    I --> J
    
    J --> K[Phase 4: Parse & Score]
    K --> L[Compile audit report]
    L --> M[Export JSON + CSV]

Phase 0: Setup

import os
import json
import time
import requests
import concurrent.futures
from requests.auth import HTTPBasicAuth
from datetime import datetime

auth = HTTPBasicAuth(
    os.environ["DATAFORSEO_LOGIN"],
    os.environ["DATAFORSEO_PASSWORD"]
)

# Audit configuration
AUDIT_CONFIG = {
    "target_domain": "example.com",
    "target_url": "https://example.com",
    "location_code": 2840,              # United States
    "language_code": "en",
    "keywords": [
        "roofing contractor miami",
        "roof repair miami",
        "metal roofing miami",
        "emergency roof repair miami",
        "commercial roofing miami"
    ],
    "max_crawl_pages": 100,
    "backlink_limit": 200
}

audit_id = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
print(f"Audit ID: {audit_id}")
print(f"Target: {AUDIT_CONFIG['target_domain']}")

Phase 1: Fire Parallel Crawls

Backlink analysis and on-page crawl are independent. Run them simultaneously.

def post_backlinks_task(auth, domain, limit=200):
    """Post a backlinks referring domains request."""
    
    payload = [
        {
            "target": domain,
            "limit": limit,
            "include_subdomains": True,
            "order_by": ["rank,desc"]
        }
    ]
    
    response = requests.post(
        "https://api.dataforseo.com/v3/backlinks/referring_domains/live",
        auth=auth,
        json=payload
    )
    return response.json()

def post_onpage_task(auth, url, max_pages=100):
    """Post an on-page crawl task."""
    
    payload = [
        {
            "target": url,
            "max_crawl_pages": max_pages,
            "crawl_delay": 2,
            "enable_content_parsing": True,
            "respect_sitemap": True
        }
    ]
    
    response = requests.post(
        "https://api.dataforseo.com/v3/on_page/task_post",
        auth=auth,
        json=payload
    )
    return response.json()

# Fire both in parallel using ThreadPoolExecutor
results = {}

with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    backlinks_future = executor.submit(
        post_backlinks_task, auth,
        AUDIT_CONFIG["target_domain"],
        AUDIT_CONFIG["backlink_limit"]
    )
    onpage_future = executor.submit(
        post_onpage_task, auth,
        AUDIT_CONFIG["target_url"],
        AUDIT_CONFIG["max_crawl_pages"]
    )
    
    # Collect results as they complete
    results["backlinks_raw"] = backlinks_future.result()
    results["onpage_task_id"] = onpage_future.result()["tasks"][0]["id"]

print(f"Backlinks: response received")
print(f"On-page crawl task ID: {results['onpage_task_id']}")

Phase 2: Wait for On-Page Crawl

Backlink results are live (instant). On-page crawl takes time.

def wait_for_onpage_crawl(auth, task_id, max_minutes=15):
    """Poll until crawl finishes."""
    
    deadline = time.time() + (max_minutes * 60)
    
    while time.time() < deadline:
        response = requests.get(
            f"https://api.dataforseo.com/v3/on_page/summary/{task_id}",
            auth=auth
        )
        data = response.json()
        result = data["tasks"][0]["result"][0]
        
        status = result.get("crawl_progress", "unknown")
        pages = result.get("pages_crawled", 0)
        print(f"  On-page crawl: {status} ({pages} pages)")
        
        if status == "finished":
            return result
        
        time.sleep(20)
    
    return None

print("\nWaiting for on-page crawl...")
onpage_summary = wait_for_onpage_crawl(auth, results["onpage_task_id"])

Phase 3: SERP + Keywords (Parallel)

Now fire SERP rank tracking and keyword volume simultaneously.

def post_serp_tasks(auth, keywords, location_code, language_code):
    """Post SERP tasks for keyword rank tracking."""
    
    payload = [
        {
            "keyword": kw,
            "location_code": location_code,
            "language_code": language_code,
            "device": "desktop",
            "depth": 100
        }
        for kw in keywords
    ]
    
    response = requests.post(
        "https://api.dataforseo.com/v3/serp/google/organic/tasks_post",
        auth=auth,
        json=payload
    )
    data = response.json()
    
    task_ids = [t["id"] for t in data["tasks"] if t["status_code"] == 20100]
    return task_ids

def get_keyword_volumes(auth, keywords, location_code, language_code):
    """Get search volume for all keywords."""
    
    payload = [
        {
            "keywords": keywords,
            "location_code": location_code,
            "language_code": language_code
        }
    ]
    
    response = requests.post(
        "https://api.dataforseo.com/v3/keywords_data/google_ads/search_volume/live",
        auth=auth,
        json=payload
    )
    data = response.json()
    
    volume_map = {}
    for item in data["tasks"][0].get("result", []):
        volume_map[item["keyword"]] = item.get("search_volume", 0)
    
    return volume_map

# Fire both in parallel
with concurrent.futures.ThreadPoolExecutor(max_workers=2) as executor:
    serp_future = executor.submit(
        post_serp_tasks, auth,
        AUDIT_CONFIG["keywords"],
        AUDIT_CONFIG["location_code"],
        AUDIT_CONFIG["language_code"]
    )
    volume_future = executor.submit(
        get_keyword_volumes, auth,
        AUDIT_CONFIG["keywords"],
        AUDIT_CONFIG["location_code"],
        AUDIT_CONFIG["language_code"]
    )
    
    serp_task_ids = serp_future.result()
    volume_data = volume_future.result()

print(f"SERP tasks posted: {len(serp_task_ids)}")
print(f"Volume data received for {len(volume_data)} keywords")

Phase 3b: Poll SERP Results

def collect_serp_results(auth, task_ids, target_domain, max_wait=300):
    """Wait for SERP tasks and collect position data."""
    
    deadline = time.time() + max_wait
    ready_ids = set()
    serp_results = {}
    
    while time.time() < deadline and len(ready_ids) < len(task_ids):
        response = requests.get(
            "https://api.dataforseo.com/v3/serp/google/organic/tasks_ready",
            auth=auth
        )
        data = response.json()
        
        for task in data["tasks"]:
            for result in task.get("result", []):
                tid = result["id"]
                if tid in task_ids and tid not in ready_ids:
                    ready_ids.add(tid)
        
        print(f"SERP tasks ready: {len(ready_ids)}/{len(task_ids)}")
        
        if len(ready_ids) < len(task_ids):
            time.sleep(30)
    
    # Fetch results for each ready task
    for task_id in ready_ids:
        resp = requests.get(
            f"https://api.dataforseo.com/v3/serp/google/organic/task_get/advanced/{task_id}",
            auth=auth
        )
        task_data = resp.json()["tasks"][0]
        keyword = task_data["data"]["keyword"]
        items = task_data["result"][0].get("items", []) if task_data.get("result") else []
        
        # Find target domain position
        position = None
        for item in items:
            if item.get("type") == "organic" and target_domain in (item.get("url") or ""):
                position = item["rank_absolute"]
                break
        
        serp_results[keyword] = {"position": position, "task_id": task_id}
    
    return serp_results

serp_results = collect_serp_results(auth, set(serp_task_ids), AUDIT_CONFIG["target_domain"])

Phase 4: Compile and Export

def compile_audit_report(config, serp_results, volume_data, onpage_summary, backlinks_raw):
    """Compile all results into a structured audit report."""
    
    # Parse backlinks
    backlink_task = backlinks_raw["tasks"][0]
    backlink_result = backlink_task["result"][0] if backlink_task.get("result") else {}
    referring_domains = backlink_result.get("items", [])
    toxic_count = sum(1 for d in referring_domains if d.get("spam_score", 0) >= 60)
    
    # Parse on-page summary
    pages_crawled = (onpage_summary or {}).get("pages_crawled", 0)
    pages_with_errors = (onpage_summary or {}).get("pages_crawled_with_errors", 0)
    
    # Keyword summary
    keyword_summary = []
    for kw in config["keywords"]:
        keyword_summary.append({
            "keyword": kw,
            "volume": volume_data.get(kw, 0),
            "position": serp_results.get(kw, {}).get("position")
        })
    
    ranked_keywords = [k for k in keyword_summary if k["position"] is not None]
    
    report = {
        "audit_id": audit_id,
        "domain": config["target_domain"],
        "generated_at": datetime.utcnow().isoformat(),
        "backlinks": {
            "referring_domains": len(referring_domains),
            "toxic_domains": toxic_count,
            "toxic_pct": round(toxic_count / len(referring_domains) * 100, 1) if referring_domains else 0
        },
        "on_page": {
            "pages_crawled": pages_crawled,
            "pages_with_errors": pages_with_errors,
            "error_rate_pct": round(pages_with_errors / pages_crawled * 100, 1) if pages_crawled else 0
        },
        "serp": {
            "keywords_tracked": len(config["keywords"]),
            "keywords_ranked": len(ranked_keywords),
            "avg_position": round(
                sum(k["position"] for k in ranked_keywords) / len(ranked_keywords), 1
            ) if ranked_keywords else None
        },
        "keywords": keyword_summary
    }
    
    return report

audit_report = compile_audit_report(
    AUDIT_CONFIG,
    serp_results,
    volume_data,
    onpage_summary,
    results["backlinks_raw"]
)

# Export
report_path = f"audit-{audit_id}.json"
with open(report_path, "w") as f:
    json.dump(audit_report, f, indent=2)

print(f"\nAudit report saved: {report_path}")
print(f"\nSUMMARY")
print(f"{'Referring domains:':<30} {audit_report['backlinks']['referring_domains']:,}")
print(f"{'Toxic domains:':<30} {audit_report['backlinks']['toxic_domains']} ({audit_report['backlinks']['toxic_pct']}%)")
print(f"{'Pages crawled:':<30} {audit_report['on_page']['pages_crawled']}")
print(f"{'Keywords ranked:':<30} {audit_report['serp']['keywords_ranked']} / {audit_report['serp']['keywords_tracked']}")
print(f"{'Avg position:':<30} {audit_report['serp']['avg_position']}")

Rate Limit Management

DataForSEO enforces limits on concurrent requests. When running multi-phase pipelines:

API Section	Recommended Concurrency	Notes
Backlinks	5 concurrent tasks	Live endpoint — no delay needed
On-Page	1 crawl at a time per domain	Be respectful to the target site
SERP	100 keywords per request	Batch efficiently
Keywords Data	1,000 keywords per request	Very high limit

Add delays between task_post calls when posting more than 10 tasks at once:

import time

# Stagger task posts by 0.5s each
for i, batch in enumerate(batches):
    response = requests.post(endpoint, auth=auth, json=batch)
    if i < len(batches) - 1:
        time.sleep(0.5)