開発者ガイド - OpenAI Python SDK で AI 搭載データアシスタントを構築

自然言語を使用してライブデータソースをクエリできるインテリジェントなデータアシスタントを構築します。このガイドでは、OpenAI の GPT モデルと CData Connect AI を組み合わせて、データへの会話型アクセスを可能にする Python アプリケーションを作成する方法を説明します。

注意: このガイドでは Google スプレッドシートをデータソースとして使用していますが、CData Connect AI がサポートする 350 以上のデータソースのいずれにも同じ原則が適用されます。

このガイドを完了すると、以下の機能を持つ Python アプリケーションが完成します：

CData Connect AI を通じて 350 以上のデータソースに接続
利用可能なデータベース、スキーマ、テーブルを自動的に検出
自然言語を使用した SQL クエリの実行
コンテキストを持つマルチターン会話の維持
リアルタイムでのレスポンスストリーミング

アーキテクチャの概要

このアプリケーションは、Model Context Protocol（MCP）を使用して OpenAI の GPT モデルとデータソースを橋渡しします：

┌─────────────────┐     ┌──────────────────┐     ┌─────────────────┐
│                 │     │                  │     │                 │
│  Python         │---->│  CData Connect   │---->│  データソース    │
│  アプリケーション │     │  AI MCP サーバー  │     │  (300 種類以上)  │
│                 │<----│                  │<----│                 │
└─────────────────┘     └──────────────────┘     └─────────────────┘
        |                        |
        |    ツールの検出         |
        |    & 実行             |
        v                        |
┌─────────────────┐              |
│                 │              |
│  OpenAI API     │--------------┘
│  (GPT-4 など)    │   自然言語から
│                 │   SQL への変換
└─────────────────┘

仕組み：

Python アプリケーションが CData Connect AI の MCP サーバーに接続
MCP サーバーがデータ検出とクエリ用のツールを公開
OpenAI の GPT モデルがこれらのツールを使用して自然言語を SQL に変換
クエリがライブデータソースに対して実行
結果が返され、AI によって解釈されます

前提条件

このガイドには以下が必要です：

システムにインストールされた Python 3.9+
OpenAI API キー（有料アカウントが必要）
CData Connect AI アカウント（無料トライアル利用可能）
サンプル Google スプレッドシートデータ用の Google アカウント

はじめに

概要

手順の概要は以下の通りです：

Google スプレッドシートでサンプルデータをセットアップ
CData Connect AI を設定し、パーソナルアクセストークンを作成
Python プロジェクトをセットアップし、依存関係をインストール
チャットアプリケーションを構築して実行

ステップ 1: Google スプレッドシートでサンプルデータをセットアップ

機能をデモンストレーションするために、顧客データを含むサンプル Google スプレッドシートを使用します。このデータセットには、アカウント、販売機会、サポートチケット、使用状況メトリクスが含まれています。

サンプル顧客ヘルススプレッドシートに移動します
ファイル > コピーを作成をクリックして、Google ドライブに保存します
わかりやすい名前を付けます（例：「demo_organization」）- これは後で必要になります

スプレッドシートには 4 つのシートが含まれています：

account: 会社情報（名前、業界、収益、従業員数）
opportunity: 販売パイプラインデータ（ステージ、金額、確率）
tickets: サポートチケット（優先度、ステータス、説明）
usage: 製品使用状況メトリクス（ジョブ実行回数、処理レコード数）

ステップ 2: CData Connect AI の設定

2.1 サインアップまたはログイン

https://www.cdata.com/ai/signup/ で新規アカウントを作成するか、 https://cloud.cdata.com/ でログインします
新規アカウントを作成する場合は、登録プロセスを完了します

2.2 Google スプレッドシート接続を追加

ログイン後、左側のナビゲーションメニューでSourcesをクリックし、Add Connectionをクリックします
Add Connection パネルからGoogle Sheetsを選択します
接続を設定します：
- Spreadsheetプロパティに、コピーしたシートの名前を設定します（例：「demo_organization」）
- Sign inをクリックして Google OAuth で認証します
認証後、Permissionsタブに移動し、ユーザーがアクセス権を持っていることを確認します

