from typing import Optional
from langchain_core.tools import tool
from scrapegraph_py import ScrapeGraphAI, MarkdownFormatConfig, JsonFormatConfig
sgai = ScrapeGraphAI() # reads SGAI_API_KEY from env
def _unwrap(result):
"""Return the SDK response payload as a plain dict."""
if result.error:
raise RuntimeError(f"ScrapeGraph error: {result.error}")
data = result.data
return data.model_dump() if hasattr(data, "model_dump") else data
# --- content endpoints -------------------------------------------------------
@tool
def scrape(url: str) -> dict:
"""Fetch a web page and return its content as markdown."""
return _unwrap(sgai.scrape(url=url, formats=[MarkdownFormatConfig()]))
@tool
def extract(url: str, prompt: str) -> dict:
"""Extract structured data from a web page using a natural-language prompt."""
return _unwrap(sgai.extract(prompt=prompt, url=url))
@tool
def search(query: str, num_results: int = 3) -> dict:
"""Run an AI web search; returns ranked results with fetched content."""
return _unwrap(sgai.search(query=query, num_results=num_results))
# --- crawl (async job) -------------------------------------------------------
@tool
def crawl_start(url: str, max_depth: int = 2, max_pages: int = 10) -> dict:
"""Start a multi-page crawl job. Returns a dict including the crawl `id`."""
return _unwrap(sgai.crawl.start(
url=url, max_depth=max_depth, max_pages=max_pages,
formats=[MarkdownFormatConfig()],
))
@tool
def crawl_get(crawl_id: str) -> dict:
"""Fetch the status and result of a crawl job."""
return _unwrap(sgai.crawl.get(crawl_id))
@tool
def crawl_stop(crawl_id: str) -> dict:
"""Stop a running crawl."""
return _unwrap(sgai.crawl.stop(crawl_id))
@tool
def crawl_resume(crawl_id: str) -> dict:
"""Resume a stopped crawl."""
return _unwrap(sgai.crawl.resume(crawl_id))
@tool
def crawl_delete(crawl_id: str) -> dict:
"""Delete a crawl job."""
return _unwrap(sgai.crawl.delete(crawl_id))
# --- monitor (scheduled jobs) ------------------------------------------------
@tool
def monitor_create(url: str, interval: str, name: Optional[str] = None, prompt: Optional[str] = None) -> dict:
"""Create a scheduled monitor. If `prompt` is given each tick stores JSON
extraction; otherwise it stores markdown. `interval` is cron syntax,
e.g. "0 9 * * *" for daily at 9am."""
formats = [JsonFormatConfig(prompt=prompt)] if prompt else [MarkdownFormatConfig()]
return _unwrap(sgai.monitor.create(url=url, interval=interval, name=name, formats=formats))
@tool
def monitor_list() -> list:
"""List all monitors."""
return _unwrap(sgai.monitor.list())
@tool
def monitor_get(monitor_id: str) -> dict:
"""Get one monitor by id."""
return _unwrap(sgai.monitor.get(monitor_id))
@tool
def monitor_pause(monitor_id: str) -> dict:
"""Pause a monitor."""
return _unwrap(sgai.monitor.pause(monitor_id))
@tool
def monitor_resume(monitor_id: str) -> dict:
"""Resume a paused monitor."""
return _unwrap(sgai.monitor.resume(monitor_id))
@tool
def monitor_delete(monitor_id: str) -> dict:
"""Delete a monitor."""
_unwrap(sgai.monitor.delete(monitor_id))
return {"deleted": monitor_id}
@tool
def monitor_activity(monitor_id: str) -> dict:
"""Get the recent runs of a monitor."""
return _unwrap(sgai.monitor.activity(monitor_id))
# --- account / history -------------------------------------------------------
@tool
def history_list(service: Optional[str] = None, page: int = 1, limit: int = 20) -> dict:
"""List recent API request history, optionally filtered by service."""
return _unwrap(sgai.history.list(service=service, page=page, limit=limit))
@tool
def history_get(request_id: str) -> dict:
"""Get a single history entry by request id."""
return _unwrap(sgai.history.get(request_id))
@tool
def credits() -> dict:
"""Check remaining ScrapeGraph API credits."""
return _unwrap(sgai.credits())
ALL_TOOLS = [
scrape, extract, search,
crawl_start, crawl_get, crawl_stop, crawl_resume, crawl_delete,
monitor_create, monitor_list, monitor_get,
monitor_pause, monitor_resume, monitor_delete, monitor_activity,
history_list, history_get, credits,
]
