❄️
Snowflake Cortex Agents：REST API 利用ガイド

2025/09/09に公開
 はじめに2025年9月上旬、Snowflake Cortex AgentsにAdminオブジェクトがリリースされました。これにより、従来の単純なAPI呼び出しから一歩進み、Snowflake Intelligenceと同様のPlanning (計画策定) やTask Decomposition (タスク分解) が可能になり、より複雑な処理を実行できるようになりました。
この記事では、この新しいREST APIを使いこなすための主要なエンドポイントと、それを呼び出すためのPythonコードをまとめて解説します。特に、以下の3つの要素に焦点を当ててご紹介します。
実装: Python標準のrequestsライブラリを使用し、RESTエンドポイントのみを呼び出します。
セキュリティ: トークンやアカウント情報はコードから分離し、関数引数として安全に渡します。
ストリーミング: 応答はSSE（Server-Sent Events）でストリーミングされ、テキスト、テーブル、チャートなどが段階的に届きます。
各コンポーネントについて詳しく説明していきます。

 1. Agents
 概要と機能Agentsエンドポイントは、特定のタスクを実行するエージェントの作成、実行、管理を行います。
エージェントの作成: 指示（instructions）、使用モデル、ツール、ツールリソースを定義して、特定の業務（例: 売上データ分析、会話ログ検索）を実行するエージェントを登録します。
エージェントの実行 (:run): エージェントにプロンプトを送信し、テキストやテーブル、チャート、さらにはツール呼び出しの進捗状況などをSSEでリアルタイムに受け取ります。
エージェントの削除: 登録済みのエージェントを削除します。
応答の多様性: Cortex Agentsは、単純なテキストだけでなく、以下の多様な形式で応答を生成できます。

text: 通常のテキスト応答。

table: SQLクエリの結果のような構造化された表形式のデータ。

chart: Vega-Lite形式のJSON仕様（chart_spec）で、インタラクティブなチャートをクライアント側で描画できます。

file: データファイルへの参照（例: 画像やPDF）。

代表的なツール:

cortex_analyst_text_to_sql: セマンティックモデル（YAML）を利用し、自然言語をSQLに変換して結果を取得します。

cortex_search: 事前に定義された検索サービスで、非構造化データを検索します。

web_search: インターネット上の情報をリアルタイムで検索する機能です。これにより、エージェントは最新の情報を参照して応答を生成できます。

チャート生成: 応答内のchart_spec（Vega-Lite JSON）を受け取ることで、クライアント側で簡単に可視化できます。

 主要エンドポイント

アクション
HTTPメソッド
パス


作成
POST
/api/v2/databases/{DB}/schemas/{SCHEMA}/agents

実行
POST
/api/v2/databases/{DB}/schemas/{SCHEMA}/agents/{AGENT}:run

削除
DELETE
/api/v2/databases/{DB}/schemas/{SCHEMA}/agents/{AGENT}


 代表的なリクエストボディ（作成・抜粋）{
  "name": "sales_intelligence_agent",
  "comment": "Salesデータの構造化分析と会話検索をオーケストレーション",
  "instructions": {
    "response": "構造化→Analyst、非構造→Search。数値は簡潔に。",
    "orchestration": "両方必要な場合は出典を分けて提示。手法を先に明示。"
  },
  "models": { "orchestration": "claude-4-sonnet" },
  "tools": [
    { "tool_spec": { "type": "cortex_analyst_text_to_sql", "name": "cortex_analyst_sales_metrics" } },
    { "tool_spec": { "type": "cortex_search", "name": "sales_conversation_search" } }
  ],
  "tool_resources": {
    "cortex_analyst_sales_metrics": {
      "semantic_model_file": "@sales_intelligence.data.models/sales_metrics_model.yaml",
      "execution_environment": { "type": "warehouse", "warehouse": "SALES_INTELLIGENCE_WH" }
    },
    "sales_conversation_search": {
      "name": "sales_intelligence.data.sales_conversation_search",
      "max_results": 10,
      "id_column": "conversation_id"
    }
  }
}

 SSE でのイベント例（観測しやすい種類）
response.status: ステータス（planning / executing_tools など）

response.thinking(.delta): 思考過程（可視化される環境では参照可）

