| from typing import Any, Optional |
| from smolagents.tools import Tool |
| import os |
| import urllib |
| import json |
|
|
| class SpecStreamFindWebData(Tool): |
| name = "specstream_find_web_data" |
| description = "Reads a request for web data and returns how SpecStream understood it, plus the maximum price. Nothing runs and nothing is charged until the person approves that maximum. Step 1: call with request only, show the reading and the price to the person. Step 2: only after the person says yes, call again with quote_id and confirm_max_cents set to the quoted maximum in cents. That starts the work and returns a find_id; read the rows with the specstream_get_rows tool. Needs SPECSTREAM_API_KEY." |
| inputs = {'request': {'type': 'string', 'description': 'The data wanted, in plain words.'}, 'count': {'type': 'integer', 'description': 'How many results to price, 1 to 500. Leave empty for the default.', 'nullable': True}, 'quote_id': {'type': 'string', 'description': 'The quote_id from step 1. Only with confirm_max_cents, after the person approved.', 'nullable': True}, 'confirm_max_cents': {'type': 'integer', 'description': 'The maximum in cents from step 1, exactly as quoted. Set only after the person approved it.', 'nullable': True}} |
| output_type = "string" |
|
|
| def _call(self, method: str, path: str, body: dict | None = None) -> dict: |
| import json |
| import os |
| from urllib.error import HTTPError, URLError |
| from urllib.request import Request, urlopen |
|
|
| key = os.environ.get("SPECSTREAM_API_KEY", "").strip() |
| base = os.environ.get("SPECSTREAM_API_BASE", "https://api.specstream.ai").rstrip("/") |
| data = None if body is None else json.dumps(body).encode("utf-8") |
| req = Request( |
| base + path, |
| data=data, |
| method=method, |
| headers={ |
| "Authorization": "Bearer " + key, |
| "Accept": "application/json", |
| "Content-Type": "application/json", |
| "User-Agent": "specstream-smolagents/0.1", |
| }, |
| ) |
| try: |
| with urlopen(req, timeout=120) as resp: |
| payload = json.loads(resp.read().decode("utf-8")) |
| except HTTPError as exc: |
| try: |
| payload = json.loads(exc.read().decode("utf-8")) |
| except ValueError: |
| payload = {} |
| error = payload.get("error") if isinstance(payload, dict) else None |
| message = error.get("message") if isinstance(error, dict) else None |
| raise RuntimeError(message or "SpecStream answered HTTP %d" % exc.code) |
| except (URLError, OSError): |
| raise RuntimeError("SpecStream could not be reached") |
| found = payload.get("data") if isinstance(payload, dict) else None |
| return found if isinstance(found, dict) else {} |
|
|
| def _dollars(self, cents: int, currency: str) -> str: |
| symbol = "$" if currency == "USD" else currency + " " |
| return "%s%.2f" % (symbol, cents / 100) |
|
|
| def forward( |
| self, |
| request: str, |
| count: int | None = None, |
| quote_id: str | None = None, |
| confirm_max_cents: int | None = None, |
| ) -> str: |
| import os |
| from urllib.parse import quote as urlquote |
|
|
| if not os.environ.get("SPECSTREAM_API_KEY", "").strip(): |
| return "Set SPECSTREAM_API_KEY to a SpecStream API key (https://specstream.ai/workspace?utm_source=smolagents)." |
| call = self._call |
| dollars = self._dollars |
|
|
| try: |
| if quote_id is None and confirm_max_cents is None: |
| reading = call("POST", "/v1/parse", {"q": request, "locale": "en-US"}) |
| ast_hash = reading.get("ast_hash") |
| if not ast_hash: |
| return "SpecStream could not read that request. Name what you want and where." |
| sentence = reading.get("read_back") or reading.get("canonical_sentence") or "" |
| body = {"ast_hash": ast_hash, "operation": "find"} |
| if count is not None: |
| body["requested_units"] = count |
| quote = call("POST", "/v1/capacity/quotes", body) |
| if not quote.get("quote_id"): |
| return "SpecStream could not price this request yet." |
| cents = int(quote.get("maximum_cents") or 0) |
| money = dollars(cents, str(quote.get("currency") or "USD")) |
| return ( |
| "SpecStream read it as: %s\n" |
| "Maximum price: %s. You pay for each result delivered; a search that finds nothing is free.\n" |
| "quote_id: %s\n" |
| "confirm_max_cents: %d\n" |
| "Nothing has run. Ask the person to approve %s. Only after a yes, call this tool again " |
| "with quote_id and confirm_max_cents." % (sentence, money, quote["quote_id"], cents, money) |
| ) |
| if not quote_id or confirm_max_cents is None: |
| return "To run it, give both quote_id and confirm_max_cents. Nothing ran." |
| quoted = call("GET", "/v1/capacity/quotes/" + urlquote(quote_id, safe="")) |
| if quoted.get("quote_id") != quote_id: |
| return "That quote was not found. Nothing ran." |
| cents = int(quoted.get("maximum_cents") or 0) |
| if int(confirm_max_cents) != cents: |
| return "confirm_max_cents must equal the quoted maximum, %d. Nothing ran." % cents |
| started = call( |
| "POST", |
| "/v1/capacity/executions", |
| { |
| "quote_id": quote_id, |
| "quote_hash": quoted.get("quote_hash"), |
| "idempotency_key": "smolagents-" + quote_id, |
| "confirm_max_cents": cents, |
| }, |
| ) |
| find_id = started.get("find_id") |
| if not find_id: |
| return "SpecStream accepted the approval but has not started the work yet. Call again in a moment." |
| return "Started. find_id: %s. Read the rows with specstream_get_rows; the run takes a few minutes." % find_id |
| except RuntimeError as exc: |
| return "Nothing ran. %s" % exc |
|
|
| def __init__(self, *args, **kwargs): |
| self.is_initialized = False |
|
|
