CData Connect AI を使用して Dataiku と Snowflake のデータを統合

Yazhini G
Technical Marketing Engineer

CData Connect AI のリモート MCP サーバーを活用して、Dataiku エージェントからリアルタイムのSnowflake のデータにセキュアにクエリ・操作できるようにします。

Dataiku は、チームがガバナンスされた環境内で機械学習や生成 AI プロジェクトを設計、デプロイ、管理できるコラボレーティブなデータサイエンス・AI プラットフォームです。エージェントと GenAI フレームワークにより、カスタムワークフローとモデルオーケストレーションを通じてデータを分析、生成、操作できるインテリジェントエージェントを構築できます。

Dataiku を CData Connect AI の組み込み MCP（Model Context Protocol）サーバーと統合することで、これらのエージェントはリアルタイムのSnowflake のデータにセキュアにアクセスできるようになります。この統合により、Dataiku のエージェント実行環境と CData のガバナンスされたエンタープライズ接続レイヤーが橋渡しされ、すべてのクエリや指示が手動エクスポートやステージングなしで、承認されたデータソースに対して安全に実行されます。

この記事では、Connect AI での Snowflake 接続の設定、MCP サポートを含む Dataiku の Python コード環境の準備、そして Dataiku 内から直接リアルタイムのSnowflake のデータにクエリ・操作できるエージェントの作成方法を説明します。

Snowflake データ連携について

CData は、Snowflake のライブデータへのアクセスと統合を簡素化します。お客様は CData の接続機能を以下の目的で活用しています：

Snowflake データを迅速かつ効率的に読み書きできます。
指定された Warehouse、Database、Schema のメタデータを動的に取得できます。
OAuth、OKTA、Azure AD、Azure マネージドサービス ID、PingFederate、秘密鍵など、さまざまな方法で認証できます。

多くの CData ユーザーは、CData ソリューションを使用して、お気に入りのツールやアプリケーションから Snowflake にアクセスし、さまざまなシステムからデータを Snowflake にレプリケートして、包括的なウェアハウジングと分析を行っています。

CData ソリューションとの Snowflake 統合についての詳細は、ブログをご覧ください：https://jp.cdata.com/blog/snowflake-integrations

はじめに

ステップ 1：Dataiku 用の Snowflake 接続を設定

Dataiku から Snowflake への接続は、CData Connect AI のリモート MCP サーバーによって実現されます。Dataiku からSnowflake のデータを操作するには、まず CData Connect AI で Snowflake 接続を作成・設定します。

Connect AI にログインし、Sources をクリック、次に Add Connection をクリック

Add Connection パネルから「Snowflake」を選択

Snowflake に接続するために必要な認証プロパティを入力します。
それでは、Snowflake データベースに接続していきましょう。認証に加えて、以下の接続プロパティを設定します。
- Url：お使いのSnowflake URL を指定します。例：https://orgname-myaccount.snowflakecomputing.com
  - Legacy URL を使用する場合：https://myaccount.region.snowflakecomputing.com
  - ご自身のURL は以下のステップで確認できます。
    1. Snowflake UI の左下にあるユーザー名をクリックします
    2. Account ID にカーソルを合わせます
    3. Copy Account URL アイコンをクリックして、アカウントURL をコピーします
- Database（オプション）：CData 製品によって公開されるテーブルとビューを、特定のSnowflake データベースのものに制限したい場合に設定します
- Schema（オプション）：CData 製品によって公開されるテーブルとビューを、特定のSnowflake データベーススキーマのものに制限したい場合に設定します
Snowflakeへの認証

CData 製品では、Snowflake ユーザー認証、フェデレーション認証、およびSSL クライアント認証をサポートしています。認証するには、User とPassword を設定し、AuthScheme プロパティで認証方法を選択してください。

キーペア認証

ユーザーアカウントに定義されたプライベートキーを使用してセキュアなトークンを作成し、キーペア認証で接続することも可能です。この方法で接続するには、AuthScheme をPRIVATEKEY に設定し、以下の値を設定してください。
- User：認証に使用するユーザーアカウント
- PrivateKey：プライベートキーを含む.pem ファイルへのパスなど、ユーザーに使用されるプライベートキー
- PrivateKeyType：プライベートキーを含むキーストアの種類（PEMKEY_FILE、PFXFILE など）
- PrivateKeyPassword：指定されたプライベートキーのパスワード
その他の認証方法については、ヘルプドキュメントの「Snowflakeへの認証」セクションをご確認ください。
Save & Test をクリック
Permissions タブを開き、ユーザーベースの権限を設定

Personal Access Token を追加

Personal Access Token（PAT）は、Dataiku から Connect AI への接続を認証するために使用されます。きめ細かいアクセス制御を維持するため、統合ごとに個別の PAT を作成することをお勧めします

Connect AI アプリの右上にある歯車アイコン（）をクリックして Settings を開く
Settings ページで Access Tokens セクションに移動し、 Create PAT をクリック
PAT にわかりやすい名前を付けて Create をクリック

トークンが表示されたらコピーして安全な場所に保存してください。再度表示されることはありません

Snowflake 接続の設定と PAT の生成が完了したら、Dataiku から CData MCP Server 経由でSnowflake のデータに接続できます。

ステップ 2：Dataiku とコード環境を準備

Dataiku の専用 Python コード環境が、MCP ベースの通信に必要なランタイムサポートを提供します。Dataiku エージェントを CData Connect AI に接続できるようにするには、Python 環境を作成し、エージェントとサーバー間の対話に必要な MCP クライアント依存関係をインストールします。

Dataiku Cloud で Code Envs を開く

Add a code env をクリックして DSS 設定ウィンドウを開く