response.tool_use / response.tool_result(.status): ツール呼び出しと結果の進行

message.delta: テキスト/チャート/テーブルの増分

response.chart: チャート仕様（Vega-Lite の chart_spec）

response: 最終応答（content[]）

error: ストリーミングエラー

 curl サンプル作成
curl -X POST \
  "https://$SNOWFLAKE_ACCOUNT.snowflakecomputing.com/api/v2/databases/$SNOWFLAKE_DATABASE/schemas/$SNOWFLAKE_SCHEMA/agents" \
  -H "Authorization: Bearer $SNOWFLAKE_PAT" \
  -H "Content-Type: application/json" \
  -d @create_agent.json
実行（SSE）
curl -N -X POST \
  "https://$SNOWFLAKE_ACCOUNT.snowflakecomputing.com/api/v2/databases/$SNOWFLAKE_DATABASE/schemas/$SNOWFLAKE_SCHEMA/agents/$AGENT_NAME:run" \
  -H "Authorization: Bearer $SNOWFLAKE_PAT" \
  -H "Content-Type: application/json" \
  -H "Accept: text/event-stream" \
  -d '{"messages":[{"role":"user","content":[{"type":"text","text":"How many records each sales stage has."}]}]}'
削除
curl -X DELETE \
  "https://$SNOWFLAKE_ACCOUNT.snowflakecomputing.com/api/v2/databases/$SNOWFLAKE_DATABASE/schemas/$SNOWFLAKE_SCHEMA/agents/$AGENT_NAME" \
  -H "Authorization: Bearer $SNOWFLAKE_PAT"

 Python（REST 専用：環境変数などの共通設定は外部で管理）import json, requests

def _agents_base(account, database, schema):
    return f"https://{account}.snowflakecomputing.com/api/v2/databases/{database}/schemas/{schema}/agents"

def _headers(pat, accept=None):
    h = {"Authorization": f"Bearer {pat}", "Content-Type": "application/json"}
    if accept:
        h["Accept"] = accept
    return h

def create_agent(account, database, schema, pat, agent_spec: dict):
    url = _agents_base(account, database, schema)
    resp = requests.post(url, headers=_headers(pat), json=agent_spec, timeout=60)
    resp.raise_for_status()
    return resp.json() if resp.headers.get("Content-Type","").startswith("application/json") else resp.text

def delete_agent(account, database, schema, agent_name, pat):
    url = f"{_agents_base(account, database, schema)}/{agent_name}"
    resp = requests.delete(url, headers=_headers(pat), timeout=60)
    resp.raise_for_status()
    return resp.json() if resp.headers.get("Content-Type","").startswith("application/json") else resp.text

def _iter_sse(iter_lines):
    event, buf = None, []
    for raw in iter_lines:
        if raw is None:
            continue
        line = raw.decode("utf-8", errors="ignore")
        if line.startswith("event:"):
            event = line.split("event:", 1)[1].strip()
        elif line.startswith("data:"):
            buf.append(line.split("data:", 1)[1].strip())
        elif line.strip() == "":
            if event is not None:
                data = "\n".join(buf).strip()
                yield event, data
            event, buf = None, []
    if event is not None:
        data = "\n".join(buf).strip()
        yield event, data

