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

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.

1
1. Create a SpecStream API key
In the SpecStream workspace, open API keys 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. 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
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. 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
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. 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
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. 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
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. 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, and balance and billing 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, 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 is $0.80 per 1,000 pages. The REST API reference 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, or write to [email protected] if a step here doesn't match what you see.

SpecStream for Deepgram Voice Agent covers what it does in Deepgram Voice Agent 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].