from scrapegraph_py import ScrapeGraphAI
from llama_index.core.tools import FunctionTool
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI

sgai = ScrapeGraphAI()  # reads SGAI_API_KEY

def scrape(url: str) -> str:
    """Fetch a page and return its markdown content."""
    result = sgai.scrape(url)
    if result.status == "error":
        raise RuntimeError(result.error)
    return result.data.results.get("markdown", {}).get("data", [""])[0]

agent = FunctionAgent(
    tools=[FunctionTool.from_defaults(fn=scrape)],
    llm=OpenAI(model="gpt-4o"),
)

from pydantic import BaseModel, Field
from typing import List
from scrapegraph_py import ScrapeGraphAI

class FounderSchema(BaseModel):
    name: str = Field(description="Name of the founder")
    role: str = Field(description="Role of the founder in the company")
    linkedin: str = Field(description="LinkedIn profile of the founder")

class PricingPlanSchema(BaseModel):
    tier: str = Field(description="Name of the pricing tier")
    price: str = Field(description="Price of the plan")
    credits: int = Field(description="Number of credits included in the plan")

class SocialLinksSchema(BaseModel):
    linkedin: str
    twitter: str
    github: str

class CompanyInfoSchema(BaseModel):
    company_name: str
    description: str
    founders: List[FounderSchema] = Field(default_factory=list)
    pricing_plans: List[PricingPlanSchema] = Field(default_factory=list)
    social_links: SocialLinksSchema

sgai = ScrapeGraphAI()

res = sgai.extract(
    "Extract info about the company",
    url="https://scrapegraphai.com/",
    schema=CompanyInfoSchema.model_json_schema(),
)

if res.status == "success":
    print(res.data.json_data)

from pydantic import BaseModel, Field
from typing import List
from scrapegraph_py import ScrapeGraphAI

class RepositorySchema(BaseModel):
    name: str = Field(description="Name of the repository (e.g. 'owner/repo')")
    description: str = Field(description="Description of the repository")
    stars: int = Field(description="Star count")
    forks: int = Field(description="Fork count")
    today_stars: int = Field(description="Stars gained today")
    language: str = Field(description="Programming language used")

class ListRepositoriesSchema(BaseModel):
    repositories: List[RepositorySchema]

sgai = ScrapeGraphAI()

res = sgai.extract(
    "Extract only the first ten trending repositories",
    url="https://github.com/trending",
    schema=ListRepositoriesSchema.model_json_schema(),
)

if res.status == "success":
    for repo in res.data.json_data["repositories"]:
        print(f"{repo['name']} — {repo['stars']} ★")

from pydantic import BaseModel, Field
from typing import List
from scrapegraph_py import ScrapeGraphAI

class NewsItemSchema(BaseModel):
    category: str = Field(description="Category of the news (e.g. 'Health', 'Environment')")
    title: str = Field(description="Title of the news article")
    link: str = Field(description="URL to the news article")
    author: str = Field(description="Author of the news article")

class ListNewsSchema(BaseModel):
    news: List[NewsItemSchema]

sgai = ScrapeGraphAI()

res = sgai.extract(
    "Extract the first 10 news articles on the page",
    url="https://www.wired.com/category/science/",
    schema=ListNewsSchema.model_json_schema(),
)

if res.status == "success":
    for item in res.data.json_data["news"]:
        print(f"[{item['category']}] {item['title']}")

from pydantic import BaseModel, Field
from typing import List
from scrapegraph_py import ScrapeGraphAI, FetchConfig

class HouseListingSchema(BaseModel):
    price: int = Field(description="Price of the house in USD")
    bedrooms: int
    bathrooms: int
    square_feet: int = Field(description="Total square footage of the house")
    address: str
    city: str
    state: str
    zip_code: str
    tags: List[str] = Field(description="Tags like 'New construction' or 'Large garage'")
    agent_name: str
    agency: str

class HousesListingsSchema(BaseModel):
    houses: List[HouseListingSchema]

sgai = ScrapeGraphAI()

