> 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-web-search. Every page, with what SpecStream does and costs: https://specstream.ai/llms.txt # Deepgram Voice Agent web search Give a Deepgram voice agent live web search. Define `search_web` in `agent.think.functions` and answer it from your own code with the SpecStream Search API. The agent speaks the answer and names the site that printed it. [SpecStream for Deepgram Voice Agent web searchAgent frameworks & builders](https://specstream.ai/integrations/deepgram-voice-agent-web-search.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-web-search). 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-web-search) and click **New key**. Name it for the agent and choose **Call the APIs** under **What the key can do**. A key with that choice can search and read pages, which the two functions need. 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 functions run in your code, so the key stays on your server and Deepgram never sees it. 2. 2 ## 2. Define search\_web in the Settings message 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`. `search_web` takes one parameter, `q`, the question in everyday words. `read_page` takes a `url`, for a page the caller names or one a search returned. Put each block of this guide in one file, `web_search_agent.py`, in order: web\_search\_agent.py ```python import asyncio import json import os import httpx import websockets FUNCTIONS = [ { "name": "search_web", "description": ( "Search the live web for a fact the caller asks about, such as news, prices, opening hours, " "schedules, scores or who holds a role. Returns the answer, the page it was read on and the top results." ), "parameters": { "type": "object", "properties": { "q": { "type": "string", "description": "The question in everyday words, with every name and place spelled out", }, }, "required": ["q"], }, }, { "name": "read_page", "description": "Read the main text of one web page the caller names, or one that search_web returned.", "parameters": { "type": "object", "properties": { "url": {"type": "string", "description": "The full address of the page, starting with https://"}, }, "required": ["url"], }, }, ] ``` 3. 3 ## 3. Call the Search API from each function `search_web` sends `GET /v1/web-search` with `q` and `num=5`. The reply's `data.answer` holds the `value` as the pages print it and the `page` it was read on. `data.sources` holds the results in rank order, each with its `title`, `site`, `snippet` and `url`. The function passes the agent only those fields, so its context stays short and the spoken answer comes quickly. `read_page` sends the address to the [Extract API](https://specstream.ai/docs/extract-api.md), `POST /v1/extract`, and passes on the page's main text, cut at 4,000 characters: web\_search\_agent.py ```python specstream = httpx.AsyncClient( base_url="https://api.specstream.ai", headers={"Authorization": f"Bearer {os.environ['SPECSTREAM_API_KEY']}"}, timeout=20, ) async def search_web(q: str) -> dict: r = await specstream.get("/v1/web-search", params={"q": q, "num": 5}) r.raise_for_status() data = r.json()["data"] answer = data.get("answer") or {} return { "answer": answer.get("value"), "answer_page": answer.get("page"), "results": [ {"title": s.get("title"), "site": s.get("site"), "snippet": s.get("snippet"), "url": s["url"]} for s in data.get("sources", [])[:5] ], } async def read_page(url: str) -> dict: r = await specstream.post("/v1/extract", json={"urls": [url], "format": "text"}, timeout=40) r.raise_for_status() page = r.json()["data"]["results"][0] if page["status"] != "ok": return {"status": page["status"], "next": "Tell the caller the page could not be read."} return {"title": page.get("title"), "text": (page.get("text") or "")[:4000]} ``` 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. Then 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. Deepgram waits for that response, so every path sends one. A search with no reply in 20 seconds returns an error the agent can say aloud: web\_search\_agent.py ```python HANDLERS = {"search_web": search_web, "read_page": read_page} 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.TimeoutException: result = {"error": "No reply in time. Tell the caller and offer to try again."} except httpx.HTTPStatusError as e: result = {"error": e.response.text[:300]} 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)) ``` 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: web\_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 are a voice assistant with live web search. For anything that can change, such as news, " "prices, opening hours, schedules, scores or who holds a role, call search_web before you answer. " "Answer in one or two short sentences and say which site the answer came from. " "Never read a web address aloud. When the caller names a page, call read_page." ), "functions": FUNCTIONS, }, "speak": {"provider": {"type": "deepgram", "model": "aura-2-thalia-en"}}, "greeting": "Hello. What would you like to know?", }, } 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 web_search_agent.py`. Ask about something from this week, such as a score, a price or an opening time. The agent calls `search_web` and says the answer with the site it came from. Give it a page address and ask what the page says, and it calls `read_page`. ## What the agent gets back The Search API reads the live web at the moment of the call. When the pages agree on one value, `answer.value` is that value as they print it: a full name, a date with its year or a number with its unit. `answer.page` is the page it was read on. When no result prints a value that answers the question, `answer` is null, and the agent speaks from `results` or asks the caller to narrow the question. The full reply also carries `rows` for a request that asks for many results, with the page and read time of each value, and `price_usd`, what the call cost. The [Search API reference](https://specstream.ai/docs/search-api.md) lists every field and parameter. ## Calls sent before the caller finishes Deepgram sends a function call as soon as the model asks for it, which can be before the caller has finished speaking. If the caller keeps talking, Deepgram sends `FunctionCallCancelled` and the code above stops the task without a response. A search that already reached SpecStream may still be charged. To pay only for searches in confirmed turns, add `"defer_until_eot": True` to `search_web` in `FUNCTIONS`. The call then waits until the caller's turn ends, and that wait is added to the answer. Most agents leave search undeferred: a search changes nothing and costs a fraction of a cent. ## Why the function runs in your code Deepgram can also call a function on an address you name, when the function carries an `endpoint` with `url`, `method` and `headers`. That puts your SpecStream key in the `Settings` message Deepgram holds. Client side, the key stays in your process, you cut the reply down to what the agent needs to speak, and you choose what the agent says when a search fails. ## If something doesn't work | What you see | What to do | | --- | --- | | An error with `401` | The key is unknown, expired or revoked. Create a new key and set it as `SPECSTREAM_API_KEY`. | | An error with `403` | The key can't search, or it reached its total spending limit. Create a key with **Call the APIs**, or raise the limit under **API keys**. | | An error with `402` | The balance is too low. Add balance in your SpecStream workspace. | | `answer` is null | No result printed a value that answers the question. The agent can speak from `results`, or ask the caller for a name, a place or a date. | | The agent goes quiet after a function call | Deepgram waits for a `FunctionCallResponse`. Check that every path sends one with the call's `id`. | ## Cost Each `search_web` call is one Search API call at $5 per 1,000 searches of up to 10 results. `num` is 5, so every call is charged at that rate, and a search that returns no results costs nothing. Each `read_page` call reads one page at $0.80 per 1,000 pages, and a page with no text is not charged. Neither needs a quote or an approval. The key's total spending limit is the ceiling. 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 the rest. ## Remove Take `search_web` and `read_page` out of `agent.think.functions`. Then open **API keys** in your SpecStream workspace and revoke the key, which cuts the agent off at once. 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 web search](https://specstream.ai/integrations/deepgram-voice-agent-web-search.md) covers what it does in Deepgram Voice Agent web 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).