File size: 6,344 Bytes
ff819b8 e9bc357 ff819b8 e9bc357 ff819b8 | 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 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 | 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
|