> SpecStream is the live web data engine for teams and agents. This is the Markdown copy of https://specstream.ai/docs/integrations/livekit-agents-deep-search. Every page, with what SpecStream does and costs: https://specstream.ai/llms.txt # LiveKit Agents Deep Search Give a LiveKit voice agent Deep Search as three function tools. The caller describes the rows they need, the agent reads back the request and the most it can cost, and after a yes SpecStream returns every match on the live web as a table. Source URL and read time on every value. [SpecStream for LiveKit Agents Deep SearchAgent frameworks & builders](https://specstream.ai/integrations/livekit-agents-deep-search.md) ## What you need - Python 3.10 or newer, the LiveKit CLI (`lk`) and a LiveKit Cloud project. The steps start from LiveKit's `agent-starter-python` project, which pins `livekit-agents~=1.8`. - A SpecStream API key with a total spending limit. New accounts start with $10 of credit, no card. [Create an account](https://specstream.ai/signup?utm_source=livekit-agents-deep-search). 1. 1 ## 1. Create a SpecStream API key In the SpecStream workspace, open [API keys](https://specstream.ai/workspace/keys?utm_source=livekit-agents-deep-search) and click **New key**. Name it for the agent and choose **Call the APIs** under **What the key can do**: it holds **Authorize paid work**, which starting a search needs. Enter a **Total spending limit**, the most the agent may spend across all its calls, then click **Create key**. Copy the key from **Copy your key now**, because it is shown once. The tools run in your agent process, so the key never reaches the room, the caller or the model. 2. 2 ## 2. Create the agent project Create a project from LiveKit's Python starter, install its dependencies when the CLI asks, and add `httpx`, the async HTTP client the tools use. Then put the key in the project's `.env.local`, beside the LiveKit keys the CLI wrote there: Terminal ```bash lk agent init my-agent --template agent-starter-python cd my-agent uv add httpx echo 'SPECSTREAM_API_KEY=YOUR_SPECSTREAM_API_KEY' >> .env.local ``` 3. 3 ## 3. Write the Deep Search tools `@function_tool` turns each function into a tool, and LiveKit sends its docstring to the model as the tool description. Deep Search takes three tools, because a search runs longer than one turn of the conversation: `price_deep_search` sends the request to `POST /v1/parse`, which returns `read_back`, the request in plain words, and `ast_hash`. It then asks `POST /v1/capacity/quotes` for the price of that many rows. The quote's `maximum_cents` is the most the search can cost. Pricing spends nothing, and the quote stays open for 30 minutes. `start_deep_search` approves the quote with `POST /v1/capacity/executions`. The amount comes from the quote your code kept, never from the model's arguments, and the quote's own ID is the request key, so a repeated call cannot start a second search. It returns at once and leaves a background task that checks the search every 10 seconds and has the agent tell the caller when the table is ready. `read_deep_search` reads the first 10 rows from `GET /v1/finds/{id}/rows` with `format=markdown`: one table with a column for each fact the caller asked for and a **Source** column. A failed call raises `ToolError`, and LiveKit passes its message to the model, so the agent tells the caller why the call failed and the session keeps running. Save this as `src/specstream_deep_search.py`: src/specstream\_deep\_search.py ```python import asyncio import os import httpx from livekit.agents import RunContext, function_tool from livekit.agents.llm import ToolError specstream = httpx.AsyncClient(base_url="https://api.specstream.ai", timeout=60) quotes = {} # quote_id -> quote: the amount to approve comes from SpecStream, never from the model watchers = set() # background tasks that tell the caller when a table is ready DONE = {"completed", "partial", "failed", "canceled", "expired"} async def call(method: str, path: str, **kwargs) -> dict: headers = {"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"} r = await specstream.request(method, path, headers=headers, **kwargs) if r.is_error: try: message = r.json()["error"]["message"] except (ValueError, KeyError, TypeError): message = r.text[:300] raise ToolError(f"SpecStream answered {r.status_code}: {message}") return r.json()["data"] @function_tool async def price_deep_search(context: RunContext, request: str, rows: int) -> dict: """Price a table of rows from the live web: companies, places, people or other records that meet the caller's conditions, with the facts to bring back. Spends nothing. Read back how SpecStream understood the request and the maximum price, then ask the caller to approve it. Args: request: What each row is, the conditions it must meet and the facts to bring back, in the caller's words. rows: How many rows the caller wants, 1 to 100. """ reading = await call("POST", "/v1/parse", json={"q": request}) if reading["status"] != "runnable": return { "status": reading["status"], "read_back": reading.get("read_back"), "next": "Ask the caller for the missing detail, then price again.", } rows = max(1, min(int(rows), 100)) # the row limit holds even if the model asks for more quote = await call( "POST", "/v1/capacity/quotes", json={"ast_hash": reading["ast_hash"], "operation": "find", "requested_units": rows}, ) quotes[quote["quote_id"]] = quote return { "read_back": reading["read_back"], "rows": rows, "maximum_usd": quote["maximum_cents"] / 100, "quote_id": quote["quote_id"], } @function_tool async def start_deep_search(context: RunContext, quote_id: str) -> str: """Start a priced Deep Search. Call it only after the caller said yes to the maximum price you read out. The rows arrive later: tell the caller the search started and keep talking. Args: quote_id: The quote_id that price_deep_search returned. """ quote = quotes.pop(quote_id, None) if quote is None: raise ToolError("Unknown or used quote. Price the request again.") run = await call( "POST", "/v1/capacity/executions", json={ "quote_id": quote["quote_id"], "quote_hash": quote["quote_hash"], "confirm_max_cents": quote["maximum_cents"], "idempotency_key": quote["quote_id"], }, ) task = asyncio.create_task(tell_when_done(context.session, run["find_id"])) watchers.add(task) task.add_done_callback(watchers.discard) return f"The search started. Its find_id is {run['find_id']}." @function_tool async def read_deep_search(context: RunContext, find_id: str) -> str: """Read the first rows of a Deep Search started in this call, as a table with a Source column. Args: find_id: The find_id that start_deep_search returned. """ find = await call("GET", f"/v1/finds/{find_id}") page = await call("GET", f"/v1/finds/{find_id}/rows", params={"format": "markdown", "limit": 10}) return f"Status: {find['status']}. Rows so far: {page['accepted_count']}.\n\n{page['table']}" async def tell_when_done(session, find_id: str) -> None: while True: await asyncio.sleep(10) try: find = await call("GET", f"/v1/finds/{find_id}") except (ToolError, httpx.HTTPError): continue # a failed status check is tried again on the next pass if find["status"] in DONE: break await session.generate_reply( instructions=( f"Tell the caller the table for find {find_id} is ready with " f"{find['accepted_count']} rows, and offer to read the first ones." ) ) ``` 4. 4 ## 4. Give the tools to the agent In `src/agent.py`, import the three tools, add the rules below to the end of the starter's instructions, and pass the tools in `tools`. The rules make the agent read the price aloud and wait for a yes before it starts anything: src/agent.py ```python from specstream_deep_search import price_deep_search, read_deep_search, start_deep_search DEEP_SEARCH_RULES = """ # Deep Search - When the caller wants many records with the same facts on each, call price_deep_search with their request and the number of rows. - Read back how the request was understood and the maximum price in dollars, and wait for a clear yes. - Call start_deep_search only after that yes. Never start a search the caller did not approve. - When the table is ready, call read_deep_search and speak the first rows in short sentences, naming the site each row was read on. """ class Assistant(Agent): def __init__(self) -> None: super().__init__( llm=inference.LLM(model="google/gemma-4-31b-it"), instructions=textwrap.dedent( """\ ...the starter's instructions, unchanged... """ ) + DEEP_SEARCH_RULES, tools=[price_deep_search, start_deep_search, read_deep_search], ) ``` 5. 5 ## 5. Run it and ask Start the agent with `lk agent dev` and talk to it in the [Agent Console (opens in a new tab)](https://docs.livekit.io/testing/agent-console). Ask for rows and say how many. The agent reads back the request and the maximum price. Say yes, and it starts the search and keeps talking. When the table is ready the agent says so, and asking for the rows makes it call `read_deep_search`. Spoken to the agent ```text Find 20 coffee roasters in Portland, Oregon that sell wholesale, with their website and the city they roast in. ``` ## What the agent gets back Deep Search turns one request in plain words into rows from the live web: companies, local businesses, people, grants, tenders, papers and other records, each with the facts the caller asked for. Every row meets the conditions in the request. `read_deep_search` hands the model a Markdown table. Its **Source** column holds the page most of the row's values were read on, and a value read on another page carries its own link, so the agent can name the site behind each row it speaks. The model reads the table and the caller hears sentences. For the full rows in your own code, call `GET /v1/finds/{id}/rows` without `format`: each row's `cells` hold the column's `label`, the value as `display`, and a `provenance` with `source_url` and `observed_at`. `format=csv` returns the same page as one CSV text. Follow `next_cursor` for the rows past the first page. ## Why the search runs in the background A Deep Search can run for minutes, and up to four hours on the largest lists. A LiveKit tool that waited for it would hold the turn, so `start_deep_search` returns as soon as the search starts, and the caller can keep talking. The background task keeps the `session` from the tool's `RunContext`. When the search ends it calls `session.generate_reply`, and the agent tells the caller the table is ready. The task set keeps a reference to each running check until it ends. ## How spending stays under control - Pricing spends nothing. `price_deep_search` only reads the request and returns a quote. - The caller approves the price. The rules in the instructions make the agent read the maximum aloud and wait for a yes before `start_deep_search`. - The model cannot change the amount. `confirm_max_cents` comes from the quote your code kept, and a search never costs more than it. - The row limit holds in code. `price_deep_search` caps `rows` at 100 whatever number the model sends. - The key has a ceiling. Its total spending limit caps what the agent can spend across all calls, and a call over it is refused with `403`. ## If something doesn't work | What you see | What to do | | --- | --- | | `price_deep_search` returns a status other than `runnable` | The request needs more detail, such as the place or the facts to bring back. The agent asks the caller and prices it again. | | The agent says `Unknown or used quote` | The quote was used, or the agent process restarted. Price the request again. A quote also expires after 30 minutes. | | `SpecStream answered 402` | The balance is too low for the approved maximum. Add balance in your SpecStream workspace. | | `SpecStream answered 403` | The key has no **Authorize paid work**, or the search would go over its total spending limit. Raise the limit under **API keys**. | | `SpecStream answered 401` | The key is unknown, expired or revoked. Create a new key and set it as `SPECSTREAM_API_KEY` in `.env.local`. | | `KeyError: 'SPECSTREAM_API_KEY'` | The agent process has no key. Check the line in `.env.local`, or set the variable where the agent server runs. | ## Cost Pricing a request, reading rows and checking a search's status cost nothing. A search is charged per delivered row: 5¢, 15¢ or $1 a row depending on how much each row needs, plus 5¢ for a person's email and 15¢ for a phone when the request asks for them. Duplicates and rows a page contradicts are not charged, a search that finds nothing costs nothing, and no search costs more than the maximum the caller approved. New accounts start with $10 of credit, no card. The rates are on [pricing](https://specstream.ai/pricing.md), and [balance and billing](https://specstream.ai/docs/billing.md) covers quotes. ## Web search and page reading too For one fact in the middle of a call, the [LiveKit Agents web search guide](https://specstream.ai/docs/integrations/livekit-agents.md) adds `search_web` and `read_page` tools on the same key. The [Search API](https://specstream.ai/docs/search-api.md) is $5 per 1,000 searches of up to 10 results and the [Extract API](https://specstream.ai/docs/extract-api.md) $0.80 per 1,000 pages, with no quote and no approval. The [REST API reference](https://specstream.ai/docs/api.md) lists every call on this page. ## Remove Take the three tools out of the agent's `tools` and the rules out of its instructions. Then open **API keys** in your SpecStream workspace and revoke the key, which cuts the agent off at once. Rows already delivered stay in your account. Browse the other [tools SpecStream connects to](https://specstream.ai/docs/integrations.md), or write to [support@specstream.ai](mailto:support@specstream.ai) if a step here doesn't match what you see. ## More [SpecStream for LiveKit Agents Deep Search](https://specstream.ai/integrations/livekit-agents-deep-search.md) covers what it does in LiveKit Agents Deep Search on one page. Every other tool has its guide in [Connect your tools](https://specstream.ai/docs/integrations.md). You see the most a request can cost before it starts, and the matches are paid from your balance. [Billing](https://specstream.ai/docs/billing.md) has the details. For help, email [support@specstream.ai](mailto:support@specstream.ai).