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