# Anti-bot heavy sites need stealth + JS rendering
res = sgai.extract(
    "Extract information about houses for sale",
    url="https://www.zillow.com/san-francisco-ca/",
    schema=HousesListingsSchema.model_json_schema(),
    fetch_config=FetchConfig(mode="js", stealth=True, wait=2000),
)

from scrapegraph_py import ScrapeGraphAI, MarkdownFormatConfig
from llama_index.core.tools import FunctionTool
from llama_index.core.agent import ReActAgent
from llama_index.llms.openai import OpenAI

sgai = ScrapeGraphAI()

def scrape(url: str) -> str:
    """Fetch a page and return its markdown content."""
    res = sgai.scrape(url, formats=[MarkdownFormatConfig()])
    if res.status != "success":
        return res.error or ""
    return res.data.results.get("markdown", {}).get("data", [""])[0]

def extract(url: str, prompt: str) -> dict:
    """Extract structured data from a URL using the given prompt."""
    res = sgai.extract(prompt, url=url)
    return res.data.json_data if res.status == "success" else {"error": res.error}

tools = [FunctionTool.from_defaults(fn=f) for f in (scrape, extract)]

agent = ReActAgent.from_tools(
    tools,
    llm=OpenAI(model="gpt-4o"),
    verbose=True,
)

response = agent.chat(
    "Extract all the keyboard names and prices from "
    "https://www.ebay.com/sch/i.html?_nkw=keyboards"
)
print(response)

from scrapegraph_py import (
    ScrapeGraphAI,
    MarkdownFormatConfig, HtmlFormatConfig, JsonFormatConfig,
)
from llama_index.core.tools import FunctionTool

sgai = ScrapeGraphAI()

def scrape(url: str, format: str = "markdown") -> dict:
    """Fetch `url` and return the requested format.

    format: one of "markdown", "html", "json".
    """
    entries = {
        "markdown": MarkdownFormatConfig(mode="reader"),
        "html": HtmlFormatConfig(),
        "json": JsonFormatConfig(prompt="Extract the main content"),
    }
    result = sgai.scrape(url, formats=[entries[format]])
    if result.status == "error":
        return {"error": result.error}
    return result.data.results

scrape_tool = FunctionTool.from_defaults(fn=scrape)

from scrapegraph_py import ScrapeGraphAI
from llama_index.core.tools import FunctionTool

sgai = ScrapeGraphAI()

def extract(url: str, prompt: str, schema: dict | None = None) -> dict:
    """Extract structured data from `url` per `prompt`."""
    result = sgai.extract(prompt, url=url, schema=schema)
    if result.status == "error":
        return {"error": result.error}
    return result.data.json_data

extract_tool = FunctionTool.from_defaults(fn=extract)

from scrapegraph_py import ScrapeGraphAI
from llama_index.core.tools import FunctionTool

sgai = ScrapeGraphAI()

def search(
    query: str,
    num_results: int = 5,
    prompt: str | None = None,
    time_range: str | None = None,
    country: str | None = None,
) -> dict:
    """Search the web and return structured results.

    time_range: "past_hour", "past_24_hours", "past_week", "past_month", "past_year".
    country: two-letter ISO country code (e.g. "us", "it").
    """
    result = sgai.search(
        query,
        num_results=num_results,
        prompt=prompt,
        time_range=time_range,
        location_geo_code=country,
    )
    if result.status == "error":
        return {"error": result.error}
    return {
        "results": [{"title": r.title, "url": r.url} for r in result.data.results],
        "json_data": result.data.json_data,
    }

search_tool = FunctionTool.from_defaults(fn=search)

import time
from scrapegraph_py import ScrapeGraphAI, MarkdownFormatConfig
from llama_index.core.tools import FunctionTool

sgai = ScrapeGraphAI()