2.3 パーソナルアクセストークンの作成

Python アプリケーションは、パーソナルアクセストークン（PAT）を使用して Connect AI と認証します。

右上の歯車アイコンをクリックして設定を開きます
Access Tokensセクションに移動します
Create PATをクリックします
トークンに名前を付け（例：「OpenAI Python App」）、Createをクリックします
重要: トークンはすぐにコピーしてください - 一度しか表示されません！

ステップ 3: Python プロジェクトのセットアップ

3.1 GitHub からクローン（推奨）

すべてのサンプルを含む完全なプロジェクトをクローンします：


git clone https://github.com/CDataSoftware/connectai-openai-agent.git
cd connectai-openai-agent
pip install -r requirements.txt

3.2 代替方法: ゼロから作成

新しいプロジェクトディレクトリを作成し、依存関係をインストールします：


mkdir connectai-openai-agent
cd connectai-openai-agent
pip install openai httpx python-dotenv

3.3 環境変数の設定

プロジェクトルートに .env ファイルを作成します：


# OpenAI 設定
OPENAI_API_KEY=sk-your-openai-api-key-here
OPENAI_MODEL=gpt-4o

# CData Connect AI 設定
[email protected]
CDATA_PAT=your-personal-access-token-here
MCP_SERVER_URL=https://mcp.cloud.cdata.com/mcp

プレースホルダーの値を実際の認証情報に置き換えてください。

ステップ 4: コードアーキテクチャの理解

プロジェクトは 3 つの主要コンポーネントで構成されています：

4.1 Config クラス

設定と認証情報の管理を処理します：


from dataclasses import dataclass
import os
import base64

@dataclass
class Config:
    """Connect AI OpenAI エージェントの設定。"""
    openai_api_key: str
    cdata_email: str
    cdata_pat: str
    openai_model: str = "gpt-4o"
    mcp_server_url: str = "https://mcp.cloud.cdata.com/mcp"

    @classmethod
    def from_env(cls) -> "Config":
        """環境変数から設定を作成。"""
        return cls(
            openai_api_key=os.getenv("OPENAI_API_KEY"),
            cdata_email=os.getenv("CDATA_EMAIL"),
            cdata_pat=os.getenv("CDATA_PAT"),
            openai_model=os.getenv("OPENAI_MODEL", "gpt-4o"),
            mcp_server_url=os.getenv("MCP_SERVER_URL", "https://mcp.cloud.cdata.com/mcp"),
        )

    def get_auth_header(self) -> str:
        """Base64 認証ヘッダーを生成。"""
        credentials = f"{self.cdata_email}:{self.cdata_pat}"
        return base64.b64encode(credentials.encode()).decode()

4.2 MCPClient クラス

CData Connect AI MCP サーバーとの HTTP 通信を処理します：


import json
import sys
import httpx

