DeepResearch API

from valyu import Valyu

valyu = Valyu()

# Create a research task
task = valyu.deepresearch.create(
    query="What are the key differences between RAG and fine-tuning for LLMs?",
    mode="standard"
)

if task.success:
    print(f"Task created: {task.deepresearch_id}")
    
    # Wait for completion with progress updates
    result = valyu.deepresearch.wait(
        task.deepresearch_id,
        on_progress=lambda s: print(f"Status: {s.status}")
    )
    
    if result.status == "completed":
        print(result.output)
        print(f"Cost: ${result.cost}")

# Use fast mode for quick lookups
task = valyu.deepresearch.create(
    query="What is quantum computing?",
    mode="fast"
)

# Use heavy mode for complex research
task = valyu.deepresearch.create(
    query="Analyze the competitive landscape of cloud computing in 2024",
    mode="heavy"
)

# Use max mode for exhaustive research
task = valyu.deepresearch.create(
    query="Comprehensive analysis of AI safety research with fact verification",
    mode="max"
)

task = valyu.deepresearch.create(
    query="Explain quantum computing advancements in 2024",
    output_formats=["markdown"]
)

task = valyu.deepresearch.create(
    query="Write a report on renewable energy trends",
    output_formats=["markdown", "pdf"]
)

result = valyu.deepresearch.wait(task.deepresearch_id)

if result.pdf_url:
    print(f"PDF available at: {result.pdf_url}")

task = valyu.deepresearch.create(
    query="Research competitor pricing in the SaaS market",
    output_formats=[{
        "type": "object",
        "properties": {
            "competitors": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "pricing_model": {"type": "string"},
                        "price_range": {"type": "string"},
                        "key_features": {
                            "type": "array",
                            "items": {"type": "string"}
                        }
                    },
                    "required": ["name", "pricing_model"]
                }
            },
            "market_summary": {"type": "string"},
            "recommendations": {
                "type": "array",
                "items": {"type": "string"}
            }
        },
        "required": ["competitors", "market_summary"]
    }]
)

result = valyu.deepresearch.wait(task.deepresearch_id)

if result.output_type == "json":
    data = result.output
    for competitor in data["competitors"]:
        print(f"{competitor['name']}: {competitor['pricing_model']}")

task = valyu.deepresearch.create(
    query="Research competitor pricing in the SaaS market",
    output_formats=[{
        "type": "object",
        "properties": {
            "competitors": {
                "type": "array",
                "items": {
                    "type": "object",
                    "properties": {
                        "name": {"type": "string"},
                        "pricing_model": {"type": "string"}
                    }
                }
            }
        },
        "required": ["competitors"]
    }]
)

task = valyu.deepresearch.create(
    query="Research AI trends",
    search={
        "search_type": "proprietary",
        "included_sources": ["academic", "finance"],
        "start_date": "2024-01-01",
        "end_date": "2024-12-31",
        "country_code": "US"  # Filter web results by country
    }
)

task = valyu.deepresearch.create(
    query="Analyse Tesla's Q3 2026 earnings. Screenshot their investor relations page and any revenue charts. Calculate YoY revenue growth rates and operating margins.",
    mode="heavy",
    tools={
        "code_execution": { "enabled": True },
        "screenshots": { "enabled": True, "max_calls": 5 },
        "browser_use": { "enabled": True, "max_calls": 2 },
        "charts": True,
    }
)

# Limit browser sessions and screenshots to save cost
task = valyu.deepresearch.create(
    query="Research AWS pricing changes in 2026",
    tools={
        "browser_use": { "enabled": True, "max_calls": 2 },
        "screenshots": { "enabled": True, "max_calls": 5 },
    }
)

result = valyu.deepresearch.wait(task.deepresearch_id)

for image in result.images or []:
    if image.image_type == "screenshot":
        print(f"Screenshot of {image.source_url}: {image.image_url}")

