Snowflake Cortex Agents Research

Give a Snowflake Cortex Agent a research tool. One SQL script creates a stored procedure that takes a question and returns a written report from the live web, with a numbered list of the pages it cites and when each was read. You register it as a custom tool of resource type Procedure, in SQL or in Snowsight.

SpecStream for Snowflake Cortex Agents ResearchAgent 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 key. Get your key on the Search API screen of your SpecStream workspace. New accounts start with $10 of credit, no card.

1. Allow calls to api.specstream.ai

In a Snowsight worksheet, run this as ACCOUNTADMIN. It creates the role your agent's users get, a warehouse that runs the tools, a network rule for SpecStream's host, a secret that holds your key, and the External Access Integration that binds them. Put your key in place of YOUR_SPECSTREAM_KEY. Snowflake stores it encrypted, and no query prints it.

Each statement is IF NOT EXISTS. If you already followed the Cortex Agents web search guide, it keeps your role, secret and integration, and you can go to the next step.

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;

2. Create the research procedure

research_report takes the question and a depth: quick, standard or deep. It starts the report with POST /v1/research, then reads GET /v1/research/{id} every 5 seconds for up to 90 seconds. A quick report takes about a minute, so it usually comes back in the same call: the text with [1] markers, then a Sources list with each cited page's title, URL and read time.

A standard report takes about four minutes and a deep report about five. For those, the call returns the report ID, which starts with rr_, and the next step's function reads the report when it is done.

It is a procedure, not a function, because it starts paid work: a function could run once for every row of a SELECT. EXECUTE AS OWNER lets it use the secret while the agent's role has no access to it.

Snowflake worksheet
CREATE OR REPLACE PROCEDURE specstream.tools.research_report(question STRING, depth 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)
EXECUTE AS OWNER
AS
$$
import hashlib
import time

import _snowflake
import requests

ENDED = ("completed", "failed", "cancelled", "payment_required")
WAIT_SECONDS = 90  # how long one call waits for the report before it returns the report ID


class Refused(Exception):
    pass


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


def shown(run):
    status = run.get("status")
    if status == "completed":
        report = run.get("report") or {}
        sources = [
            "[" + str(source["n"]) + "] " + source["title"] + " " + source["url"] + " (read " + str(source["read_at"]) + ")"
            for source in report.get("sources") or []
            if source.get("cited")
        ]
        return str(report.get("content") or "") + "\n\nSources\n" + "\n".join(sources)
    if status == "payment_required":
        return (
            "The report is written, but the balance cannot pay for it. Add balance in the SpecStream "
            "workspace, then call read_research_report with report ID " + str(run.get("id")) + "."
        )
    if status in ENDED:
        reason = (run.get("error") or {}).get("message") or status
        return "The report did not finish and costs nothing: " + str(reason)
    return (
        "The report is still in work (stage: " + str(run.get("stage")) + "). Call read_research_report "
        "with report ID " + str(run.get("id")) + " in a minute."
    )


def run(session, question, depth):
    question = (question or "").strip()
    depth = (depth or "quick").strip().lower()
    if not question:
        return "Give the question the report answers."
    if depth not in ("quick", "standard", "deep"):
        return "Set depth to quick, standard or deep."
    # The same question at the same depth in the same hour gets the report it already started,
    # so a repeated tool call never buys the report twice.
    hour = time.strftime("%Y-%m-%dT%H", time.gmtime())
    once = "snowflake-" + hashlib.sha256((question + "|" + depth + "|" + hour).encode()).hexdigest()[:32]
    try:
        report = call(
            "POST",
            "/v1/research",
            json={"input": question, "depth": depth},
            headers={"Idempotency-Key": once},
        )
        deadline = time.monotonic() + WAIT_SECONDS
        while report.get("status") not in ENDED and time.monotonic() < deadline:
            time.sleep(5)
            report = call("GET", "/v1/research/" + report["id"])
    except Refused as refused:
        return refused.args[0]
    return shown(report)
$$;

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

3. Create the read function

read_research_report takes a report ID and returns the finished report, or the stage the report is in: planning, searching, reading, writing or checking. It only reads and costs nothing, so it is a function.

