はじめに

　Snowflake Cortex Agentsは，Snowflake上でRAGやText-to-SQLなどの機能を持つAIエージェントを構築できる機能である．Snowflake Intelligenceからは簡単に利用できるが，外部アプリケーションから呼び出したいケースも多いのではないだろうか．

　本記事では，ローカルのPython環境からCortex Agentsを呼び出す方法について，実際に試行錯誤した過程も含めて解説する．公式ドキュメントだけでは分かりにくかった部分や，実装時にハマったポイントも共有したい．

背景と課題

　Cortex Agentsは2025年2月にパブリックプレビューとなって，同年11月に一般提供された比較的新しい機能である．Snowflake内で完結する場合は問題ないが，以下のようなケースでは外部からの呼び出しが必要になる．

既存のWebアプリケーションにCortex Agentsを組み込みたい
ローカルでの開発・デバッグを効率化したい
複数のSnowflakeアカウントのエージェントを統合管理したい

　公式ドキュメントにはREST APIの仕様が記載されているが，実際に動かそうとするとつまずくことが多かった．

実装のポイント

エンドポイントの正しい形式を見つける

　最初に苦労したのは，正しいAPIエンドポイントの特定である．ドキュメントを見ると汎用的なエンドポイントが紹介されているが，実際にCortex Agentsを呼び出すには，エージェント固有のエンドポイントを使用する必要があった．

　汎用エンドポイントでも200が返ってくるため一見成功しているように見えるが，実はエージェントの設定が無視され，単なるLLM呼び出しになってしまう．この違いに気づくまでにかなりの時間を費やした．正しいエンドポイント形式は以下のとおりである．

https://{account}.snowflakecomputing.com/api/v2/databases/{database}/schemas/{schema}/agents/{agent_name}:run

認証とリクエストの構築

　Snowflakeへの接続にはsnowflake-connectorを使用し，そこからセッショントークンを取得してREST APIの認証に利用する．

import snowflake.connector
import requests

# Snowflake接続
conn = snowflake.connector.connect(
    user="your_user",
    password="your_password",
    account="your_account",
    database="your_database",
    schema="your_schema",
    role="your_role",
    warehouse="your_warehouse"
)

# トークン取得
token = conn.rest.token

# APIリクエスト
headers = {
    "Authorization": f'Snowflake Token="{token}"',
    "Content-Type": "application/json; charset=utf-8",
    "Accept": "text/event-stream"
}

　リクエストボディはシンプルで，messagesにユーザーの入力を含めるだけである．エージェント固有エンドポイントを使用する場合，agentパラメータは不要である点に注意してほしい．

request_body = {
    "messages": [
        {
            "role": "user",
            "content": [{"type": "text", "text": "質問内容"}]
        }
    ]
}

Server-Sent Eventsの処理

　Cortex AgentsのAPIはServer-Sent Events形式でストリーミングレスポンスを返す．各イベントはdata:プレフィックスで始まり，JSON形式でテキストの断片が送られてくる．

response = requests.post(
    endpoint,
    headers=headers,
    json=request_body,
    timeout=120,
    stream=False  # 簡易的に全体を取得する場合
)

content = response.content.decode('utf-8')

# レスポンス解析
full_text = ""
for line in content.split('\n'):
    if line.startswith("data:"):
        data_str = line[5:].strip()
        if data_str and data_str != "[DONE]":
            data = json.loads(data_str)
            if "text" in data:
                full_text += data["text"]

動作確認

　エンドポイントの違いによる動作の差を確認するため，エージェントに自己紹介を求めてみた．

　汎用エンドポイントを使用した場合，Claudeの標準的な自己紹介が返ってきた．一方，エージェント固有エンドポイントでは，そのエージェントに設定されたツールや役割を認識した回答が得られた．具体的には「data_to_chart」「text-to-SQL」などのツール定義を参照した応答となり，Cortex Agentsが正しく呼び出されていることが確認できた．

Streamlitでの実装例

　検証結果を踏まえ，ローカルのStreamlitアプリからCortex Agentsを呼び出すサンプルを作成した．エージェント一覧の動的取得，エージェントの切り替え，会話履歴の管理などの機能を実装している．エージェント一覧は以下のSQLで取得できる．

SHOW AGENTS IN SCHEMA {database}.{schema}

まとめ

　ローカル環境からCortex Agentsを呼び出すためのポイントを紹介した．

エージェント固有エンドポイントを使用する
snowflake-connectorからセッショントークンを取得して認証に利用する
レスポンスはSSE形式で返されるため適切に解析する
エージェント一覧はSHOW AGENTSで取得できる

