Snowflake Cortex Agents Deep Search

Give a Snowflake Cortex Agent Deep Search as three custom tools. The user describes the rows they need, the agent shows how the request was read 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 Snowflake Cortex Agents Deep SearchAgent frameworks & builders

What you need

A Snowflake account where you can use the ACCOUNTADMIN role. Creating an External Access Integration needs the CREATE INTEGRATION privilege on the account.
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 Snowflake and check Authorize paid work under Permissions: starting a search needs it.
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.

2. Create the role, the External Access and the quote table

In a Snowsight worksheet, run this as ACCOUNTADMIN. Put your key in place of YOUR_SPECSTREAM_KEY. Snowflake stores it encrypted, and no query prints it. Each statement is IF NOT EXISTS, so if you already followed the Cortex Agents web search guide, it keeps your role, secret and integration.

The deep_search_quotes table keeps each price the agent gets. The start procedure reads the amount to approve from this table, never from the model.

Snowflake worksheet
USE ROLE ACCOUNTADMIN;

CREATE ROLE IF NOT EXISTS specstream_agent_role;
GRANT DATABASE ROLE SNOWFLAKE.CORTEX_USER TO ROLE specstream_agent_role;
SET my_user = CURRENT_USER();
GRANT ROLE specstream_agent_role TO USER IDENTIFIER($my_user);

CREATE DATABASE IF NOT EXISTS specstream;
CREATE SCHEMA IF NOT EXISTS specstream.tools;
GRANT USAGE ON DATABASE specstream TO ROLE specstream_agent_role;
GRANT USAGE ON SCHEMA specstream.tools TO ROLE specstream_agent_role;
GRANT CREATE AGENT ON SCHEMA specstream.tools TO ROLE specstream_agent_role;

CREATE WAREHOUSE IF NOT EXISTS specstream_agent_wh
  WAREHOUSE_SIZE = 'XSMALL'
  AUTO_SUSPEND = 60
  AUTO_RESUME = TRUE
  INITIALLY_SUSPENDED = TRUE;
GRANT USAGE ON WAREHOUSE specstream_agent_wh TO ROLE specstream_agent_role;

CREATE NETWORK RULE IF NOT EXISTS specstream.tools.specstream_api_rule
  MODE = EGRESS
  TYPE = HOST_PORT
  VALUE_LIST = ('api.specstream.ai:443');

CREATE SECRET IF NOT EXISTS specstream.tools.specstream_api_key
  TYPE = GENERIC_STRING
  SECRET_STRING = 'YOUR_SPECSTREAM_KEY';

CREATE EXTERNAL ACCESS INTEGRATION IF NOT EXISTS specstream_api_access
  ALLOWED_NETWORK_RULES = (specstream.tools.specstream_api_rule)
  ALLOWED_AUTHENTICATION_SECRETS = (specstream.tools.specstream_api_key)
  ENABLED = TRUE;

CREATE TABLE IF NOT EXISTS specstream.tools.deep_search_quotes (
  quote_id STRING,
  quote_hash STRING,
  maximum_cents INTEGER,
  priced_at TIMESTAMP_LTZ DEFAULT CURRENT_TIMESTAMP()
);

3. Create the price procedure

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 and keeps the quote in the table. The quote's maximum_cents is the most the search can cost. Pricing spends nothing, and the quote stays open for 30 minutes.

It is a procedure, not a function, because it writes to the table. EXECUTE AS OWNER lets it write there while the agent's role has no access to the table.

Snowflake worksheet
CREATE OR REPLACE PROCEDURE specstream.tools.price_deep_search(request STRING, row_count INTEGER)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = 3.12
PACKAGES = ('snowflake-snowpark-python', 'requests')
HANDLER = 'run'
EXTERNAL_ACCESS_INTEGRATIONS = (specstream_api_access)
SECRETS = ('cred' = specstream.tools.specstream_api_key)
EXECUTE AS OWNER
AS
$$
import _snowflake
import requests


class Refused(Exception):
    pass


def call(method, path, **kwargs):
    reply = requests.request(
        method,
        "https://api.specstream.ai" + path,
        headers={"Authorization": "Bearer " + _snowflake.get_generic_secret_string("cred")},
        timeout=60,
        **kwargs,
    )
    try:
        body = reply.json()
    except ValueError:
        body = {}
    if reply.status_code >= 400:
        raise Refused({"status": reply.status_code, "error": body.get("error")})
    return body["data"]


