【徹底解説】Agentforce とHive のデータをMCP Server で簡単連携
Agentforce は、エンタープライズチーム向けに設計されたAI プラットフォームです。宣言的なツールとAPI ファーストなアーキテクチャで、アシスタントの構成やCRM ベースのワークフロー自動化を簡単に実現できます。
しかし、エージェントがHive 以外のデータを扱う必要がある場合、多くの実装では定期的な同期や独自のミドルウェアを使用して外部システムをローカルデータベースに複製しています。この方法では、複雑性の増加、ガバナンスとメンテナンスのオーバーヘッド、同期の遅延が発生し、エージェントのリアルタイム性能が制限されてしまいます。
CData Connect AIは、300以上のビジネスアプリケーション、データベース、ERP、会計ツールへのリアルタイムアクセスを提供することで、このギャップを埋めます。CData のリモートMCP Serverを使用することで、Agentforce エージェントは実行時に最新のコンテキストに沿ったデータをセキュアに読み書きし、アクションを実行できます。そのため、より少ない構成要素で、より強力な制御のもと、根拠に基づいた応答、リアルタイムの意思決定、システム間の自動化を実現できます。
本記事では、Connect AI でHive への接続を設定し、MCP Server をAgentforce に登録し、Hive データをクエリするワークフローを構築するために必要な手順を説明します。
前提条件:
- CData Connect AI アカウント
- Python 3.9 以上のインストール
- Agentforce が有効化されたHive アカウントと、接続アプリの作成/管理権限
必要な認証情報のチェックリスト
開始する前に、以下の認証情報を準備してください。
CData Connect AI MCP
- 「USERNAME」: アカウントのメールアドレス
- 「PAT:」 Connect AIの「Settings」から「アクセストークン」をクリック（一度だけコピー可能）
- 「MCP_BASE_URL: 」https://mcp.cloud.cdata.com/mcp
- 「MCP_AUTH: 」Base64エンコードされた(EMAIL:PAT)。EMAILはConnect AIのメールアドレス、PATはConnect AIのパーソナルアクセストークンです（下記参照）。
MCP_AUTH のBase64 エンコード版は、Base64 エンコーディングツールを使用して作成できます。
Agentforce
- 「SFDC_DOMAIN」: ドメインのURL (https://your_domain.my.salesforce.com)
- 「SFDC_CLIENT_ID & SFDC_CLIENT_SECRET」: 接続アプリから取得
- 「AGENT_ID」: Agentforce Agents でエージェントを開き、URL の末尾にある18文字のIDをコピー
ステップ1: Agentforce 連携用のHive への接続を設定
それでは、Agentforce からCData Connect AI のリモートMCP 経由でHive への接続を設定していきましょう。Agentforce からHive を操作するには、まずCData Connect AI でHive へのコネクションを作成し、設定します。
- Connect AI にログインして「Sources」をクリック。次に「Add Connection」をクリックします。
- 一覧から「Hive」を選択します。
-
Hive に接続するために必要な認証プロパティを入力します。
Apache Hive への接続を確立するには以下を指定します。
- Server：HiveServer2 をホストするサーバーのホスト名またはIP アドレスに設定。
- Port：HiveServer2 インスタンスへの接続用のポートに設定。
- TransportMode：Hive サーバーとの通信に使用するトランスポートモード。有効な入力値は、BINARY およびHTTP です。デフォルトではBINARY が選択されます。
- AuthScheme：使用される認証スキーム。有効な入力値はPLAIN、LDAP、NOSASL、およびKERBEROS です。デフォルトではPLAIN が選択されます。
- CData 製品においてTLS/SSL を有効化するには、UseSSL をTrue に設定します
- 「 Save & Test」をクリック。
-
Hive ページの「Permissions」タブに移動し、ユーザーごとの権限を更新します。
パーソナルアクセストークンの追加
パーソナルアクセストークン（PAT）は、Agentforce からConnect AI への接続を認証するために使用されます。アクセスの粒度を維持するため、サービスごとに個別のPAT を作成するのがベストプラクティスです。
- Connect AI アプリの右上にある歯車アイコン（）をクリックして、設定ページを開きます。
- 設定ページで、アクセストークンセクションに移動し、PATを作成をクリックします。
-
PATに名前を付けて、作成をクリックします。
- パーソナルアクセストークンは作成時のみ表示されるため、必ずコピーして安全に保管してください。
接続が設定され、PAT が生成されたので、Agentforce からSalesforce に接続する準備が整いました。
ステップ2: Agentforce を有効化し、Experience Cloud サイトを公開
- Salesforce の設定ページで「Einstein Setup」に移動し、「Turn on Einstein」トグルを有効化します。
- Agentforce Agents に移動し、「Agentforce」をオンにします。
- 次に、Coral Cloud をホストする「Experience Cloud サイト」を公開します。 設定で「すべてのサイト」を検索し、「Builder（for coral-cloud）」を選択して「Publish」をクリックし、「Confirm」をクリックします。
- Experience Builder メニューをクリック
- Salesforce Settings をクリック
- ブラウザを更新して設定を再読み込みします。Experience Site のブラウザタブは閉じてもOKです。
ステップ3: EinsteinServiceAgentユーザーとプロファイルを設定する
- 設定のクイック検索で「Users」を検索して選択
- EinsteinServiceAgent ユーザーの横にある編集をクリック
- プロファイルを「Einstein Agent User」に更新
- その他の設定はそのまま
- 「保存」をクリック
ステップ4: エージェントを作成し、外部接続をアタッチ
- 設定で「Agentforce Agents」を検索し、「新しいエージェント」をクリックします。
- 「Create with Gen AI」をクリック
- 概要、名前、役割、会社情報を入力します。Agent User = EinsteinServiceAgent Userと設定しましょう。
- 「Keep a record of conversations with enhanced event logs」にチェックを入れ、「Save」をクリックします。
- 「Create」をクリック
- 「Connections」を追加し、「Needs Setup」を選択して「External App」をクリック
- Connection Type = API を設定し、「Integratoin Name」に任意の名前を入力します。
- 「新しい接続アプリ」をクリック（次のステップで設定します）
ステップ5: Salesforce接続アプリを作成・設定する（クライアント認証情報）
- 接続アプリの作成を許可します。
- 「New Connected App」をクリックし、「Name」と「Admin Email」を入力
- 「Enable OAuth Settings」をオンにしてCallback URL: https://login.salesforce.com に設定しましょう。
- OAuth スコープは、以下をすべて追加してください。
- Access chatbot services (chatbot_api)
- Access the Salesforce API Platform (sfap_api)
- Manage user data via APIs (api)
- Perform requests at any time (refresh_token, offline_access)
- 以下の選択を解除してください。
- PKCE
- Require Secret for Web Server Flow
- Require Secret for Refresh Token Flow
- 以下を選択してください。
- Enable Client Credentials Flow
- Issue JWT-based access tokens for named users.
- Enable the Save option
- アプリの「Manage」ページで「Edit Policies」をクリック
- 「Permitted Users」は、自社のポリシーに合わせて設定してください。
- 「Client Credentials Flow」でRun As: choose a user with at least API Only access を選択します。
- 「JWT-based Access Tokens」を有効化し、Token Timeout: 30 minutesを設定して「保存」をクリックします。
- 最後に、「Agent Connection」に戻り、今作成したアプリを選択して「Save」をクリックします。
- 「App Manager」から「Consumer Key」と「Consumer Secret」を取得
- 作成したアプリを見つけて、「Manage Consumer Details」を表示します。
- 「Consumer Key」と「Consumer Secret」をコピー
ステップ6：連携に必要な残りのID を入手
- 「My Domain」: 設定でMy Domain に移動し、トークンリクエスト用の...my.salesforce.com URLをコピーします。
- 「AGENT_ID」: 「Agentforce Agents」でエージェントを開き、URL の末尾から18文字のID をコピー
ステップ7: Python プロジェクトをセットアップ
まずは、必要となるPython パッケージをインストールしていきます。
pip install requests asyncio langchain-mcp-adapters
次に、Connect AIおよびAgentforce への接続に必要となる「config.py」を作成します（プレースホルダーを実際の値に置き換えてください）。以下を定義します。
- 「MCP」: USERNAME、PAT、MCP_BASE_URL、およびMCP_AUTH = base64(username:pat)を計算
- 「Salesforce」: SFDC_DOMAIN、SFDC_CLIENT_ID、SFDC_CLIENT_SECRET、AGENT_ID
import base64 --- MCP (CData Connect AI) --- EMAIL = "[email protected]" PAT = "your_PAT" MCP_BASE_URL = "https://mcp.cloud.cdata.com/mcp" MCP_AUTH = base64.b64encode(f"{EMAIL}:{PAT}".encode()).decode() --- Salesforce Agentforce --- SFDC_DOMAIN = "https://your_domain.my.salesforce.com" SFDC_CLIENT_ID = "your_SFDC_CLIENT_ID" SFDC_CLIENT_SECRET = "your_SFDC_CLIENT_SECRET" AGENT_ID = "your_AGENT_ID"
連携を実行するためのPython スクリプト（フローと主要な関数）
以下に、Connect AI 経由でHive とAgentforce を連携してAgentforce がHive のデータを取得するPython スクリプトを用意しました。Python スクリプトは3つの内容で構成されています。
- Salesforce / Einstein Agent関数
- get_salesforce_token() - OAuth2トークンを取得
- start_agent_session() - エージェントセッションを開始
- post_message_to_agent() - エージェントにメッセージを送信
- MCP関数
- query_mcp() - カタログを取得し、MCPツールからサンプルクエリを実行
- 全体のオーケストレーション
- 「MCP関数」を呼び出し
- 「Salesforceトークン」を取得し、「エージェントセッション」を開始
- 「MCP結果」を推論のためにエージェントに送信
- 「エージェントの応答」を出力
スクリプトを実行するには、スクリプトが保存されているディレクトリでターミナルを開き、以下を実行します。
python agentforce_script.py
agentforce_script.py
import asyncio
import requests
import time
import uuid
from langchain_mcp_adapters.client import MultiServerMCPClient
from config import SFDC_DOMAIN, SFDC_CLIENT_ID, SFDC_CLIENT_SECRET, AGENT_ID, MCP_BASE_URL, MCP_AUTH
---------------- Salesforce / Einstein Agent ----------------
def get_salesforce_token():
"""クライアント認証情報を使用して新しいSalesforce OAuthトークンを取得します。"""
print(" 新しいSalesforceトークンをリクエスト中...")
token_url = f"{SFDC_DOMAIN}/services/oauth2/token"
data = {
"grant_type": "client_credentials",
"client_id": SFDC_CLIENT_ID,
"client_secret": SFDC_CLIENT_SECRET,
}
headers = {"Content-Type": "application/x-www-form-urlencoded"}
resp = requests.post(token_url, data=data, headers=headers)
resp.raise_for_status()
token_data = resp.json()
print(" アクセストークンを取得しました。")
return token_data["access_token"], token_data["instance_url"]
def start_agent_session(access_token):
"""Salesforce Einstein Agentとのセッションを開始します。"""
session_url = f"https://api.salesforce.com/einstein/ai-agent/v1/agents/{AGENT_ID}/sessions"
payload = {
"externalSessionKey": str(uuid.uuid4()),
"instanceConfig": {"endpoint": SFDC_DOMAIN},
"streamingCapabilities": {"chunkTypes": ["Text"]},
"bypassUser": True,
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}",
}
resp = requests.post(session_url, json=payload, headers=headers)
resp.raise_for_status()
session_data = resp.json()
print(f" エージェントセッションを開始しました: {session_data['sessionId']}")
return session_data["sessionId"]
def post_message_to_agent(access_token, session_id, message_text):
"""Salesforce Einstein Agentセッションにメッセージを送信します。"""
messages_url = f"https://api.salesforce.com/einstein/ai-agent/v1/sessions/{session_id}/messages"
payload = {
"message": {
"sequenceId": int(time.time() * 1000),
"type": "Text",
"text": message_text,
},
"variables": [],
}
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {access_token}",
}
resp = requests.post(messages_url, json=payload, headers=headers)
resp.raise_for_status()
return resp.json()
---------------- MCP部分 ----------------
async def query_mcp():
"""カタログとサンプルクエリをMCP Serverから取得します。"""
mcp_client = MultiServerMCPClient(
connections={
"default": {
"transport": "streamable_http",
"url": MCP_BASE_URL,
"headers": {"Authorization": f"Basic {MCP_AUTH}"},
}
}
)
tools = await mcp_client.get_tools()
print("検出されたMCPツール:", [t.name for t in tools])
# カタログのリスト取得
get_catalogs_tool = next(t for t in tools if t.name == "getCatalogs")
catalogs = await get_catalogs_tool.ainvoke({})
print("カタログ:", catalogs)
# サンプルクエリの実行
query_tool = next(t for t in tools if t.name == "queryData")
query_result = await query_tool.ainvoke({
"query": "SELECT * FROM [ApacheHive].[PUBLIC].[EMPLOYEES] LIMIT 5"
})
print("クエリ結果:", query_result)
return catalogs, query_result
---------------- メイン ----------------
async def main():
# ステップ1: MCP
catalogs, query_result = await query_mcp()
# ステップ2: Salesforceトークンとエージェント
access_token, _ = get_salesforce_token()
session_id = start_agent_session(access_token)
# ステップ3: MCP結果を指示と共にエージェントに送信
context_message = (
f"あなたは親切なアシスタントです。\n"
f"以下のMCPデータが取得されました:\n\n"
f"カタログ: {catalogs}\n"
f"クエリ結果: {query_result}\n\n"
f"このデータに基づいてユーザーの質問に答えてください: "
f"利用可能なすべてのカタログをリストし、クエリ結果を要約してください。"
)
response = post_message_to_agent(access_token, session_id, context_message)
print("\n エージェントの応答:")
print(response)
if name == "main":
asyncio.run(main())
サンプル出力
Salesforce Agentforce でデータ駆動型エージェントを構築
Agentforce を使用すると、AI エージェントをオーケストレーションしてCRM ワークフローを自動化できます。CData Connect AI とリモートMCP Server によるリアルタイムデータ接続を活用すれば、エージェントはCRM、ERP、データベース、財務プラットフォームなど、スタック全体のリアルタイムでコンテキストに沿ったデータを活用して動作できます。
Connect AI は14日間の無償トライアルがご利用いただけます。今すぐサインアップ、Agentforce を300以上の外部システムに接続してみましょう！