def crawl(
    url: str,
    max_depth: int = 2,
    max_pages: int = 50,
    include_patterns: list[str] | None = None,
    exclude_patterns: list[str] | None = None,
) -> dict:
    """Crawl a site and return pages as markdown once the job completes."""
    start = sgai.crawl.start(
        url,
        formats=[MarkdownFormatConfig()],
        max_depth=max_depth,
        max_pages=max_pages,
        include_patterns=include_patterns,
        exclude_patterns=exclude_patterns,
    )
    if start.status == "error":
        return {"error": start.error}

    crawl_id = start.data.id
    while True:
        status = sgai.crawl.get(crawl_id)
        if status.data.status in ("completed", "failed", "stopped"):
            return status.data.model_dump()
        time.sleep(2)

crawl_tool = FunctionTool.from_defaults(fn=crawl)

from scrapegraph_py import ScrapeGraphAI, MarkdownFormatConfig
from llama_index.core.tools import FunctionTool

sgai = ScrapeGraphAI()

def create_monitor(
    url: str,
    name: str,
    interval: str,
    webhook_url: str | None = None,
) -> dict:
    """Create a recurring monitor (cron `interval`) that tracks changes on `url`."""
    result = sgai.monitor.create(
        url,
        interval,
        name=name,
        formats=[MarkdownFormatConfig()],
        webhook_url=webhook_url,
    )
    if result.status == "error":
        return {"error": result.error}
    return {"cron_id": result.data.cron_id}

monitor_tool = FunctionTool.from_defaults(fn=create_monitor)

from scrapegraph_py import ScrapeGraphAI, MarkdownFormatConfig
from llama_index.core.tools import FunctionTool
from llama_index.core.agent.workflow import FunctionAgent
from llama_index.llms.openai import OpenAI

sgai = ScrapeGraphAI()

def scrape(url: str) -> str:
    res = sgai.scrape(url)
    if res.status != "success":
        return res.error or ""
    return res.data.results.get("markdown", {}).get("data", [""])[0]

def extract(url: str, prompt: str) -> dict:
    res = sgai.extract(prompt, url=url)
    return res.data.json_data if res.status == "success" else {"error": res.error}

def search(query: str, num_results: int = 5) -> list[dict]:
    res = sgai.search(query, num_results=num_results)
    if res.status == "error":
        return [{"error": res.error}]
    return [{"title": r.title, "url": r.url} for r in res.data.results]

def crawl(url: str, max_pages: int = 20) -> dict:
    res = sgai.crawl.start(url, formats=[MarkdownFormatConfig()], max_pages=max_pages)
    return {"crawl_id": res.data.id} if res.status == "success" else {"error": res.error}

def create_monitor(url: str, name: str, interval: str) -> dict:
    res = sgai.monitor.create(
        url, interval, name=name, formats=[MarkdownFormatConfig()],
    )
    return {"cron_id": res.data.cron_id} if res.status == "success" else {"error": res.error}

tools = [FunctionTool.from_defaults(fn=f) for f in (
    scrape, extract, search, crawl, create_monitor,
)]

agent = FunctionAgent(
    tools=tools,
    llm=OpenAI(model="gpt-4o"),
    system_prompt=(
        "You are a web research assistant powered by ScrapeGraphAI v2. "
        "Pick the most specific tool for the job: scrape for a single page, "
        "extract for structured data, search for open-web questions, "
        "crawl for multi-page jobs, and create_monitor for recurring jobs."
    ),
)

response = await agent.run(
    "Research the latest blog posts on scrapegraphai.com and summarize them."
)
print(response)

from scrapegraph_py import AsyncScrapeGraphAI
from llama_index.core.tools import FunctionTool

async def scrape(url: str) -> str:
    async with AsyncScrapeGraphAI() as sgai:
        res = await sgai.scrape(url)
        if res.status == "error":
            raise RuntimeError(res.error)
        return res.data.results.get("markdown", {}).get("data", [""])[0]

scrape_tool = FunctionTool.from_defaults(async_fn=scrape)

from llama_index.core.agent.workflow import ReActAgent
from llama_index.llms.anthropic import Anthropic

agent = ReActAgent(
    tools=tools,
    llm=Anthropic(model="claude-sonnet-4-6"),
    verbose=True,
)