def run(session, request, row_count):
    try:
        reading = call("POST", "/v1/parse", json={"q": request})
        if reading.get("status") != "runnable":
            return {
                "status": reading.get("status"),
                "read_back": reading.get("read_back"),
                "next": "Ask the user for the missing detail, then price again.",
            }
        rows = max(1, min(int(row_count or 25), 100))  # the row limit holds whatever the model sends
        quote = call(
            "POST",
            "/v1/capacity/quotes",
            json={"ast_hash": reading["ast_hash"], "operation": "find", "requested_units": rows},
        )
    except Refused as refused:
        return refused.args[0]
    session.sql(
        "INSERT INTO specstream.tools.deep_search_quotes (quote_id, quote_hash, maximum_cents) VALUES (?, ?, ?)",
        params=[quote["quote_id"], quote["quote_hash"], quote["maximum_cents"]],
    ).collect()
    return {
        "read_back": reading["read_back"],
        "rows": rows,
        "maximum_usd": quote["maximum_cents"] / 100,
        "quote_id": quote["quote_id"],
    }
$$;

GRANT USAGE ON PROCEDURE specstream.tools.price_deep_search(STRING, INTEGER)
  TO ROLE specstream_agent_role;

4. Create the start procedure

start_deep_search takes only the quote_id. It reads the quote's hash and maximum from the table, deletes the row so the quote is used once, and approves it with POST /v1/capacity/executions. The quote's own ID is the request key, so a repeated call cannot start a second search.

Snowflake worksheet
CREATE OR REPLACE PROCEDURE specstream.tools.start_deep_search(quote_id STRING)
RETURNS VARIANT
LANGUAGE PYTHON
RUNTIME_VERSION = 3.12
PACKAGES = ('snowflake-snowpark-python', 'requests')
HANDLER = 'run'
EXTERNAL_ACCESS_INTEGRATIONS = (specstream_api_access)
SECRETS = ('cred' = specstream.tools.specstream_api_key)
EXECUTE AS OWNER
AS
$$
import _snowflake
import requests


class Refused(Exception):
    pass


def call(method, path, **kwargs):
    reply = requests.request(
        method,
        "https://api.specstream.ai" + path,
        headers={"Authorization": "Bearer " + _snowflake.get_generic_secret_string("cred")},
        timeout=60,
        **kwargs,
    )
    try:
        body = reply.json()
    except ValueError:
        body = {}
    if reply.status_code >= 400:
        raise Refused({"status": reply.status_code, "error": body.get("error")})
    return body["data"]


def run(session, quote_id):
    kept = session.sql(
        "SELECT quote_hash, maximum_cents FROM specstream.tools.deep_search_quotes "
        "WHERE quote_id = ? AND priced_at > DATEADD(minute, -30, CURRENT_TIMESTAMP())",
        params=[quote_id],
    ).collect()
    if not kept:
        return {"error": "Unknown, used or expired quote. Price the request again."}
    session.sql("DELETE FROM specstream.tools.deep_search_quotes WHERE quote_id = ?", params=[quote_id]).collect()
    try:
        started = call(
            "POST",
            "/v1/capacity/executions",
            json={
                "quote_id": quote_id,
                "quote_hash": kept[0]["QUOTE_HASH"],
                "confirm_max_cents": kept[0]["MAXIMUM_CENTS"],
                "idempotency_key": quote_id,
            },
        )
    except Refused as refused:
        return refused.args[0]
    return {
        "find_id": started["find_id"],
        "next": "The search runs for minutes. Call read_deep_search with this find_id when the user asks for the rows.",
    }
$$;

GRANT USAGE ON PROCEDURE specstream.tools.start_deep_search(STRING)
  TO ROLE specstream_agent_role;

5. Create the read function

read_deep_search reads the find's status from GET /v1/finds/{id} and the first 10 rows from GET /v1/finds/{id}/rows with format=markdown: one table with a column for each fact the user asked for and a Source column. It only reads, so it is a function.