Snowflake worksheet
CREATE OR REPLACE FUNCTION specstream.tools.read_research_report(report_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

ENDED = ("completed", "failed", "cancelled", "payment_required")


class Refused(Exception):
    pass


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


def shown(run):
    status = run.get("status")
    if status == "completed":
        report = run.get("report") or {}
        sources = [
            "[" + str(source["n"]) + "] " + source["title"] + " " + source["url"] + " (read " + str(source["read_at"]) + ")"
            for source in report.get("sources") or []
            if source.get("cited")
        ]
        return str(report.get("content") or "") + "\n\nSources\n" + "\n".join(sources)
    if status == "payment_required":
        return (
            "The report is written, but the balance cannot pay for it. Add balance in the SpecStream "
            "workspace, then call read_research_report with report ID " + str(run.get("id")) + "."
        )
    if status in ENDED:
        reason = (run.get("error") or {}).get("message") or status
        return "The report did not finish and costs nothing: " + str(reason)
    return (
        "The report is still in work (stage: " + str(run.get("stage")) + "). Call read_research_report "
        "with report ID " + str(run.get("id")) + " in a minute."
    )


def run(report_id):
    try:
        return shown(call("GET", "/v1/research/" + (report_id or "").strip()))
    except Refused as refused:
        return refused.args[0]
$$;

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

4. Try both from SQL

The CALL starts one quick report. SpecStream takes its $0.10 from your balance when the report is delivered.

Snowflake worksheet
USE ROLE specstream_agent_role;
USE WAREHOUSE specstream_agent_wh;

-- Starts one quick report and spends $0.10 when it is delivered.
CALL specstream.tools.research_report(
  'What changed in the EU AI Act obligations for general purpose models in 2026?',
  'quick'
);

-- If the call returned a report ID, put it in place of REPORT_ID. Reading costs nothing.
SELECT specstream.tools.read_research_report('REPORT_ID');

5. Create the agent

Register both 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 say the price and wait for a yes before it starts a report. The budget gives one answer 300 seconds, enough for the 90 seconds the procedure waits. 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.research_agent
  COMMENT = 'Research reports from the live web through SpecStream'
  FROM SPECIFICATION
$$
{
  "models": {
    "orchestration": "auto"
  },
  "orchestration": {
    "budget": {
      "seconds": 300,
      "tokens": 32000
    }
  },
  "instructions": {
    "orchestration": "When the user asks a question that needs a written answer from the live web, with sources, tell them the price of a report: $0.10 for quick, $0.30 for standard, $1.00 for deep. Use quick unless the user asks for more depth. Call research_report only after the user approves that price in their own message. Never start a report the user did not approve. If research_report returns a report ID in place of the report, tell the user the report is in work and call read_research_report with that ID when they ask for it.",
    "response": "Show the report as the tool returns it, with its [1] markers and its Sources list. Add no claim the report does not make."
  },
  "tools": [
    {
      "tool_spec": {
        "type": "generic",
        "name": "research_report",
        "description": "Write a research report that answers one question from the live web, with numbered sources. Costs $0.10 (quick), $0.30 (standard) or $1.00 (deep), so call it only after the user approved the price. Returns the report text, or a report ID when the report needs more time.",
        "input_schema": {
          "type": "object",
          "properties": {
            "question": {
              "type": "string",
              "description": "The question the report answers, in the user's words, with the scope, the dates and the places that matter."
            },
            "depth": {
              "type": "string",
              "description": "quick, standard or deep. Use quick unless the user asked for more depth."
            }
          },
          "required": [
            "question",
            "depth"
          ]
        }
      }
    },
    {
      "tool_spec": {
        "type": "generic",
        "name": "read_research_report",
        "description": "Read a research report by its report ID: the finished text with its sources, or the stage it is in. Costs nothing.",
        "input_schema": {
          "type": "object",
          "properties": {
            "report_id": {
              "type": "string",
              "description": "The report ID that research_report returned. It starts with rr_."
            }
          },
          "required": [
            "report_id"
          ]
        }
      }
    }
  ],
  "tool_resources": {
    "research_report": {
      "type": "procedure",
      "identifier": "SPECSTREAM.TOOLS.RESEARCH_REPORT",
      "execution_environment": {
        "type": "warehouse",
        "warehouse": "SPECSTREAM_AGENT_WH"
      }
    },
    "read_research_report": {
      "type": "function",
      "identifier": "SPECSTREAM.TOOLS.READ_RESEARCH_REPORT",
      "execution_environment": {
        "type": "warehouse",
        "warehouse": "SPECSTREAM_AGENT_WH"
      }
    }
  }
}
$$;

6. Ask the agent

In Snowsight, open AI & ML, then Agents, and select RESEARCH_AGENT. Ask a question in the agent playground. The agent names the price, and after your yes it returns the report with its sources.

Agent playground

Write a quick research report: which US states passed data broker laws since 2024, and what does each require? Cite the pages.

Add the tool to an agent you already have

An agent your team already uses takes the same procedure, with no new agent. Snowflake's docs now list Snowflake Intelligence as Snowflake CoWork (opens in a new tab), and an agent your users reach there is edited in the same place. Grant its role USAGE on the procedure and the function first, then add each one in Snowsight:

In the navigation menu, select AI & ML, then Agents, select the agent and select Edit.
Select Tools, find Custom tools and select + Add.
For Name, enter research_report. For Resource type, select procedure.
For Custom tool identifier, select SPECSTREAM.TOOLS.RESEARCH_REPORT. The parameters question and depth appear by themselves. Give each the description from the agent spec above.
For Warehouse, select SPECSTREAM_AGENT_WH. Snowflake does not choose one for you.
For Description, paste the tool's description from the agent spec above, with the three prices, then select Add.
Add read_research_report the same way with the resource type function, then select Save.
Under Orchestration, paste the planning instructions from the agent spec, and set the time limit in Budget configuration to 300 seconds.

What the agent gets back

A research report answers one question in plain words. SpecStream plans the question, searches the live web, reads the pages and writes the answer. Each cited sentence carries a [1] marker, and the Sources list under the text gives that page's title, its URL and the time it was read in this run.

The tools return the report as text, which is what a Cortex Agent shows best. In your own code, GET /v1/research/{id} returns more: report.citations holds each cited sentence with its passage, and report.sources lists every page read, cited or not. Send "output": {"type": "json", "schema": {...}} to get JSON in the shape of your schema in place of text, and include_domains or exclude_domains to choose the sites. The research docs list every field.

How spending stays under control

One report has one fixed price for its depth. No report costs more than $1.00.
The user approves the price. The agent's instructions make it name the price and wait for a yes before research_report.
A repeated call buys nothing new. The procedure sends an Idempotency-Key built from the question, the depth and the hour, so the same question in the same hour returns the report it already started.
The charge comes at delivery. A report that fails or is cancelled costs nothing.
The role gets USAGE on the procedure and the function, not on the integration or the secret, so its users cannot read the key or reach another host.
To keep deep reports out, remove "deep" from the depth check in the procedure and run the CREATE OR REPLACE again.

If something doesn't work

What you see	What to do
`The report is still in work`	The report needs more than the 90 seconds one call waits. Ask the agent for it again in a minute, or run `SELECT specstream.tools.read_research_report('rr_...');`.
`SpecStream answered 401`	The key in the secret is unknown, expired or revoked. Create a new key and run `ALTER SECRET specstream.tools.specstream_api_key SET SECRET_STRING = 'NEW_KEY';`.
`SpecStream answered 402`, or `the balance cannot pay for it`	The balance does not hold the report's price. Add balance in your SpecStream workspace. A report that was already written is delivered on the next read.
`SpecStream answered 422`	A field is wrong, and the reply names it. The question holds up to 15,000 characters.
`SpecStream answered 429`	The account used its calls for the minute. Wait and ask again. Nothing is charged.
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`.
The agent stops before the report arrives	The agent's budget ended first. Set the time limit in Budget configuration to 300 seconds, or `budget.seconds` in the agent spec.
Snowflake refuses the tool for one user	Their role has no `USAGE` on the procedure or the function. Run the two `GRANT USAGE` statements for that role.

Cost

A report costs $0.10 at quick, $0.30 at standard and $1.00 at deep, taken from your SpecStream balance once, when the report is delivered. Reading a report costs nothing. Snowflake bills the warehouse that runs the tools, as it bills any warehouse, and that includes the seconds the procedure waits for the report. The rates are on pricing.

Web search, page reading and tables 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. For a table of every match with the same facts on each row, the Deep Search guide adds three more tools. The REST API reference lists every call on this page.

Remove

Run DROP AGENT specstream.tools.research_agent;, DROP PROCEDURE specstream.tools.research_report(STRING, STRING); and DROP FUNCTION specstream.tools.read_research_report(STRING);. If no other SpecStream tool uses them, drop the integration, the secret and the network rule too, then revoke the key in your SpecStream workspace. Reports already delivered stay in your account. Questions: [email protected].

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