def run_agent_sse(
    account, database, schema, agent_name, pat, messages,
    tool_choice=None, thread_id=None, parent_message_id=None, timeout=300
):
    url = f"{_agents_base(account, database, schema)}/{agent_name}:run"
    body = {"messages": messages}
    if tool_choice:
        body["tool_choice"] = tool_choice
    if thread_id is not None:
        body["thread_id"] = thread_id
    if parent_message_id is not None:
        body["parent_message_id"] = parent_message_id

    with requests.post(url, headers=_headers(pat, accept="text/event-stream"), json=body, stream=True, timeout=timeout) as resp:
        resp.raise_for_status()
        request_id = resp.headers.get("X-Snowflake-Request-Id")
        final_response, texts, charts = None, [], []
        for event, data_str in _iter_sse(resp.iter_lines()):
            if event == "response.text":
                try:
                    d = json.loads(data_str) if data_str else {}
                    if isinstance(d.get("text"), str):
                        texts.append(d["text"])
                except Exception:
                    pass
            elif event == "response.chart":
                try:
                    d = json.loads(data_str) if data_str else {}
                    spec_str = d.get("chart_spec") or (d.get("chart") or {}).get("chart_spec")
                    if spec_str:
                        charts.append(json.loads(spec_str) if isinstance(spec_str, str) else spec_str)
                except Exception:
                    pass
            elif event == "message.delta":
                try:
                    d = json.loads(data_str) if data_str else {}
                    for c in (d.get("delta") or {}).get("content", []):
                        if c.get("type") == "text" and isinstance(c.get("text"), str):
                            texts.append(c["text"])
                        if c.get("type") == "chart":
                            spec_str = (c.get("chart") or {}).get("chart_spec")
                            if spec_str:
                                charts.append(json.loads(spec_str) if isinstance(spec_str, str) else spec_str)
                except Exception:
                    pass
            elif event == "response":
                try:
                    final_response = json.loads(data_str) if data_str else {}
                except Exception:
                    pass
        return {"request_id": request_id, "final_response": final_response, "texts": texts, "charts": charts}

 Snowflake Cortex Agents 実行結果の例下記が出力結果例となります。（ログが長くなるので一部省略しております）
