開発者ガイド - Claude Agent SDK で本番環境対応の AI エージェントを構築

ステートフルな会話、自動コンテキスト管理、ネイティブ MCP ツール統合を備えた本番環境対応の AI エージェントを構築します。 このガイドでは、Claude Agent SDK と CData Connect AI を使用して、データへの会話型アクセスを可能にする Python アプリケーションを作成する方法を説明します。

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

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

  • CData Connect AI を通じて 350 以上のデータソースに接続
  • 自動コンテキスト管理によるステートフルな会話の維持
  • 自然言語を使用した SQL クエリの実行
  • ターン間でコンテキストが保持されるマルチターン会話の処理
  • Claude Code を支える同じフレームワークからの本番環境対応パターンの使用

なぜ Claude Agent SDK を使うのか？

Claude Agent SDK は、AI エージェントを構築するためのいくつかの重要な利点を提供します：

  • 自動コンテキスト管理: 手動での会話履歴の追跡が不要 - SDK が自動的に処理します
  • ステートフルな会話: 複数のターンにわたってコンテキストを維持し、自然で流れるようなインタラクションを実現
  • ネイティブツール統合: ツールは自動呼び出しライフサイクルとエラーハンドリングを備えた第一級オブジェクトです
  • コンテキスト圧縮: トークン制限に近づくと自動的にコンテキストを圧縮
  • 本番環境対応: Claude Code を支える同じエージェントハーネス上に構築
  • Async/Await アーキテクチャ: パフォーマンス向上のためのモダンな Python 非同期パターン

アーキテクチャの概要

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

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

仕組み：

  1. Python アプリケーションが HTTP 経由で CData Connect AI の MCP サーバーに接続
  2. MCP ツールが検出され、Claude Agent SDK 用にラップされます
  3. Agent SDK がステートフルな会話とツール実行を管理
  4. Claude が自然言語を SQL クエリに変換
  5. 結果が返され、AI によって解釈されます

前提条件

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


はじめに

概要

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

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

mkdir connectai-claude-agent cd connectai-claude-agent pip install claude-agent-sdk python-dotenv requests

3.3 環境変数の設定

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

# Anthropic 設定 ANTHROPIC_API_KEY=your_anthropic_api_key # CData Connect AI 設定 [email protected] CDATA_ACCESS_TOKEN=your_personal_access_token

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


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

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

4.1 MCPClient クラス

CData Connect AI MCP サーバーとの HTTP 通信を処理します。Basic 認証を使用した認証を管理し、Server-Sent Events（SSE）レスポンスを解析します：

import json import base64 import sys import requests class MCPClient: """HTTP 経由で CData Connect AI MCP サーバーと対話するクライアント。""" def __init__(self, server_url: str, email: str = None, access_token: str = None): self.server_url = server_url.rstrip('/') self.session = requests.Session() # MCP JSON-RPC 用のデフォルトヘッダーを設定 self.session.headers.update({ 'Content-Type': 'application/json', 'Accept': 'application/json, text/event-stream', 'User-Agent': f'CDataConnectAI-ClaudeAgent (Python/{sys.version_info.major}.{sys.version_info.minor})', }) if email and access_token: # Basic 認証: email:personal_access_token credentials = f"{email}:{access_token}" encoded_credentials = base64.b64encode(credentials.encode()).decode() self.session.headers.update({ 'Authorization': f'Basic {encoded_credentials}' }) def _parse_sse_response(self, response_text: str) -> dict: """Server-Sent Events（SSE）レスポンスを解析。""" 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.session.post( self.server_url, json={"jsonrpc": "2.0", "method": "tools/list", "params": {}, "id": 1} ) response.raise_for_status() 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.session.post( self.server_url, json={ "jsonrpc": "2.0", "method": "tools/call", "params": {"name": tool_name, "arguments": arguments}, "id": 2 } ) response.raise_for_status() result = self._parse_sse_response(response.text) return result.get("result", {})

4.2 MCPAgentChatbot クラス

アプリケーションの中核部分 - MCP クライアントを Claude Agent SDK に接続し、ステートフルな会話と自動コンテキスト管理を実現します：

from typing import Optional, Dict, Any from functools import partial from claude_agent_sdk import query, ClaudeSDKClient, ClaudeAgentOptions, tool, create_sdk_mcp_server class MCPAgentChatbot: """CData Connect AI 用の本番環境対応 AI エージェントチャットボット。""" def __init__(self, mcp_server_url: str, email: str = None, access_token: str = None): self.mcp_client = MCPClient(mcp_server_url, email, access_token) # MCP サーバーから利用可能なツールを読み込み print("CData Connect AI MCP サーバーに接続中...") self.mcp_tools_list = self.mcp_client.list_tools() print(f"MCP サーバーから {len(self.mcp_tools_list)} 個のツールを読み込みました") # Agent SDK ツールラッパーを作成 self.agent_tools = self._create_agent_tools() # Agent SDK 用の MCP サーバーを作成 self.mcp_server = create_sdk_mcp_server( name="cdata_connect", tools=self.agent_tools ) async def _tool_handler(self, tool_name: str, args: Dict[str, Any]) -> Dict[str, Any]: """MCP ツールを呼び出し、結果を返す。""" result = self.mcp_client.call_tool(tool_name, args) return { "content": [{ "type": "text", "text": json.dumps(result, indent=2) }] } def _create_agent_tools(self) -> list: """MCP ツール用の Agent SDK ツールラッパーを作成。""" agent_tools = [] for tool_info in self.mcp_tools_list: tool_name = tool_info.get("name") tool_description = tool_info.get("description", "") tool_schema = tool_info.get("inputSchema", {}) agent_tool = tool( name=tool_name, description=tool_description, input_schema=tool_schema )(partial(self._tool_handler, tool_name)) agent_tools.append(agent_tool) return agent_tools