Snowflake worksheet
CREATE OR REPLACE FUNCTION specstream.tools.read_deep_search(find_id STRING)
RETURNS STRING
LANGUAGE PYTHON
RUNTIME_VERSION = 3.12
PACKAGES = ('snowflake-snowpark-python', 'requests')
HANDLER = 'run'
EXTERNAL_ACCESS_INTEGRATIONS = (specstream_api_access)
SECRETS = ('cred' = specstream.tools.specstream_api_key)
AS
$$
import _snowflake
import requests


class Refused(Exception):
    pass


def call(method, path, **kwargs):
    reply = requests.request(
        method,
        "https://api.specstream.ai" + path,
        headers={"Authorization": "Bearer " + _snowflake.get_generic_secret_string("cred")},
        timeout=60,
        **kwargs,
    )
    try:
        body = reply.json()
    except ValueError:
        body = {}
    if reply.status_code >= 400:
        raise Refused({"status": reply.status_code, "error": body.get("error")})
    return body["data"]


def run(find_id):
    try:
        find = call("GET", "/v1/finds/" + find_id)
        page = call("GET", "/v1/finds/" + find_id + "/rows", params={"format": "markdown", "limit": 10})
    except Refused as refused:
        return "SpecStream answered " + str(refused.args[0]["status"]) + ": " + str(refused.args[0]["error"])
    return "Status: " + find["status"] + ". Rows so far: " + str(find["accepted_count"]) + ".\n\n" + page["table"]
$$;

GRANT USAGE ON FUNCTION specstream.tools.read_deep_search(STRING)
  TO ROLE specstream_agent_role;

6. Try the three tools from SQL

The first call spends nothing. The second starts a search and spends up to the maximum the first one returned, so run it only for a request you want.

Snowflake worksheet
USE ROLE specstream_agent_role;
USE WAREHOUSE specstream_agent_wh;

-- Prices the request and spends nothing.
CALL specstream.tools.price_deep_search('Coffee roasters in Portland, Oregon that sell wholesale, with their website', 20);

-- Spends up to the maximum above. Put the quote_id from the price in place of QUOTE_ID.
CALL specstream.tools.start_deep_search('QUOTE_ID');

-- Put the find_id from the start in place of FIND_ID.
SELECT specstream.tools.read_deep_search('FIND_ID');

7. Create the agent

Register the three tools as custom tools. Each tool_spec of type generic gives the agent the tool's inputs, and tool_resources maps it to its procedure or function and the warehouse that runs it. The instructions make the agent show the price and wait for a yes before it starts anything. The specification is YAML; this one is written as JSON, which YAML reads as it is.

Snowflake worksheet
USE ROLE specstream_agent_role;
USE WAREHOUSE specstream_agent_wh;

CREATE OR REPLACE AGENT specstream.tools.deep_search_agent
  COMMENT = 'Deep Search tables through SpecStream'
  FROM SPECIFICATION
