Content Endpoint Guide

from valyu import Valyu

valyu = Valyu()  # Uses VALYU_API_KEY from env

data = valyu.contents(
    urls=[
        "https://en.wikipedia.org/wiki/Artificial_intelligence",
    ],
    response_length="medium",
    extract_effort="auto",
)
print(data["results"][0]["content"][:500])

from valyu import Valyu

valyu = Valyu()

data = valyu.contents(
    urls=["https://example.com/article"],
    extract_effort="auto",
    screenshot=True,
)
print(data["results"][0]["screenshot_url"])

from valyu import Valyu

valyu = Valyu()

data = valyu.contents(
    urls=["https://example.com/article"],
    extract_effort="normal",
    summary=False,
)
print(data["results"][0]["content"][:300])

from valyu import Valyu

valyu = Valyu()

data = valyu.contents(
    urls=["https://example.com/article"],
    extract_effort="auto",
    summary=True,
)
print(data["results"][0]["content"])

from valyu import Valyu

valyu = Valyu()

data = valyu.contents(
    urls=["https://example.com/research-paper"],
    extract_effort="auto",
    summary="Summarise the methodology, key findings, and practical applications in 2-3 paragraphs",
)
print(data["results"][0]["content"])

from valyu import Valyu

valyu = Valyu()

data = valyu.contents(
    urls=["https://example.com/product-page"],
    extract_effort="auto",
    summary={
        "type": "object",
        "properties": {
            "product_name": {"type": "string", "description": "Name of the product"},
            "price": {"type": "number", "description": "Product price in USD"},
            "features": {
                "type": "array",
                "items": {"type": "string"},
                "maxItems": 5,
                "description": "Key product features",
            },
            "availability": {
                "type": "string",
                "enum": ["in_stock", "out_of_stock", "preorder"],
                "description": "Product availability status",
            },
        },
        "required": ["product_name", "price"],
    },
)
print(data["results"][0]["content"])

from valyu import Valyu

valyu = Valyu()

# Submit async job — returns immediately with a job_id
job = valyu.contents(
    urls=[
        "https://example.com/page1",
        "https://example.com/page2",
        "https://example.com/page3",
        # ... up to 50 URLs
    ],
    async_mode=True,
    webhook_url="https://your-app.com/webhooks/valyu",  # optional
)

print(f"Job ID: {job['job_id']}")

# Option 1: Block until complete (SDK handles polling)
result = valyu.wait_for_contents_job(
    job["job_id"],
    poll_interval=5,        # seconds between polls
    max_wait_time=3600,     # max seconds to wait
    on_progress=lambda s: print(f"  {s['status']} — batch {s.get('current_batch', '?')}/{s.get('total_batches', '?')}"),
)

# Option 2: Or pass wait=True when submitting to auto-poll
result = valyu.contents(
    urls=["https://example.com/page1", "https://example.com/page2"],
    async_mode=True,
    wait=True,  # blocks until job completes
)

for r in result["results"]:
    print(f"{r['title']}: {r['length']} characters")
print(f"Total cost: ${result['actual_cost_dollars']}")

import time

while True:
    status = valyu.get_contents_job(job["job_id"])
    print(f"Status: {status['status']} ({status.get('current_batch', '?')}/{status.get('total_batches', '?')} batches)")

    if status["status"] in ("completed", "partial", "failed"):
        break
    time.sleep(2)

if status["status"] in ("completed", "partial"):
    for result in status["results"]:
        print(f"{result['title']}: {result['length']} characters")

if status["status"] in ("partial", "failed"):
    print(f"Error: {status.get('error', 'Unknown')}")

import hmac
import hashlib

def verify_webhook(
    payload: bytes,
    signature_header: str,
    timestamp_header: str,
    webhook_secret: str
) -> bool:
    signed_payload = f"{timestamp_header}.{payload.decode('utf-8')}"

    expected = hmac.new(
        webhook_secret.encode("utf-8"),
        signed_payload.encode("utf-8"),
        hashlib.sha256
    ).hexdigest()

    received = signature_header.removeprefix("sha256=")
    return hmac.compare_digest(expected, received)