result = valyu.deepresearch.wait(task.deepresearch_id)

for image in result.images or []:
    if image.image_type == "chart":
        print(f"{image.chart_type}: {image.image_url}")

import base64

# Read and encode a PDF
with open("report.pdf", "rb") as f:
    pdf_data = base64.b64encode(f.read()).decode()

task = valyu.deepresearch.create(
    query="Summarize the key findings and compare with market trends",
    mode="heavy",
    files=[{
        "data": f"data:application/pdf;base64,{pdf_data}",
        "filename": "report.pdf",
        "mediaType": "application/pdf",
        "context": "Q4 2024 financial report"  # Optional context
    }]
)

task = valyu.deepresearch.create(
    query="Research the top 20 AI companies in 2024",
    deliverables=[
        "CSV file with company names, founding year, and funding",
        "PowerPoint presentation with 5 slides summarizing key findings"
    ]
)

result = valyu.deepresearch.wait(task.deepresearch_id)

if result.deliverables:
    for deliverable in result.deliverables:
        if deliverable.status == "completed":
            print(f"✅ {deliverable.title}")
            print(f"   Download: {deliverable.url}")
            if deliverable.row_count:
                print(f"   Rows: {deliverable.row_count}")

from valyu import Deliverable

task = valyu.deepresearch.create(
    query="Analyze startup funding trends in 2024",
    deliverables=[
        Deliverable(
            type="xlsx",
            description="Startup funding data with company details",
            columns=["Company", "Funding Amount", "Date", "Investors", "Stage"],
            include_headers=True,
            sheet_name="Funding Data"
        ),
        Deliverable(
            type="pptx",
            description="Executive summary presentation",
            slides=10,
            template="modern"
        ),
        Deliverable(
            type="pdf",
            description="Detailed analysis report with charts and insights"
        )
    ]
)

result = valyu.deepresearch.wait(task.deepresearch_id)

# Access deliverables
for deliverable in result.deliverables:
    print(f"{deliverable.type.upper()}: {deliverable.title}")
    print(f"Status: {deliverable.status}")
    if deliverable.status == "completed":
        print(f"Download: {deliverable.url}")
    elif deliverable.status == "failed":
        print(f"Error: {deliverable.error}")

class DeliverableResult:
    id: str
    request: str  # Original description
    type: str  # "csv" | "xlsx" | "pptx" | "docx" | "pdf"
    status: str  # "completed" | "failed"
    title: str  # Generated filename
    url: str  # Download URL (token-signed, expires)
    s3_key: str
    row_count: int | None  # For CSV/Excel
    column_count: int | None  # For CSV/Excel
    error: str | None  # If failed
    created_at: str  # ISO 8601 timestamp string

task = valyu.deepresearch.create(
    query="Compare the approaches described in these articles",
    urls=[
        "https://example.com/article-1",
        "https://example.com/article-2"
    ]
)

result = valyu.deepresearch.wait(task.deepresearch_id)

if result.status == "completed":
    print(result.output)

def on_progress(status):
    if status.progress:
        pct = (status.progress.current_step / status.progress.total_steps) * 100
        print(f"Progress: {pct:.0f}% - Step {status.progress.current_step}/{status.progress.total_steps}")
    print(f"Status: {status.status}")

result = valyu.deepresearch.wait(
    task.deepresearch_id,
    poll_interval=5,      # Check every 5 seconds
    max_wait_time=900,    # Timeout after 15 minutes (standard mode)
    on_progress=on_progress
)

