| 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 |
|
|
