Haystack

Ask for live web data in plain words and get it back as Haystack Documents, one per row, each with the page it was read from and when. The specstream-haystack package gives a pipeline a component and an agent a tool, and you set the most a request may cost in your own code.

SpecStream for HaystackAgent frameworks & builders

What you need

Python 3.10 or later and Haystack 2.14 or later.
A SpecStream account. Starting is free: new accounts get $10 of work to use within 30 days, with no card.
A SpecStream API key with a spending limit.

1. Install the package

Install specstream-haystack where your pipelines run. It pulls in haystack-ai if you don't have it yet.

Terminal
pip install -U specstream-haystack

2
2. Create an API key
In your SpecStream workspace, open Agents & API, go to the API keys tab, and click Create API key. Label it for the project, such as "Haystack pipelines", and keep the permissions that are already on, including Authorize paid work.
Click Set a total limit and enter the most this key may spend, across every pipeline that uses it. A key lasts 30 days unless you change Expires after, days, which goes up to 365. Copy the key from the next screen. SpecStream shows it once.
The components read the key from SPECSTREAM_API_KEY. To keep it under another name, pass api_key=Secret.from_env_var("YOUR_NAME"), using Haystack's Secret. Set the variable in the environment where the pipeline runs:
Terminal
export SPECSTREAM_API_KEY="ss_live_..."

3. Add the fetcher to a pipeline

SpecStreamListFetcher takes a request in plain words and returns Documents. max_price_usd is required, and a request starts only when SpecStream's price for it is at or below that amount.

This pipeline passes the rows to a chat model with the source page beside each one. Use any chat generator in place of OpenAIChatGenerator:

pipeline.py
from haystack import Pipeline
from haystack.components.builders import ChatPromptBuilder
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack_integrations.components.fetchers.specstream import SpecStreamListFetcher

template = [
    ChatMessage.from_user(
        "Use only these rows to answer.\n"
        "{% for doc in documents %}"
        "{{ doc.content }} (source: {{ doc.meta.source_url }})\n"
        "{% endfor %}"
        "Question: {{ question }}"
    )
]

pipeline = Pipeline()
pipeline.add_component("fetcher", SpecStreamListFetcher(max_price_usd=3, max_results=25))
pipeline.add_component("prompt", ChatPromptBuilder(template=template, required_variables=["documents", "question"]))
pipeline.add_component("llm", OpenAIChatGenerator(model="gpt-4o"))
pipeline.connect("fetcher.documents", "prompt.documents")
pipeline.connect("prompt.prompt", "llm.messages")

result = pipeline.run({
    "fetcher": {"request": "Breweries in Portland, Oregon that host events, with their website"},
    "prompt": {"question": "Summarize each brewery in one line, with its website."},
})
print(result["llm"]["replies"][0].text)

4. Give an agent the tool

SpecStreamListTool wraps the fetcher as a tool named specstream_find_list. The model writes the request and picks how many rows. The price limit stays in your code, out of the model's reach.

The tool hands the model each row as text with its source page after it, so the agent can cite what it found:

agent.py
from haystack.components.agents import Agent
from haystack.components.generators.chat import OpenAIChatGenerator
from haystack.dataclasses import ChatMessage
from haystack_integrations.tools.specstream import SpecStreamListTool

agent = Agent(
    chat_generator=OpenAIChatGenerator(model="gpt-4o"),
    tools=[SpecStreamListTool(max_price_usd=5)],
)

result = agent.run(
    messages=[ChatMessage.from_user("List food banks in Texas with their latest revenue, and link the source for each.")]
)
print(result["last_message"].text)

What comes back

Each row becomes one Document. content is the row as JSON, with a key for each column of the request. Ask for dental clinics with a phone number and the keys are the clinic and its phone number. A value the page doesn't state comes back as an empty string.

run also returns list_id, status, list_link, and a message that is empty unless the list was still running when the wait ended.

meta holds these keys. If a column has the same name as one of them, its name gets "(column)" added so it can't replace the real value:

Key	What it holds
`source_url`	The page the row was read from.
`checked_at`	When SpecStream read that page.
`list_id`	The list the row belongs to.
`list_link`	A link to the list in your workspace.
`position`	The row's place in the list.
`request`	The request you sent.
`fields`	The row's columns as a dict.

Price limits and retries

max_price_usd lives in your code. The model can't read it or raise it.
SpecStream prices the request before it starts anything. When the most it could cost is above max_price_usd, nothing runs, nothing is charged, and SpecStreamError says so.
You pay for rows that match. A request that finds nothing is free.
A retried step doesn't pay twice. The same request with the same max_results and max_price_usd, sent again within the hour, reaches the list that already started.

When a request takes longer

A list can take up to about five minutes. run reads the rows every poll_interval seconds until the list ends or wait_seconds runs out. If the wait runs out first, run returns the rows found so far, sets status to "running", and fills message. Send the same request again within the hour to pick the list back up, or open list_link to see the rest in your workspace.

The statuses "completed" and "partial" mean the request is done. "failed", "canceled", and "expired" also end the wait.

When a request can't run

SpecStreamError covers these cases:

The request is too vague to price. The error quotes how SpecStream read it. Add the place or more detail and run it again.
The price is above max_price_usd. Lower max_results or raise the limit.
There is no key. The error asks you to set SPECSTREAM_API_KEY.
SpecStream refuses a call, for example because the key expired. The error starts with "SpecStream answered", then the status code and SpecStream's own message.
SpecStream can't be reached. The error says "Couldn't reach SpecStream" and why.

Settings

SpecStreamListFetcher takes these arguments. SpecStreamListTool takes the same ones, plus name, which defaults to "specstream_find_list", and description.

A pipeline that holds the fetcher saves and loads with Haystack's dumps and loads. The YAML holds the name of the environment variable, never the key.

Argument	Default	What it does
`max_price_usd`	Required	The most one request may cost, in US dollars. Must be above 0.
`api_key`	`SPECSTREAM_API_KEY`	A Haystack `Secret` that holds your API key.
`max_results`	25	How many matching rows to look for. `run` can override it for one call.
`wait_seconds`	420	How long `run` waits for the list to finish.
`poll_interval`	3.0	Seconds between reads while it waits.
`locale`	"en-US"	The language the request is read in.
`base_url`	"https://api.specstream.ai"	The SpecStream API address.

Cost

The package is free and MIT licensed. SpecStream charges your balance for matching rows only, from 10 cents each, and the limit in your code caps every request. Rates are on the pricing page.

Remove

Run pip uninstall specstream-haystack and take the component out of your pipelines. To stop the key working everywhere, delete it under Agents & API. Lists you already ran stay in your workspace.

Source: GitHub (opens in a new tab). Package: PyPI (opens in a new tab). Other tools to connect are in Connect your tools. Questions about this package go to [email protected].

SpecStream for Haystack covers what it does in Haystack on one page. Every other tool has its guide in Connect your tools. You see the most a request can cost before it starts, and only the matches come out of your balance. Billing has the details. For help, email [email protected].