DSS で New Python env をクリック。名前を付け（例：MCP_Package）、Python 3.10 を選択（3.10 から 3.13 がサポートされています）

Packages to install を開き、以下の pip パッケージを追加：

httpx
anyio
langchain-mcp-adapters

Containerized execution を開き、Container runtime additions の下で Agent tool MCP servers support を選択

Rebuild env をチェックし、Save and update をクリックしてパッケージをインストール
Dataiku Cloud に戻り、Overview を開いて Open instance をクリック

+ New project をクリックして Blank project を選択。プロジェクト名を入力

ステップ 3：Dataiku エージェントを作成して MCP サーバーに接続

Dataiku エージェントは、Dataiku ワークスペースと CData MCP Server 間のブリッジとして機能します。この接続を有効にするには、カスタムコードベースエージェントを作成し、設定した Python 環境を割り当て、Connect AI の認証情報を埋め込んで、エージェントがリアルタイムのSnowflake のデータにクエリ・操作できるようにします。

Agents & GenAI Models に移動し、Create your first agent をクリック

Code agent を選択し、名前を付け、Agent version で Asynchronous agent without streaming を選択

上部のタブから Settings を選択。Code env selection で Default Python code env を作成した環境（例：MCP_Package）に設定

Agent の Design タブに戻り、以下のコードを貼り付けます。EMAIL と PAT を自分の値に置き換えてください



import os
import base64
from typing import Dict, Any, List

from dataiku.llm.python import BaseLLM
from langchain_mcp_adapters.client import MultiServerMCPClient

# ---------- Persistent MCP client (cached between calls) ----------
_MCP_CLIENT = None

def _get_mcp_client() -> MultiServerMCPClient:
    """Create (or reuse) a MultiServerMCPClient to CData Cloud MCP."""
    global _MCP_CLIENT
    if _MCP_CLIENT is not None:
        return _MCP_CLIENT

    # Set creds via env/project variables ideally
    EMAIL = os.getenv("CDATA_EMAIL", "YOUR_EMAIL")
    PAT   = os.getenv("CDATA_PAT",   "YOUR_PAT")
    BASE_URL = "https://mcp.cloud.cdata.com/mcp"

    if not EMAIL or PAT == "YOUR_PAT":
        raise ValueError("Set CDATA_EMAIL and CDATA_PAT as env variables or inline in the code.")

    token = base64.b64encode(f"{EMAIL}:{PAT}".encode()).decode()
    headers = {"Authorization": f"Basic {token}"}

    _MCP_CLIENT = MultiServerMCPClient(
        connections={
            "cdata": {
                "transport": "streamable_http",
                "url": BASE_URL,
                "headers": headers,
            }
        }
    )
    return _MCP_CLIENT


def _pick_tool(tools, names: List[str]):
    L = [n.lower() for n in names]
    return next((t for t in tools if t.name.lower() in L), None)


async def _route(prompt: str) -> str:
    """
    Simple intent router:
      - 'list connections' / 'list catalogs' -> getCatalogs
      - 'sql: ...' or 'query: ...' -> queryData
      - otherwise -> help text
    """
    client = _get_mcp_client()
    tools = await client.get_tools()

    p = prompt.strip()
    low = p.lower()

    # 1) List connections (catalogs)
    if "list connections" in low or "list catalogs" in low:
        t = _pick_tool(tools, ["getCatalogs", "listCatalogs"])
        if not t:
            return "No 'getCatalogs' tool found on the MCP server."
        res = await t.ainvoke({})
        return str(res)[:4000]

    # 2) Run SQL
    if low.startswith("sql:") or low.startswith("query:"):
        sql = p.split(":", 1)[1].strip()
        t = _pick_tool(tools, ["queryData", "sqlQuery", "runQuery", "query"])
        if not t:
            return "No query-capable tool (queryData/sqlQuery) found on the MCP server."
        try:
            res = await t.ainvoke({"query": sql})
            return str(res)[:4000]
        except Exception as e:
            return f"Query failed: {e}"

    # 3) Help
    return (
        "Connected to CData MCP

"
        "Say **'list connections'** to view available sources, or run a SQL like:
"
        "  sql: SELECT * FROM [Salesforce1].[SYS].[Connections] LIMIT 5

"
        "Remember to use bracket quoting for catalog/schema/table names."
    )


class MyLLM(BaseLLM):
    async def aprocess(self, query: Dict[str, Any], settings: Dict[str, Any], trace: Any):
        # Extract last user message from the Quick Test payload
        prompt = ""
        try:
            prompt = (query.get("messages") or [])[-1].get("content", "")
        except Exception:
            prompt = ""

        try:
            reply = await _route(prompt)
        except Exception as e:
            reply = f"Error: {e}"

        # The template expects a dict with a 'text' key
        return {"text": reply}

クイックテストを実行

右側のパネルで Quick Test を開く
JSON コードを貼り付けて Run test をクリック


{
   "messages": [
      {
         "role": "user",
         "content": "list connections"
      }
   ],
   "context": {}
}

エージェントとチャット

Chat タブに切り替えて、「List all connections」 のようなプロンプトを試してみてください。チャット出力に接続カタログの一覧が表示されます。

CData Connect AI を入手

AI エージェントから 300 以上の SaaS、ビッグデータ、NoSQL ソースにアクセスするには、CData Connect AI をお試しください。

はじめる準備はできましたか？

CData Connect AI の詳細、または無料トライアルにお申し込みください：

無料トライアルお問い合わせ

CData Connect AI を使用して Dataiku と Snowflake のデータを統合

Snowflake データ連携について

はじめに

ステップ 1：Dataiku 用の Snowflake 接続を設定