> SpecStream is the live web data engine for teams and agents. This is the Markdown copy of https://specstream.ai/docs/integrations/deepgram-voice-agent. Every page, with what SpecStream does and costs: https://specstream.ai/llms.txt # Deepgram Voice Agent Give a Deepgram voice agent Deep Search as a client side function. The caller describes the rows they need, the agent reads back the request and the most it can cost, and after the caller says yes, SpecStream returns every match on the live web as structured rows. Source URL and read time on every value. [SpecStream for Deepgram Voice AgentAgent frameworks & builders](https://specstream.ai/integrations/deepgram-voice-agent.md) ## What you need - A Deepgram API key, and Python 3.10 or newer with websockets and httpx (`pip install "websockets>=14" httpx`). - 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=deepgram-voice-agent). 1. 1 ## 1. Create a SpecStream API key In the SpecStream workspace, open [API keys](https://specstream.ai/workspace/keys?utm_source=deepgram-voice-agent) 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. Set it as `SPECSTREAM_API_KEY` on the server that runs the agent, beside `DEEPGRAM_API_KEY`. The key stays on your server: the functions run in your code, and Deepgram never sees it. 2. 2 ## 2. Define the Deep Search functions Deepgram reads your functions from `agent.think.functions` in the `Settings` message. A function without an `endpoint` runs client side: Deepgram sends your code a `FunctionCallRequest`, and your code answers with a `FunctionCallResponse`. Deep Search takes three functions. `price_deep_search` reads the request and prices it, and spends nothing. `start_deep_search` starts the priced search, and carries `defer_until_eot: true`, so it is never sent while the caller is still talking. `read_deep_search` reads the rows found so far. The `rows` parameter is the row limit: the JSON schema caps it at 100, and the code below caps it again. Put each block of this guide in one file, `deep_search_agent.py`, in order: deep\_search\_agent.py ```python import asyncio import json import os import httpx import websockets FUNCTIONS = [ { "name": "price_deep_search", "description": ( "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." ), "parameters": { "type": "object", "properties": { "request": { "type": "string", "description": "What each row is, the conditions it must meet and the facts to bring back, in the caller's words", }, "rows": { "type": "integer", "minimum": 1, "maximum": 100, "description": "How many rows the caller wants, 1 to 100", }, }, "required": ["request", "rows"], }, }, { "name": "start_deep_search", "description": ( "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." ), "parameters": { "type": "object", "properties": { "quote_id": {"type": "string", "description": "The quote_id that price_deep_search returned"}, }, "required": ["quote_id"], }, "defer_until_eot": True, }, { "name": "read_deep_search", "description": "Read the first rows of a Deep Search started in this call, and how many rows it has so far.", "parameters": { "type": "object", "properties": { "find_id": {"type": "string", "description": "The find_id that start_deep_search returned"}, }, "required": ["find_id"], }, }, ] ``` 3. 3 ## 3. Call SpecStream from each function Each function is a few calls to the SpecStream REST API. `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, and the quote stays open for 30 minutes. `start_deep_search` approves that 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: deep\_search\_agent.py ```python specstream = httpx.AsyncClient( base_url="https://api.specstream.ai", headers={"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"}, timeout=60, ) quotes = {} # quote_id -> quote: the amount to approve comes from SpecStream, never from the model DONE = {"completed", "partial", "failed", "canceled", "expired"} async def price_deep_search(request: str, rows: int) -> dict: r = await specstream.post("/v1/parse", json={"q": request}) r.raise_for_status() reading = r.json()["data"] 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 ignores the schema r = await specstream.post( "/v1/capacity/quotes", json={"ast_hash": reading["ast_hash"], "operation": "find", "requested_units": rows}, ) r.raise_for_status() quote = r.json()["data"] quotes[quote["quote_id"]] = quote return { "read_back": reading["read_back"], "rows": rows, "maximum_usd": quote["maximum_cents"] / 100, "quote_id": quote["quote_id"], } async def start_deep_search(quote_id: str) -> dict: quote = quotes.pop(quote_id, None) if quote is None: return {"error": "Unknown or used quote. Price the request again."} r = await specstream.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"], }, ) r.raise_for_status() return {"find_id": r.json()["data"]["find_id"], "status": "running"} async def read_deep_search(find_id: str) -> dict: find = (await specstream.get(f"/v1/finds/{find_id}")).json()["data"] r = await specstream.get(f"/v1/finds/{find_id}/rows", params={"format": "markdown", "limit": 10}) r.raise_for_status() page = r.json()["data"] return {"status": find["status"], "rows_so_far": page["accepted_count"], "first_rows": page["table"]} ``` 4. 4 ## 4. Answer each FunctionCallRequest Each call in `functions` has an `id`, a `name` and `arguments` as a JSON string. Run the matching function in its own task, so the socket keeps carrying audio, and send back a `FunctionCallResponse` with the same `id` and `name` and the result as a string in `content`. A call that comes with `thought_signature` (some Gemini models send one) gets it back unchanged. A search runs longer than a turn. When it ends, `tell_when_done` sends an `InjectAgentMessage` with `behavior` set to `queue`, so the agent says the table is ready as soon as the current turn ends: deep\_search\_agent.py ```python HANDLERS = { "price_deep_search": price_deep_search, "start_deep_search": start_deep_search, "read_deep_search": read_deep_search, } running = {} # FunctionCallRequest id -> its task, so a cancelled call can be stopped async def answer(ws, call: dict): try: result = await HANDLERS[call["name"]](**json.loads(call["arguments"])) except httpx.HTTPStatusError as e: result = {"error": e.response.text[:500]} except httpx.HTTPError as e: result = {"error": str(e)} finally: running.pop(call["id"], None) reply = {"type": "FunctionCallResponse", "id": call["id"], "name": call["name"], "content": json.dumps(result)} if call.get("thought_signature"): reply["thought_signature"] = call["thought_signature"] await ws.send(json.dumps(reply)) if call["name"] == "start_deep_search" and "find_id" in result: asyncio.create_task(tell_when_done(ws, result["find_id"])) async def tell_when_done(ws, find_id: str): while True: await asyncio.sleep(10) find = (await specstream.get(f"/v1/finds/{find_id}")).json()["data"] if find["status"] in DONE: break await ws.send(json.dumps({ "type": "InjectAgentMessage", "behavior": "queue", "message": f"Your table is ready with {find['accepted_count']} rows. Shall I read the first ones?", })) ``` 5. 5 ## 5. Connect and talk to it Open the socket with your Deepgram key as `Token` in the `Authorization` header and send `Settings` first. This test types to the agent with `InjectUserMessage` and prints `ConversationText`, so it runs without a microphone. In your own agent, stream the microphone as Deepgram's guide shows, and drop `keep_alive` and `type_to_agent`. When `FunctionCallCancelled` arrives, the matching task stops and sends no response: deep\_search\_agent.py ```python SETTINGS = { "type": "Settings", "audio": { "input": {"encoding": "linear16", "sample_rate": 16000}, "output": {"encoding": "linear16", "sample_rate": 16000, "container": "none"}, }, "agent": { "language": "en", "listen": {"provider": {"type": "deepgram", "model": "nova-3"}}, "think": { "provider": {"type": "open_ai", "model": "gpt-4o-mini"}, "prompt": ( "You help callers get tables of data from the live web with Deep Search. " "Before you call start_deep_search, read back the request and the maximum price, " "and wait for the caller to say yes. Never start a search the caller did not approve." ), "functions": FUNCTIONS, }, "speak": {"provider": {"type": "deepgram", "model": "aura-2-thalia-en"}}, "greeting": "Hello. What data do you need?", }, } async def keep_alive(ws): while True: # no microphone in this test, so the socket needs a KeepAlive every 8 seconds await asyncio.sleep(8) await ws.send(json.dumps({"type": "KeepAlive"})) async def type_to_agent(ws): while True: text = await asyncio.to_thread(input, "you> ") await ws.send(json.dumps({"type": "InjectUserMessage", "content": text})) async def main(): async with websockets.connect( "wss://agent.deepgram.com/v1/agent/converse", additional_headers={"Authorization": f"Token {os.environ['DEEPGRAM_API_KEY']}"}, ) as ws: await ws.send(json.dumps(SETTINGS)) async for message in ws: if isinstance(message, bytes): continue # the agent's voice; play it as Deepgram's Build a Voice Agent guide does event = json.loads(message) if event["type"] == "SettingsApplied": asyncio.create_task(keep_alive(ws)) asyncio.create_task(type_to_agent(ws)) elif event["type"] == "ConversationText": print(f"{event['role']}: {event['content']}") elif event["type"] == "FunctionCallRequest": for call in event["functions"]: if call["client_side"]: running[call["id"]] = asyncio.create_task(answer(ws, call)) elif event["type"] == "FunctionCallCancelled": for call in event["functions"]: task = running.pop(call["id"], None) if task: task.cancel() asyncio.run(main()) ``` 6. 6 ## 6. Run it Set `DEEPGRAM_API_KEY` and `SPECSTREAM_API_KEY` and run `python deep_search_agent.py`. Type the rows you want, with how many. The agent reads back the request and the maximum price. Type yes, and it starts the search. When the table is ready, the agent says so, and asking for the rows makes it call `read_deep_search`. ## What the caller gets 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, and each value carries the URL of the page it was read on and when it was read. `read_deep_search` gives the agent the first 10 rows as a Markdown table to speak from. For the full rows, 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`. Follow `next_cursor` for the rows past the first page. ## Why the functions run in your code Deepgram can also call a function on a server you name, when the function carries an `endpoint` with `url`, `method` and `headers`. Deep Search needs three calls in order and a quote kept between two turns of the conversation, so these functions run client side. Your SpecStream key stays in your process, and the approved amount comes from the quote your code holds. ## 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 prompt tells the agent to read the maximum aloud and wait for a yes, and `defer_until_eot` holds `start_deep_search` until the caller has finished speaking. - 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 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. | | `start_deep_search` returns `Unknown or used quote` | The quote was used, or the process restarted. Price the request again. A quote also expires after 30 minutes. | | An error with `402` | The balance is too low for the approved maximum. Add balance in your SpecStream workspace. | | An error with `403` | The key has no **Authorize paid work**, or the search would go over its total spending limit. Raise the limit under **API keys**. | | An error with `401` | The key is unknown, expired or revoked. Create a new key and set it as `SPECSTREAM_API_KEY`. | | The agent never answers after a function call | Deepgram waits for a `FunctionCallResponse`. Check that every call path sends one with the call's `id`. | ## Cost Pricing a request, reading rows and reading 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, add a fourth function that calls the [Search API](https://specstream.ai/docs/search-api.md), `GET /v1/web-search?q=`, at $5 per 1,000 searches of up to 10 results, with no quote and no approval. To read a page the caller names, the [Extract API](https://specstream.ai/docs/extract-api.md) is $0.80 per 1,000 pages. The [REST API reference](https://specstream.ai/docs/api.md) lists every call above. ## Remove Take the three functions out of `agent.think.functions`. 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 Deepgram Voice Agent](https://specstream.ai/integrations/deepgram-voice-agent.md) covers what it does in Deepgram Voice Agent 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).