class MCPClient:
    """CData Connect AI MCP サーバー用の HTTP クライアント。"""

    def __init__(self, config: Config):
        self.config = config
        self._client = httpx.Client(
            headers={
                "Content-Type": "application/json",
                "Accept": "application/json, text/event-stream",
                "Authorization": f"Basic {config.get_auth_header()}",
                "User-Agent": f"CDataConnectAI-OpenAIAgent (Python/{sys.version_info.major}.{sys.version_info.minor})",
            },
            timeout=60.0,
        )

    def _parse_sse_response(self, response_text: str) -> dict:
        """Server-Sent Events レスポンスを解析。"""
        for line in response_text.split("
"):
            if line.startswith("data: "):
                return json.loads(line[6:])
        raise ValueError("SSE レスポンスにデータが見つかりません")

    def list_tools(self) -> list:
        """MCP サーバーから利用可能なツールを検出。"""
        response = self._client.post(
            self.config.mcp_server_url,
            json={"jsonrpc": "2.0", "method": "tools/list", "params": {}, "id": 1},
        )
        result = self._parse_sse_response(response.text)
        return result.get("result", {}).get("tools", [])

    def call_tool(self, tool_name: str, arguments: dict) -> dict:
        """MCP サーバーでツールを実行。"""
        response = self._client.post(
            self.config.mcp_server_url,
            json={
                "jsonrpc": "2.0",
                "method": "tools/call",
                "params": {"name": tool_name, "arguments": arguments},
                "id": 2,
            },
        )
        result = self._parse_sse_response(response.text)
        return result.get("result", \{})

4.3 MCPAgent クラス

OpenAI と MCP ツールを組み合わせた AI エージェント：


from openai import OpenAI

class MCPAgent:
    """CData Connect AI MCP ツールを使用した OpenAI AI エージェント。"""

    def __init__(self, config: Config):
        self.config = config
        self.mcp_client = MCPClient(config)
        self.openai_client = OpenAI(api_key=config.openai_api_key)
        self.conversation_history = []
        self._tools_cache = []

    def _get_openai_tools(self) -> list:
        """MCP ツールを OpenAI 関数形式に変換。"""
        if not self._tools_cache:
            mcp_tools = self.mcp_client.list_tools()
            for tool in mcp_tools:
                self._tools_cache.append({
                    "type": "function",
                    "function": {
                        "name": tool.get("name"),
                        "description": tool.get("description", ""),
                        "parameters": tool.get("inputSchema", {}),
                    },
                })
        return self._tools_cache

    def chat(self, message: str) -> str:
        """メッセージを送信してレスポンスを取得。"""
        self.conversation_history.append({"role": "user", "content": message})

        messages = [
            {"role": "system", "content": "あなたはデータアシスタントです..."},
            *self.conversation_history,
        ]

        tools = self._get_openai_tools()

        # 最終レスポンスを取得するまで処理
        for _ in range(10):
            response = self.openai_client.chat.completions.create(
                model=self.config.openai_model,
                messages=messages,
                tools=tools if tools else None,
            )

            assistant_message = response.choices[0].message

            if assistant_message.tool_calls:
                messages.append(assistant_message.model_dump())
                for tool_call in assistant_message.tool_calls:
                    result = self.mcp_client.call_tool(
                        tool_call.function.name,
                        json.loads(tool_call.function.arguments)
                    )
                    messages.append({
                        "role": "tool",
                        "tool_call_id": tool_call.id,
                        "content": json.dumps(result),
                    })
            else:
                content = assistant_message.content or ""
                self.conversation_history.append({"role": "assistant", "content": content})
                return content

        return "最大反復回数に達しました。"

ステップ 5: チャットアプリケーションの構築

対話型チャットアプリケーションを作成します：


#!/usr/bin/env python3
"""AI でデータをクエリするための対話型チャットアプリケーション。"""

from dotenv import load_dotenv
from src.connectai_openai import Config, MCPAgent

load_dotenv()

def main():
    print("=" * 60)
    print("CData Connect AI - OpenAI チャットアシスタント")
    print("=" * 60)

    # 初期化
    config = Config.from_env()
    agent = MCPAgent(config)
    tools = agent._get_openai_tools()
    print(f"
接続完了！{len(tools)} 個のツールが利用可能です。")

    print("
データとチャットしましょう！終了するには 'quit' と入力してください。
")

    while True:
        user_input = input("あなた: ").strip()

        if user_input.lower() in ("quit", "exit"):
            print("さようなら！")
            break

        response = agent.chat(user_input)
        print(f"
アシスタント: {response}
")

if __name__ == "__main__":
    main()

ステップ 6: アプリケーションの実行

すべての設定が完了したら、チャットアプリケーションを実行します：


python examples/basic_chat.py

以下のような出力が表示されます：

============================================================
CData Connect AI - OpenAI チャットアシスタント
============================================================

接続完了！8 個のツールが利用可能です。

データとチャットしましょう！終了するには 'quit' と入力してください。

あなた: どんなデータソースがありますか？

アシスタント: 以下のデータソースが接続されています：

1. **demo_organization** (Google スプレッドシート)
   - テーブル: account, opportunity, tickets, usage

詳しく調べたいものはありますか？

ステップ 7: クエリの例

データアシスタントで試せるプロンプトの例をいくつか紹介します：

データの検出

「接続されているデータソースは何ですか？」
「demo_organization のすべてのテーブルを見せてください」
「account テーブルにはどんなカラムがありますか？」

基本的なクエリ

「annual_revenue でソートした上位 5 件のアカウントをクエリしてください」
「優先度別のサポートチケット数を教えてください」
「オープン中のすべての商談を表示してください」

分析

「最も多くの重要なサポートチケットを持つアカウントはどれですか？」
「Aurora Healthcare Systems のヘルス状態を要約してください」
「収益は高いが製品使用率が低いアカウントを見つけてください」

ステップ 8: 利用可能な MCP ツール

AI エージェントは、以下の CData Connect AI ツールにアクセスできます：

ツール	説明
`getCatalogs`	利用可能なデータソース接続を一覧表示
`getSchemas`	特定のカタログのスキーマを取得
`getTables`	スキーマ内のテーブルを取得
`getColumns`	テーブルのカラムメタデータを取得
`queryData`	SQL クエリを実行
`getProcedures`	ストアドプロシージャを一覧表示
`getProcedureParameters`	プロシージャのパラメータ詳細を取得
`executeProcedure`	ストアドプロシージャを実行

ステップ 9: SQL クエリの形式

AI が SQL クエリを生成する際、完全修飾テーブル名を使用します：


SELECT [column1], [column2]
FROM [CatalogName].[SchemaName].[TableName]
WHERE [column1] = 'value'
ORDER BY [column2]
LIMIT 100

例えば、Google スプレッドシートの account テーブルをクエリするには：


SELECT [name], [annual_revenue], [industry]
FROM [demo_organization].[GoogleSheets].[account]
ORDER BY [annual_revenue] DESC
LIMIT 10

トラブルシューティング

認証エラー

.env の CData メールアドレスと PAT が正しいことを確認
PAT が期限切れになっていないことを確認
Connect AI アカウントがアクティブであることを確認

ツールが利用できない

Connect AI に少なくとも 1 つのデータソースが接続されていることを確認
ユーザーが接続にアクセスする権限を持っていることを確認

クエリエラー

完全修飾テーブル名を使用: [Catalog].[Schema].[Table]
getColumns ツールを使用してカラム名が存在することを確認
SQL 構文を確認（Connect AI は SQL-92 標準を使用）

次のステップ

動作する AI データアシスタントができたので、以下のことができます：

より多くのデータソースを接続: Salesforce、Snowflake、またはサポートされている 350 以上のソースを追加して、データアクセスを拡張します。
エージェントをカスタマイズ: 特定のユースケースやドメインに合わせてシステム指示を変更します。
本番アプリケーションを構築: エージェントを Web アプリ、Slack ボット、またはその他のインターフェースに統合します。
ストリーミングを追加: リアルタイムのレスポンスストリーミングのために chat_stream() を使用します。
サンプルを探索: GitHub リポジトリの追加サンプルで、データ分析ワークフロー、マルチソースクエリなどを確認してください。

リソース

GitHub リポジトリ - 完全なソースコードとサンプル
CData Connect AI ドキュメント
CData プロンプトライブラリ - さまざまなユースケース向けのサンプルプロンプト
OpenAI API ドキュメント
Model Context Protocol

CData Connect AI を始めよう

AI 搭載のデータアプリケーションを構築する準備はできましたか？CData Connect AI は、AI アプリケーションから 350 以上の SaaS、ビッグデータ、NoSQL ソースへのライブデータアクセスを提供します。

無料トライアルにサインアップして、今日からインテリジェントなデータアシスタントの構築を始めましょう！

Feedback