$$
{
  "models": {
    "orchestration": "auto"
  },
  "orchestration": {
    "budget": {
      "seconds": 120,
      "tokens": 32000
    }
  },
  "instructions": {
    "orchestration": "When the user wants many records with the same facts on each, call price_deep_search with their request and the number of rows. Show the read_back and maximum_usd it returns and ask the user to approve that maximum. Call start_deep_search only after the user approves that price in their own message. Never start a search the user did not approve. When the user asks for the rows, call read_deep_search with the find_id.",
    "response": "Show the rows as the table read_deep_search returns, with its Source column."
  },
  "tools": [
    {
      "tool_spec": {
        "type": "generic",
        "name": "price_deep_search",
        "description": "Price a table of rows from the live web: companies, places, people or other records that meet the user's conditions, with the facts to bring back. Spends nothing. Returns how SpecStream read the request, the maximum price in dollars and a quote_id.",
        "input_schema": {
          "type": "object",
          "properties": {
            "request": {
              "type": "string",
              "description": "What each row is, the conditions it must meet and the facts to bring back, in the user's words."
            },
            "row_count": {
              "type": "integer",
              "description": "How many rows the user wants, 1 to 100."
            }
          },
          "required": [
            "request",
            "row_count"
          ]
        }
      }
    },
    {
      "tool_spec": {
        "type": "generic",
        "name": "start_deep_search",
        "description": "Start a priced Deep Search. Call it only after the user approved the maximum price that price_deep_search returned. Returns a find_id.",
        "input_schema": {
          "type": "object",
          "properties": {
            "quote_id": {
              "type": "string",
              "description": "The quote_id that price_deep_search returned."
            }
          },
          "required": [
            "quote_id"
          ]
        }
      }
    },
    {
      "tool_spec": {
        "type": "generic",
        "name": "read_deep_search",
        "description": "Read the status and the first 10 rows of a Deep Search as a Markdown table with a Source column.",
        "input_schema": {
          "type": "object",
          "properties": {
            "find_id": {
              "type": "string",
              "description": "The find_id that start_deep_search returned."
            }
          },
          "required": [
            "find_id"
          ]
        }
      }
    }
  ],
  "tool_resources": {
    "price_deep_search": {
      "type": "procedure",
      "identifier": "SPECSTREAM.TOOLS.PRICE_DEEP_SEARCH",
      "execution_environment": {
        "type": "warehouse",
        "warehouse": "SPECSTREAM_AGENT_WH"
      }
    },
    "start_deep_search": {
      "type": "procedure",
      "identifier": "SPECSTREAM.TOOLS.START_DEEP_SEARCH",
      "execution_environment": {
        "type": "warehouse",
        "warehouse": "SPECSTREAM_AGENT_WH"
      }
    },
    "read_deep_search": {
      "type": "function",
      "identifier": "SPECSTREAM.TOOLS.READ_DEEP_SEARCH",
      "execution_environment": {
        "type": "warehouse",
        "warehouse": "SPECSTREAM_AGENT_WH"
      }
    }
  }
}
$$;

8
8. Ask the agent
In Snowsight, open AI & ML, then Agents, and select DEEP_SEARCH_AGENT. Ask for rows in the agent playground and say how many. The agent shows how it read the request and the maximum price. Approve it, and the agent starts the search. A search runs for minutes; ask for the rows later, in the same conversation.
Agent playground
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 user 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.

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.

How spending stays under control

Pricing spends nothing. price_deep_search only reads the request and returns a quote.
The user approves the price. The agent's instructions make it show the maximum and wait for a yes before start_deep_search.
The model cannot change the amount. confirm_max_cents comes from the quote table, and a search never costs more than it.
The row limit holds in code. price_deep_search caps the 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.
The role gets USAGE on the two procedures and the function, not on the integration, the secret or the table, so its users cannot read the key or reach another host.

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 user and prices it again.
`Unknown, used or expired quote`	The quote was used, or it is more than 30 minutes old. Price the request again.
`status` 402	The balance is too low for the approved maximum. Add balance in your SpecStream workspace.
`status` 403	The key has no Authorize paid work, or the search would go over its total spending limit. Make a key with it, then run `ALTER SECRET specstream.tools.specstream_api_key SET SECRET_STRING = 'NEW_KEY';`.
`status` 401	The key in the secret is unknown, expired or revoked. Create a new key and run `ALTER SECRET`.
A Python error that names `api.specstream.ai`	The network rule does not allow the host. Check it with `SHOW NETWORK RULES` and `DESCRIBE EXTERNAL ACCESS INTEGRATION specstream_api_access`.

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 user approved. Snowflake bills the warehouse that runs the tools, as it bills any warehouse. The rates are on pricing, and balance and billing covers quotes.

Web search and page reading too

For one fact or the text of a page, the Cortex Agents web search guide adds web_search and read_pages tools on the same integration. The Search API is $5 per 1,000 searches of up to 10 results and the Extract API $0.80 per 1,000 pages, with no quote and no approval. The REST API reference lists every call on this page.

Remove

Run DROP AGENT specstream.tools.deep_search_agent;, then DROP PROCEDURE for specstream.tools.price_deep_search(STRING, INTEGER) and specstream.tools.start_deep_search(STRING), DROP FUNCTION specstream.tools.read_deep_search(STRING); and DROP TABLE specstream.tools.deep_search_quotes;. Then open API keys in your SpecStream workspace and revoke the key. Rows already delivered stay in your account. Questions: [email protected].

SpecStream for Snowflake Cortex Agents Deep Search covers what it does in Snowflake Cortex Agents Deep Search 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].