File size: 2,378 Bytes
c4166f4 | 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 | from typing import Any, Optional
from smolagents.tools import Tool
import os
import urllib
import json
class SpecStreamWebSearch(Tool):
name = "specstream_web_search"
description = "Searches the live web in seconds with the SpecStream Search API (GET /v1/web-search, $5 per 1,000 searches, no quote needed). Returns the answer with its source URL and read time, matching rows, ranked sources and optional page passages. Needs SPECSTREAM_API_KEY."
inputs = {'q': {'type': 'string', 'description': 'The search question or request in everyday words (1 to 2,000 characters).'}, 'passages': {'type': 'boolean', 'description': 'Set True to include the main text passages of the top matching pages.', 'nullable': True}}
output_type = "string"
def forward(self, q: str, passages: bool | None = None) -> str:
import json
import os
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
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)."
query = (q or "").strip()
if not query or len(query) > 2000:
return "Provide a search query between 1 and 2,000 characters."
base = os.environ.get("SPECSTREAM_API_BASE", "https://api.specstream.ai").rstrip("/")
params = {"q": query}
if passages:
params["passages"] = "true"
req = Request(
base + "/v1/web-search?" + urlencode(params),
method="GET",
headers={
"Authorization": "Bearer " + key,
"Accept": "application/json",
"User-Agent": "specstream-smolagents/0.2",
},
)
try:
with urlopen(req, timeout=60) 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
|