Mastra × Snowflake MCP Serverでデータ分析AIエージェントを作ってみる

$ npm install -g mastra@latest 
$ mastra init

const mcp = new MCPConfiguration({
  id: "snowflake-agent-mcp",
  servers: {
    snowflake_pip: {
      command: "uv",
      args: [
        "--directory",
        "/Users/your_user_name/github.com/isaacwasserman/mcp-snowflake-server",
        "run",
        "mcp_snowflake_server",
        "--account",
        "ORG_NAME_ACCOUNT_NAME",
        "--warehouse",
        "YOUR_WH",
        "--user",
        "YOUR_USER_NAME",
        "--password",
        "YOUR_PASSWORD",
        "--role",
        "SYSADMIN",
        "--database",
        "YOUR_DATABASE",
        "--schema",
        "YOUR_SCHEMA",
      ]
    }
  },
});

export const snowflakeQueryAgent = new Agent({
  name: "snowflakeQueryAgent",
  instructions: `  
        ユーザーからの質問を分析し、適切なクエリ結果を取得してください。
 
        利用可能なデータベース: SUMO
        利用可能なスキーマ: CHANCO
        
        利用可能なテーブル:
        - BANZUKE: 力士の番付情報
        
        【BANZUKEテーブル】
        力士の番付情報（1983年1月から2024年7月までのデータを収録）
        
        - ID: 力士の固有ID
        - RIKISHI: 力士名
        - RANK: 番付の頭文字（階級と位置）
          * Y = 横綱（最高位）
          * O = 大関
          * S = 関脇
          * K = 小結
          * M = 前頭
          * J = 十両
          * Sd = 幕下
          * 数字は順位を表す（例：Y1e = 横綱東1枚目）
          * e = 東, w = 西
          * HD = 怪我などによる休場
          * YO = 横綱審議委員会からの注意
        - HEYA: 部屋名（所属する相撲部屋）
        - SHUSSHIN: 出身地
        - BIRTH_DATE: 生年月日
        - HEIGHT: 身長（cm）
        - WEIGHT: 体重（kg）
        - BASHO: 場所（年月形式、例：2023.01）
        - PREV: 前場所の番付
        - PREV_W: 前場所の勝ち数、つまりBASHOが2023.01の場合、PREV_Wは2022年11月場所の勝ち数であることに注意
        - PREV_L: 前場所の負け数、つまりBASHOが2023.01の場合、PREV_Lは2022年11月場所の負け数であることに注意
        
        日付型のデータはJSON形式に変換できないので、日付型のデータを取得する際は、
        TO_VARCHAR関数を使用して文字列型に変換してください。例：
        TO_VARCHAR(BIRTH_DATE, 'YYYY-MM-DD') as BIRTH_DATE
        勝率を計算する場合、力士の休場などで試合数が0になる場合があるので、
        CASE文を使用して分母が0にならないようにしてください。
      `,
  model: openai("gpt-4o-mini"),
  tools: {
    ...(await mcp.getTools()),
  },
});
 
 
export const snowflakeAnalysisAgent = new Agent({
  name: "snowflakeAnalysisAgent",
  instructions: `
        あなたはデータ分析と視覚化のエキスパートです。
 
        与えられたデータを分析し、以下を提供してください：
        2. データから得られる重要なインサイト（少なくとも3つ）
        3. データの分析結果の詳細な説明
 
        【データ形式について】
        入力データは以下のいずれかの形式で提供されます：
        - JSON形式の構造化データ
        - テキスト形式のデータ（例：力士の番付情報のリスト）
 
        テキスト形式のデータを受け取った場合は、まず内容を理解し、分析可能な形に構造化してください。
        例えば、力士のリストが提供された場合は、勝率、体重と勝率の関係、出身地の分布などを分析できます。
 
        必ず以下のツールを使用してください：
        - visualizeDataTool（データの視覚化）
        - snowflake_pip_append_insight（インサイトの記録）
        
        【visualizeDataToolの使用方法】
 
        visualizeDataTool({
          data: [...], // 視覚化するデータの配列（必須）
          visualizationType: "bar", "line", "pie", "table"
          title: "タイトル", // オプション
          xAxis: "X軸のカラム名", // オプション
          yAxis: "Y軸のカラム名" // オプション
        })
 
        データの特性に応じて適切な視覚化タイプを選択してください：
        - 比較データ → 棒グラフ（bar）
        - 時系列データ → 折れ線グラフ（line）
        - 構成比 → 円グラフ（pie）
 
        【インサイトについて】
        データから得られた重要な洞察を記録するには、snowflake_pip_append_insightツールを使用してください：
 
        snowflake_pip_append_insight({
          insight: "発見したインサイトの説明"
        })
 
        各インサイトは具体的で、データに基づいたものにしてください。
        例：「2023年の十両力士の平均勝率は58%で、前年比5%増加している」
 
        【分析テキストについて】
        最後に、データ全体の分析結果を詳細に説明するテキストを提供してください。
        このテキストには以下を含めてください：
        - データの概要
        - 主要な傾向や特徴
        - インサイトの詳細な解説
        - 画像は表示できないため、テキストでの説明に留めてください。
      `,
  model: openai("gpt-4o-mini"),
  tools: {
    ...snowflakeTools,
    ...(await mcp.getTools()),
  },
});

"use server";
 
import { mastra } from "@/mastra";
 
export type MessageResponse = {
  text: string;
  data?: any;
  insights?: string[];
};
 
/**
 * メッセージをSnowflakeエージェントに送信する
 * @param message ユーザーからのメッセージ
 * @returns エージェントからの応答
 */
export async function sendMessage(message: string): Promise<MessageResponse> {
  try {
    // ワークフローを実行
    const snowflakeWorkflow = mastra.getWorkflow("snowflakeWorkflow");
    const { start } = snowflakeWorkflow.createRun();
 
    console.log("userQuery", message);
 
    // ワークフローを開始し、全ステップを実行
    const result = await start({
      triggerData: { userQuery: message },
    });
 
 
    const queryResult = workflowResult.results?.queryData?.output?.queryResult;
    const analysisResult = workflowResult.results?.analyzeData?.output;
 
    // クエリ結果はあるが分析結果がない場合
    if (queryResult && (!analysisResult || !analysisResult.analysisText)) {
      return {
        text: queryResult,
        data: null,
        insights: []
      };
    }
    
    // 分析結果がある場合
    if (analysisResult) {
      return {
        text: analysisResult.analysisText || "分析が完了しました。",
        data: analysisResult.visualizationData,
        insights: analysisResult.insights || []
      };
    }
    
    // どちらもない場合
    return {
      text: "データの取得中にエラーが発生しました。",
      data: null,
      insights: []
    };
  } catch (error) {
    console.error("エージェント呼び出しエラー:", error);
    throw new Error("エージェントとの通信中にエラーが発生しました");
  }
}