File size: 2,449 Bytes
58be75e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 | from typing import Any, Optional
from smolagents.tools import Tool
import os
import re
import urllib
import json
class SpecStreamExtractPages(Tool):
name = "specstream_extract_pages"
description = "Reads the main text, title and published date of 1 to 20 web pages, PDFs, Word documents or Excel spreadsheets live with the SpecStream Extract API (POST /v1/extract, $0.80 per 1,000 pages, empty pages free, no quote needed). Needs SPECSTREAM_API_KEY."
inputs = {'urls': {'type': 'string', 'description': '1 to 20 http or https URLs separated by newlines or commas.'}, 'format': {'type': 'string', 'description': 'Output format: "markdown" (default) or "text".', 'nullable': True}}
output_type = "string"
def forward(self, urls: str, format: str | None = None) -> str:
import json
import os
import re
from urllib.error import HTTPError, URLError
from urllib.request import Request, urlopen
key = os.environ.get("SPECSTREAM_API_KEY", "").strip()
if not key:
return "Set SPECSTREAM_API_KEY to a SpecStream API key (https://specstream.ai/workspace?utm_source=smolagents)."
cleaned = [u.strip() for u in re.split(r"[\r\n,]+", urls or "") if u.strip()]
if not 1 <= len(cleaned) <= 20:
return "Provide 1 to 20 http or https URLs."
fmt = "text" if (format or "markdown").strip() == "text" else "markdown"
base = os.environ.get("SPECSTREAM_API_BASE", "https://api.specstream.ai").rstrip("/")
req = Request(
base + "/v1/extract",
data=json.dumps({"urls": cleaned, "format": fmt}).encode("utf-8"),
method="POST",
headers={
"Authorization": "Bearer " + key,
"Accept": "application/json",
"Content-Type": "application/json",
"User-Agent": "specstream-smolagents/0.2",
},
)
try:
with urlopen(req, timeout=120) as resp:
payload = json.loads(resp.read().decode("utf-8"))
except HTTPError as exc:
return "SpecStream answered HTTP %d" % exc.code
except (URLError, OSError):
return "SpecStream could not be reached"
data = payload.get("data") if isinstance(payload, dict) else {}
return json.dumps(data, ensure_ascii=False)
def __init__(self, *args, **kwargs):
self.is_initialized = False
|