LangGraph

Give a LangGraph agent the live web. The tools from langchain-specstream go into a ToolNode, so your graph searches with SpecStream's Search API, reads pages with the Extract API and asks Deep Search for whole lists, and it can stop for your yes before a paid list runs.

SpecStream for LangGraphAgent frameworks & builders

What you need

Python 3.10 or later, with LangGraph 1.x.
A SpecStream API key, made under Agents & API in your workspace (more on keys). New accounts get $10 of credit, no card.
A chat model that calls tools. The examples use init_chat_model from langchain; any LangChain chat model with bind_tools works.

1. Install the packages

langchain-specstream reads your key from SPECSTREAM_API_KEY. langchain-anthropic is there only for the model in these examples; use your own provider's package in its place.

Shell
pip install -U langgraph langchain langchain-anthropic langchain-specstream
export SPECSTREAM_API_KEY="YOUR_SPECSTREAM_API_KEY"
export ANTHROPIC_API_KEY="YOUR_ANTHROPIC_API_KEY"

2. Build the graph

specstream_tools(max_price_usd=5) returns four LangChain tools that share one key. Bind them to the model, put the same list in a ToolNode, and let tools_condition send each tool call to that node and back to the model. When the model answers without a tool call, the graph ends.

max_price_usd is the most one list may cost, in dollars. It lives on the tool in your code, so the model never sees it and can't raise it.

graph.py
from langchain.chat_models import init_chat_model
from langgraph.graph import START, MessagesState, StateGraph
from langgraph.prebuilt import ToolNode, tools_condition
from langchain_specstream import specstream_tools

tools = specstream_tools(max_price_usd=5)
model = init_chat_model("anthropic:claude-sonnet-5-5").bind_tools(tools)


def call_model(state: MessagesState):
    return {"messages": [model.invoke(state["messages"])]}


builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", tools_condition)
builder.add_edge("tools", "model")
graph = builder.compile()

3. Run it

The model picks the tool. A question with one answer goes to specstream_web_search, a page it wants to read goes to specstream_extract_pages, and a request for many records that meet conditions goes to specstream_find_list. Every tool returns JSON, with the page each value was read from.

graph.py
result = graph.invoke({"messages": [{"role": "user", "content":
    "What did the Federal Reserve decide at its latest meeting? Name the page you read it on."}]})

print(result["messages"][-1].content)

4. Pause before a paid list

A search and a page read cost a fraction of a cent and run at once. A list is priced per row, so you may want a person to see it first. Route calls to specstream_find_list to their own node, compile with a checkpointer and interrupt_before=["paid_tools"], and the graph stops there with its state saved.

graph.get_state(config).next shows where it stopped, and the last message holds the request and max_results the model chose. graph.invoke(None, config) continues from that point. To say no, don't resume, or update the state with a reply from the user.

graph.py
from typing import Literal

from langgraph.checkpoint.memory import InMemorySaver
from langgraph.graph import END


def route(state: MessagesState) -> Literal["tools", "paid_tools", "__end__"]:
    calls = state["messages"][-1].tool_calls
    if not calls:
        return END
    if any(call["name"] == "specstream_find_list" for call in calls):
        return "paid_tools"
    return "tools"


builder = StateGraph(MessagesState)
builder.add_node("model", call_model)
builder.add_node("tools", ToolNode(tools))
builder.add_node("paid_tools", ToolNode(tools))
builder.add_edge(START, "model")
builder.add_conditional_edges("model", route)
builder.add_edge("tools", "model")
builder.add_edge("paid_tools", "model")
graph = builder.compile(checkpointer=InMemorySaver(), interrupt_before=["paid_tools"])

config = {"configurable": {"thread_id": "lists-1"}}
graph.invoke({"messages": [{"role": "user", "content":
    "Find breweries in Portland, Oregon that host events, with their website."}]}, config)

state = graph.get_state(config)
print(state.next)                                # ('paid_tools',)
print(state.values["messages"][-1].tool_calls)   # the request and max_results the model chose

graph.invoke(None, config)                       # your yes: the list is priced and runs within max_price_usd

The four tools

SpecStream's Search API is the most accurate web search API in the world on SimpleQA Verified and WideSearch (benchmarks). Each answer is read from the live web when the call is made.

Tool	The model sends	What comes back	Cost
`specstream_web_search`	`q`, and `passages` for page text	`kind`, the `answer` with the page it was read from and its read time, matching `rows` and the ranked `sources`	$5 per 1,000 searches of up to 10 results
`specstream_extract_pages`	`urls` (1 to 20) and `format`	Each page's main text, title and published date, from web pages, PDF, Word and Excel	$0.80 per 1,000 pages; a page with no text is free
`specstream_find_list`	`request` and `max_results`	Every matching row, each with its `source_url` and `checked_at`, and a `list_link` to the list in your workspace	Per matching row, never above `max_price_usd`
`specstream_get_list_results`	`list_id`	Every row found so far for a list that was still running	Free

When a list costs more than the cap

specstream_find_list prices the request before anything runs. Above max_price_usd, nothing starts and the tool returns the price with over_maximum, so the model can ask for fewer rows or tell the user. A request too vague to price ("clinics") comes back as needs_edit, with how SpecStream read it, so the model can add the place or the conditions.

Long lists and retried nodes

specstream_find_list waits up to seven minutes for the list (wait_seconds, 420 by default), then returns the rows found so far and the list_id. The model can call specstream_get_list_results with that ID later, at no cost.

A node that is retried and asks for the same list within the hour reaches the list it already started, so the rows are paid for once.

Every SpecStream tool over MCP

The package covers search, page reading and lists. For schedules, exports and enriching a list you already hold, load the whole MCP server at https://api.specstream.ai/mcp with langchain-mcp-adapters and pass its tools to the same ToolNode. The client code is on the LangChain page.

Cost

A search is $5 per 1,000 searches of up to 10 results, and a page read is $0.80 per 1,000 pages, with no charge for a page that has no text. Both come from your balance with no quote. A list is billed only for the rows that match, never above the cap you set. New accounts start with $10 of credit, no card. See how billing works, and send questions to [email protected].

SpecStream for LangGraph covers what it does in LangGraph 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 the matches are paid from your balance. Billing has the details. For help, email [email protected].