=== Request ===
{
  "messages": [
    {
      "role": "user",
      "content": [
        {
          "type": "text",
          "text": "How many records each sales stage has."
        }
      ]
    }
  ]
}
HTTP: 200
Content-Type: text/event-stream
X-Snowflake-Request-Id: *** 
[response.status] {"message":"Planning the next steps","status":"planning"}
...
[response.tool_result.status] {"message":"Running cortex_analyst_sales_metrics","status":"executing_tool","tool_type":"cortex_analyst_text_to_sql","tool_use_id":"***"}
...
[response.tool_use] {"content_index":1,"input":{"previous_related_tool_result_id":"","query":"How many records does each sales stage have? Count of records by sales stage."},"name":"cortex_analyst_sales_metrics","tool_use_id":"***","type":"cortex_analyst_text_to_sql"}
...
[response.tool_result] {"content":[{"json":{"query_id":"***","result_set":{"data":[["Closed","6"],["Pending","3"],["Lost","1"]],"resultSetMetaData":{...}}}]}
...
[execution_trace] ["{\"trace_id\":\"***\",\"span_id\":\"***\",\"name\":\"Agent\",\"kind\":1,\"start_time_unix_nano\":***,\"end_time_unix_nano\":***,\"attributes\":[{\"key\":\"request_id\",\"value\":{\"stringValue\":\"***\"}}],\"status\":{\"code\":\"STATUS_CODE_OK\"}}", ... ]
Snowflake Cortex Agentsの実行は、SSE（Server-Sent Events）としてストリーミングされます。ここでは、「各セールスステージのレコード数を教えて」というプロンプトに対する応答の主要な部分を抜粋して解説します。

 実行の段階的な進捗応答は、エージェントが思考し、ツールを実行し、結果をまとめるという一連のプロセスをリアルタイムで示します。
Planning (計画策定): エージェントはまず、ユーザーの質問を分析し、最適な回答を生成するための計画を立てます。今回の例では、cortex_analyst_text_to_sqlツールを使用してSQLクエリを生成する計画を立てています。
[response.status] {"message":"Planning the next steps","status":"planning"}

[response.thinking.delta] ...（ユーザーの質問を分析し、Cortex Analystツールを使用する思考過程のテキストが段階的に表示される）...
Tool Execution (ツール実行): 計画に基づき、エージェントは指定されたツールを実行します。その後にSQLが実行され、結果が返されます。
[response.status] {"message":"Getting additional context","status":"executing_tools"}

[response.tool_result.status] ... (Cortex AnalystがSQLを生成し、実行する各ステップの進捗が表示される)...
[response.tool_result]{"content":[{"json":{"query_id":"(***)","result_set":{"data":[["Closed","6"],["Pending","3"],["Lost","1"]],...}}}]}
Final Response (最終応答): ツール実行で得られた結果をもとに、エージェントはユーザーにわかりやすい最終応答を生成します。

[response.thinking.delta] ...（SQLの結果をレビューし、最終的な回答を組み立てる思考過程）...

[response.text.delta] ...（回答テキストが段階的にストリーミングされる）...
最終的な回答

「セールスメトリクスデータに基づくと、各セールスステージのレコード数は以下の通りです：
Closed: 6 レコード
Pending: 3 レコード
Lost: 1 レコード
全てのセールスステージを合わせると合計10件のレコードがあり、そのうち60%がクローズドステージにあります。」

 2. Thread
 概要と機能Threadsエンドポイントは、複数回の対話をスレッドとして管理するための機能を提供します。
対話の継続: thread_idを指定することで、複数ラウンドにわたる会話を継続できます。
管理: スレッドの作成、一覧取得、詳細表示、更新、削除が可能です。
アプリごとの管理: origin_applicationを指定することで、アプリケーション単位でスレッドを絞り込むことができます。

 主要エンドポイント

アクション
HTTPメソッド
パス


作成
POST
/api/v2/cortex/threads

一覧
GET
/api/v2/cortex/threads?origin_application={APP}

詳細
GET
/api/v2/cortex/threads/{thread_id}?page_size=20

更新
POST
/api/v2/cortex/threads/{thread_id}

削除
DELETE
/api/v2/cortex/threads/{thread_id}


 Python（REST 専用）import requests, json

def _threads_base(account):
    return f"https://{account}.snowflakecomputing.com/api/v2/cortex/threads"

def create_thread(account, pat, origin_application: str):
    url = _threads_base(account)
    resp = requests.post(url, headers={"Authorization": f"Bearer {pat}", "Content-Type": "application/json"}, json={"origin_application": origin_application}, timeout=30)
    resp.raise_for_status()
    return resp.json()

def list_threads(account, pat, origin_application: str):
    url = f"{_threads_base(account)}?origin_application={origin_application}"
    resp = requests.get(url, headers={"Authorization": f"Bearer {pat}"}, timeout=30)
    resp.raise_for_status()
    return resp.json()

def describe_thread(account, pat, thread_id: int, page_size: int = 20):
    url = f"{_threads_base(account)}/{thread_id}?page_size={page_size}"
    resp = requests.get(url, headers={"Authorization": f"Bearer {pat}"}, timeout=30)
    resp.raise_for_status()
    return resp.json()

def update_thread(account, pat, thread_id: int, thread_name: str):
    url = f"{_threads_base(account)}/{thread_id}"
    resp = requests.post(url, headers={"Authorization": f"Bearer {pat}", "Content-Type": "application/json"}, json={"thread_name": thread_name}, timeout=30)
    resp.raise_for_status()
    return resp.text if not resp.headers.get("Content-Type","").startswith("application/json") else resp.json()

def delete_thread(account, pat, thread_id: int):
    url = f"{_threads_base(account)}/{thread_id}"
    resp = requests.delete(url, headers={"Authorization": f"Bearer {pat}"}, timeout=30)
    resp.raise_for_status()
    return resp.text if not resp.headers.get("Content-Type","").startswith("application/json") else resp.json()

 継続会話の呼び出し（:run 側での付与例）{
  "thread_id": 4864284,
  "parent_message_id": 0,
  "messages": [
    { "role": "user", "content": [{ "type": "text", "text": "Hello" }] }
  ]
}

 運用上のポイント会話継続時は thread_id と（必要なら）parent_message_id を :run のボディに付与。

origin_application を使うと、アプリ別のスレッド管理がしやすい。

 3. Feedback
 概要と機能Feedbackエンドポイントは、各リクエストに対するユーザーからの評価を送信するために使用します。
フィードバックの送信: 肯定（positive）/否定（negative）の評価、カテゴリ、メッセージを送信します。
品質改善: request_idとthread_idを紐づけることで、Snowflakeは応答品質の分析や継続的な改善にフィードバックを活用できます。

 主要エンドポイント

アクション
HTTPメソッド
パス


送信
POST
/api/v2/databases/{DB}/schemas/{SCHEMA}/agents/{AGENT}:feedback


 request_id の取得
:run の HTTP レスポンスヘッダー X-Snowflake-Request-Id から取得可能。
エラー時もヘッダーに含まれるため、まずは :run 呼び出しのレスポンスヘッダーを保存しておくのが安全。

 curl サンプルcurl -X POST \
  "https://$SNOWFLAKE_ACCOUNT.snowflakecomputing.com/api/v2/databases/$SNOWFLAKE_DATABASE/schemas/$SNOWFLAKE_SCHEMA/agents/$AGENT_NAME:feedback" \
  -H "Authorization: Bearer $SNOWFLAKE_PAT" \
  -H "Content-Type: application/json" \
  -d '{
    "request_id": "'"$REQUEST_ID"'",
    "positive": true,
    "feedback_message": "The answer was accurate and easy to understand",
    "categories": ["accuracy","helpfulness"],
    "thread_id": '"$THREAD_ID"'
  }'

 Python（REST 専用）import requests