4.3 ステートフルセッション管理

Agent SDK は、自動コンテキスト管理を備えたステートフルな会話セッションを提供します：

def create_session(self) -> ClaudeSDKClient: """ステートフルな会話セッションを作成。""" options = ClaudeAgentOptions( system_prompt="あなたは CData Connect AI データソースにアクセスできる便利なアシスタントです。", mcp_servers={"cdata_connect": self.mcp_server}, permission_mode="bypassPermissions" ) return ClaudeSDKClient(options=options) async def chat_session(self, client: ClaudeSDKClient, user_message: str) -> str: """ステートフルセッションでメッセージを送信。""" await client.query(user_message) async for message in client.receive_response(): if hasattr(message, 'result'): return str(message.result) return ""

ステップ 5: エージェントチャットボットの構築

ステートフルセッションを備えた対話型エージェントチャットボットを作成します：

#!/usr/bin/env python3 """AI でデータをクエリするための対話型エージェントチャットボット。""" import os import asyncio from dotenv import load_dotenv load_dotenv() async def interactive_mode(chatbot): """ステートフルセッションで対話モードでチャットボットを実行。""" print(" チャットボット準備完了！終了するには 'quit' と入力してください。 ") # ステートフルセッションを作成 client = chatbot.create_session() # 適切なリソースクリーンアップのために async コンテキストマネージャーを使用 async with client: while True: user_input = input("あなた: ").strip() if not user_input: continue if user_input.lower() in ['quit', 'exit', 'q']: print("さようなら！") break response = await chatbot.chat_session(client, user_input) print(f" アシスタント: {response} ") async def main(): """対話モードでチャットボットを実行。""" MCP_SERVER_URL = "https://mcp.cloud.cdata.com/mcp/" CDATA_EMAIL = os.environ.get("CDATA_EMAIL") CDATA_ACCESS_TOKEN = os.environ.get("CDATA_ACCESS_TOKEN") print("=== CData Connect AI エージェントチャットボット ===") print(f"MCP サーバー: {MCP_SERVER_URL} ") chatbot = MCPAgentChatbot(MCP_SERVER_URL, CDATA_EMAIL, CDATA_ACCESS_TOKEN) await interactive_mode(chatbot) if __name__ == "__main__": asyncio.run(main())

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

すべての設定が完了したら、エージェントチャットボットを実行します：

python agent_chatbot.py

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

=== CData Connect AI エージェントチャットボット ===
MCP サーバー: https://mcp.cloud.cdata.com/mcp/

CData Connect AI MCP サーバーに接続中...
MCP サーバーから 8 個のツールを読み込みました
  利用可能なツール:
    - getCatalogs
    - getSchemas
    - getTables
    - getColumns
    - queryData
    ... その他 3 個

チャットボット準備完了！終了するには 'quit' と入力してください。

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

ステップ 7: クエリの例

エージェントチャットボットで試せるプロンプトの例をいくつか紹介します：

データの検出

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

基本的なクエリ

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

マルチターン会話

Agent SDK はターン間でコンテキストを維持するため、自然なフォローアップの質問が可能です：

  • あなた: 「accounts テーブルを見せてください」
  • あなた: 「どれが最も収益が高いですか？」
  • あなた: 「その会社についてもっと教えてください」

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

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

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

ステップ 9: 権限モード

Claude Agent SDK は、ツール実行のために異なる権限モードを提供しています：

モード説明
defaultツールを実行する前にユーザーの承認を求める
acceptEdits編集操作を自動的に承認
bypassPermissionsすべてのツールを自動承認（CLI アプリケーションに推奨）
planエージェントが実行前にアクションを計画

トラブルシューティング

認証エラー

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

ツールが利用できない

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

Agent SDK エラー

  • claude-agent-sdk がインストールされていることを確認: pip install claude-agent-sdk
  • Python 3.8+ を使用していることを確認
  • ANTHROPIC_API_KEY が正しく設定されていることを確認

次のステップ

動作する AI エージェントができたので、以下のことができます：

  • より多くのデータソースを接続: Salesforce、Snowflake、またはサポートされている 350 以上のソースを追加して、データアクセスを拡張します。
  • エージェントをカスタマイズ: 特定のユースケースやドメインに合わせてシステムプロンプトを変更します。
  • 本番アプリケーションを構築: エージェントを Web アプリ、Slack ボット、またはその他のインターフェースに統合します。
  • 高度な機能を探索: カスタマイズ用のフック、リアルタイム応答用のストリーミング、長い会話用のコンテキスト圧縮を使用します。

リソース


CData Connect AI を始めよう

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

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