　本記事が，Cortex Agentsを外部アプリケーションに組み込もうとしている方の参考になれば幸いである．

最小構成のサンプルコード

import streamlit as st
import snowflake.connector
import requests
import json

# ============================================
# Snowflake接続設定（環境に合わせて変更）
# ============================================
SNOWFLAKE_CONFIG = {
    "user": "YOUR_USER",
    "password": "YOUR_PASSWORD",
    "account": "YOUR_ACCOUNT",      # 例: abc12345.ap-northeast-1
    "database": "YOUR_DATABASE",
    "schema": "YOUR_SCHEMA",
    "role": "YOUR_ROLE",
    "warehouse": "YOUR_WAREHOUSE"
}

# ============================================
# Snowflake接続
# ============================================
@st.cache_resource
def get_connection():
    return snowflake.connector.connect(**SNOWFLAKE_CONFIG)

# ============================================
# Agent一覧取得
# ============================================
def get_agents(conn):
    cursor = conn.cursor()
    cursor.execute(f"SHOW AGENTS IN SCHEMA {SNOWFLAKE_CONFIG['database']}.{SNOWFLAKE_CONFIG['schema']}")
    results = cursor.fetchall()
    columns = [desc[0] for desc in cursor.description]
    cursor.close()
    
    agents = []
    for row in results:
        row_dict = dict(zip(columns, row))
        name = row_dict.get("name", row_dict.get("NAME", ""))
        if name:
            agents.append(name)
    return agents

# ============================================
# Cortex Agent呼び出し
# ============================================
def call_agent(conn, agent_name: str, query: str):
    # エンドポイント構築
    endpoint = (
        f"https://{SNOWFLAKE_CONFIG['account']}.snowflakecomputing.com"
        f"/api/v2/databases/{SNOWFLAKE_CONFIG['database']}"
        f"/schemas/{SNOWFLAKE_CONFIG['schema']}"
        f"/agents/{agent_name}:run"
    )
    
    # リクエスト
    headers = {
        "Authorization": f'Snowflake Token="{conn.rest.token}"',
        "Content-Type": "application/json; charset=utf-8",
        "Accept": "text/event-stream"
    }
    
    body = {
        "messages": [
            {"role": "user", "content": [{"type": "text", "text": query}]}
        ]
    }
    
    response = requests.post(endpoint, headers=headers, json=body, timeout=120)
    
    if response.status_code != 200:
        return None, f"HTTP {response.status_code}: {response.text}"
    
    # SSEレスポンス解析
    text = ""
    for line in response.content.decode('utf-8').split('\n'):
        if line.startswith("data:"):
            data_str = line[5:].strip()
            if data_str and data_str != "[DONE]":
                try:
                    data = json.loads(data_str)
                    if "text" in data:
                        text += data["text"]
                except json.JSONDecodeError:
                    pass
    
    return text, None

# ============================================
# Streamlit UI
# ============================================
def main():
    st.title("Cortex Agent Chat")
    
    # 接続
    conn = get_connection()
    
    # エージェント選択
    agents = get_agents(conn)
    if not agents:
        st.error("エージェントが見つかりません")
        return
    
    selected_agent = st.sidebar.selectbox("Agent", agents)
    
    if st.sidebar.button("履歴クリア"):
        st.session_state.messages = []
        st.rerun()
    
    # チャット
    if "messages" not in st.session_state:
        st.session_state.messages = []
    
    for msg in st.session_state.messages:
        with st.chat_message(msg["role"]):
            st.markdown(msg["content"])
    
    if prompt := st.chat_input("質問を入力"):
        st.session_state.messages.append({"role": "user", "content": prompt})
        with st.chat_message("user"):
            st.markdown(prompt)
        
        with st.chat_message("assistant"):
            with st.spinner("回答生成中..."):
                text, error = call_agent(conn, selected_agent, prompt)
                
                if error:
                    st.error(error)
                elif text:
                    st.markdown(text)
                    st.session_state.messages.append({"role": "assistant", "content": text})
                else:
                    st.warning("応答がありません")

if __name__ == "__main__":
    main()

具体アプリ

　以下は，私が作成したCloud Run上にStreamlitアプリをデプロイしたアプリケーションとなっている．Snowflake外からSnowflakeのCortex Agentsの一覧が表示され，その呼び出すができることを確認することができる．

　一部余計なエージェントが混在しているが，こちらの発表で紹介した，汎用型の数理最適化エージェントを組み込んだCortex Agents（OPTAGENT）を呼び出すと，数理最適化のロジックに基づいた最適解の算出を行ってくれる．

参考

Snowflake Documentation - Cortex Agents 実行 API