class DeepResearchStatusResponse:
    success: bool
    deepresearch_id: str
    status: str  # "queued" | "running" | "completed" | "failed" | "cancelled" | "awaiting_input" | "paused"
    query: str
    mode: str  # "fast" | "standard" | "heavy" | "max"
    output_type: str  # "markdown" | "json" | "toon"
    output: str | dict  # Research output
    sources: list[DeepResearchSource]  # Sources used
    cost: float  # Fixed cost for the task
    cost_breakdown: DeepResearchCostBreakdown | None  # Itemized cost breakdown
    tools: DeepResearchTools | None  # Resolved tools configuration
    created_at: str  # ISO 8601 timestamp string
    completed_at: str | None  # ISO 8601 timestamp string
    pdf_url: str | None  # PDF download URL
    images: list[ImageMetadata]  # Generated images
    deliverables: list[DeliverableResult]  # Generated deliverable files
    batch_id: str | None  # Batch ID if task belongs to a batch
    batch_task_id: str | None  # Batch task identifier if task belongs to a batch
    error: str | None  # Error message if failed

class DeepResearchSource:
    title: str
    url: str
    snippet: str
    source: str  # web, pubmed, arxiv, etc.
    word_count: int
    doi: str | None  # For academic papers
    fragment: str | None  # Optional URL fragment — append to `url` for a deep-link to the cited passage

if result.cost_breakdown:
    print(f"Total: ${result.cost}")
    print(f"  Base: ${result.cost_breakdown.task}")
    if result.cost_breakdown.screenshots:
        print(f"  Screenshots: ${result.cost_breakdown.screenshots}")
    if result.cost_breakdown.code_execution:
        print(f"  Code execution: ${result.cost_breakdown.code_execution}")
    if result.cost_breakdown.deliverables:
        print(f"  Deliverables: ${result.cost_breakdown.deliverables}")

status = valyu.deepresearch.status(task_id)

print(f"Status: {status.status}")
if status.progress:
    print(f"Step {status.progress.current_step}/{status.progress.total_steps}")

# Using API key (default)
asset_data = valyu.deepresearch.get_assets(task_id, asset_id)

if asset_data:
    # asset_data is bytes containing the binary data
    with open("output.png", "wb") as f:
        f.write(asset_data)

# Using token
asset_data = valyu.deepresearch.get_assets(
    task_id,
    asset_id,
    token="asset-access-token"
)

result = valyu.deepresearch.toggle_public("dr_abc123", is_public=True)
print(result.public)  # True

# Add first instruction
response = valyu.deepresearch.update(
    task_id,
    instruction="Focus more on peer-reviewed sources from 2024"
)

if response.success:
    print("Instruction added")

# Add another instruction
response = valyu.deepresearch.update(
    task_id,
    instruction="Include a comparison table of major providers"
)

result = client.deepresearch.respond(
    task_id="dr_abc123",
    interaction_id=status.interaction.interaction_id,
    response={"approved": True},
)
print(result.status)  # "running" or "queued"

response = valyu.deepresearch.cancel(task_id)

if response.success:
    print("Task cancelled")

response = valyu.deepresearch.delete(task_id)

if response.success:
    print("Task deleted")

tasks = valyu.deepresearch.list(limit=50)

for task in tasks.data:
    print(f"{task['query'][:50]}... - {task['status']}")

task = valyu.deepresearch.create(
    query="Research market trends in AI",
    webhook_url="https://your-app.com/webhooks/deepresearch"
)

# IMPORTANT: Save the secret immediately - it's only returned once
webhook_secret = task.webhook_secret

task = valyu.deepresearch.create(query="Research query")

if not task.success:
    print(f"Failed to create task: {task.error}")
    return

try:
    result = valyu.deepresearch.wait(
        task.deepresearch_id,
        max_wait_time=1800  # 30 minutes for standard, use 7200 for heavy
    )
    
    if result.status == "completed":
        print(result.output)
    elif result.status == "failed":
        print(f"Research failed: {result.error}")
        
except TimeoutError:
    print("Task timed out - cancelling")
    valyu.deepresearch.cancel(task.deepresearch_id)
    
except ValueError as e:
    print(f"Task error: {e}")

# Production: use webhooks instead of polling
task = valyu.deepresearch.create(
    query="Research query",
    webhook_url="https://your-app.com/webhooks"
)