def post_feedback(
    account, database, schema, agent_name, pat, request_id: str,
    positive: bool, feedback_message: str, categories: list[str] | None = None, thread_id: int | None = None
):
    url = f"https://{account}.snowflakecomputing.com/api/v2/databases/{database}/schemas/{schema}/agents/{agent_name}:feedback"
    payload = {"request_id": request_id, "positive": bool(positive), "feedback_message": feedback_message}
    if categories:
        payload["categories"] = categories
    if thread_id is not None:
        payload["thread_id"] = thread_id
    resp = requests.post(url, headers={"Authorization": f"Bearer {pat}", "Content-Type": "application/json"}, json=payload, timeout=30)
    resp.raise_for_status()
    return resp.text if not resp.headers.get("Content-Type","").startswith("application/json") else resp.json()

 セキュリティ / 実装 Tips（共通）
認証トークンはコード外で安全に管理し、関数引数で受け渡す。ログ出力時は必ずマスク。
必要に応じて X-Snowflake-Authorization-Token-Type: PROGRAMMATIC_ACCESS_TOKEN を明示。

SSE は Accept: text/event-stream を付与し、長めの timeout を設定。クライアント側でイベント種類ごとに堅牢にパース。

チャートは Vega-Lite 仕様（JSON）。response.chart / message.delta / 最終 response.content[] のいずれかに chart_spec が入るため、順に探索して保存・可視化。

リソース参照（YAML/検索サービス） は Snowflake 内の権限と完全修飾パスを確認。

 付録：最低限の呼び出し順序（例）
create_agent(...) でエージェント登録（必要に応じて一度きり）

create_thread(...) でスレッドを用意（継続会話したい場合）

run_agent_sse(...) にプロンプトを送信（thread_id を付与すると継続）

post_feedback(...) で request_id に紐づけて評価を送信
不要になったら delete_thread(...)、delete_agent(...) を実行
Snowflake Cortex Agents：REST API 利用ガイド

はじめに

1. Agents

概要と機能

主要エンドポイント

代表的なリクエストボディ（作成・抜粋）

SSE でのイベント例（観測しやすい種類）

curl サンプル

Python（REST 専用：環境変数などの共通設定は外部で管理）

Snowflake Cortex Agents 実行結果の例

実行の段階的な進捗

2. Thread

概要と機能

主要エンドポイント

Python（REST 専用）

継続会話の呼び出し（:run 側での付与例）

運用上のポイント

3. Feedback

概要と機能

主要エンドポイント

`request_id` の取得

curl サンプル

Python（REST 専用）

セキュリティ / 実装 Tips（共通）

付録：最低限の呼び出し順序（例）

Discussion

アクション	HTTPメソッド	パス
作成	POST	/api/v2/databases/{DB}/schemas/{SCHEMA}/agents
実行	POST	/api/v2/databases/{DB}/schemas/{SCHEMA}/agents/{AGENT}:run
削除	DELETE	/api/v2/databases/{DB}/schemas/{SCHEMA}/agents/{AGENT}
アクション	HTTPメソッド	パス
作成	POST	/api/v2/cortex/threads
一覧	GET	/api/v2/cortex/threads?origin_application={APP}
詳細	GET	/api/v2/cortex/threads/{thread_id}?page_size=20
更新	POST	/api/v2/cortex/threads/{thread_id}
削除	DELETE	/api/v2/cortex/threads/{thread_id}