【徹底解説】Agentforce とPaylocity のデータをMCP Server で簡単連携

加藤龍彦
加藤龍彦
デジタルマーケティング
CData Connect AI のリモートMCP Serverを活用して、Agentforce がPaylocity のデータを安全に読み取り、アクションを実行できるようにする方法を解説します。

Agentforce は、エンタープライズチーム向けに設計されたAI プラットフォームです。宣言的なツールとAPI ファーストなアーキテクチャで、アシスタントの構成やCRM ベースのワークフロー自動化を簡単に実現できます。

しかし、エージェントがPaylocity 以外のデータを扱う必要がある場合、多くの実装では定期的な同期や独自のミドルウェアを使用して外部システムをローカルデータベースに複製しています。この方法では、複雑性の増加、ガバナンスとメンテナンスのオーバーヘッド、同期の遅延が発生し、エージェントのリアルタイム性能が制限されてしまいます。

CData Connect AIは、300以上のビジネスアプリケーション、データベース、ERP、会計ツールへのリアルタイムアクセスを提供することで、このギャップを埋めます。CData のリモートMCP Serverを使用することで、Agentforce エージェントは実行時に最新のコンテキストに沿ったデータをセキュアに読み書きし、アクションを実行できます。そのため、より少ない構成要素で、より強力な制御のもと、根拠に基づいた応答、リアルタイムの意思決定、システム間の自動化を実現できます。

本記事では、Connect AI でPaylocity への接続を設定し、MCP Server をAgentforce に登録し、Paylocity データをクエリするワークフローを構築するために必要な手順を説明します。

前提条件:

  • CData Connect AI アカウント
  • Python 3.9 以上のインストール
  • Agentforce が有効化されたPaylocity アカウントと、接続アプリの作成/管理権限

必要な認証情報のチェックリスト

開始する前に、以下の認証情報を準備してください。

CData Connect AI MCP

  1. 「USERNAME」: アカウントのメールアドレス
  2. 「PAT:」 Connect AIの「Settings」から「アクセストークン」をクリック（一度だけコピー可能）
  3. 「MCP_BASE_URL: 」https://mcp.cloud.cdata.com/mcp
  4. 「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 連携用のPaylocity への接続を設定

それでは、Agentforce からCData Connect AI のリモートMCP 経由でPaylocity への接続を設定していきましょう。Agentforce からPaylocity を操作するには、まずCData Connect AI でPaylocity へのコネクションを作成し、設定します。

  1. Connect AI にログインして「Sources」をクリック。次に「Add Connection」をクリックします。
  2. 一覧から「Paylocity」を選択します。
  3. Paylocity に接続するために必要な認証プロパティを入力します。

    Paylocity への接続を確立するには以下を設定します。

    • RSAPublicKey：Paylocity アカウントでRSA 暗号化が有効になっている場合は、Paylocity に関連付けられたRSA キーを設定。

      このプロパティは、Insert およびUpdate ステートメントを実行するために必須です。この機能が無効になっている場合は必須ではありません。

    • UseSandbox：サンドボックスアカウントを使用する場合はTrue に設定。
    • CustomFieldsCategory：Customfields カテゴリに設定。これは、IncludeCustomFields がtrue に設定されている場合は必須です。デフォルト値はPayrollAndHR です。
    • Key：Paylocity の公開鍵で暗号化されたAES 共通鍵（base 64 エンコード）。これはコンテンツを暗号化するためのキーです。

      Paylocity は、RSA 復号化を使用してAES 鍵を復号化します。
      これはオプションのプロパティで、IV の値が指定されていない場合、ドライバーは内部でキーを生成します。

    • IV：コンテンツを暗号化するときに使用するAES IV（base 64 エンコード）。これはオプションのプロパティで、Key の値が指定されていない場合、ドライバーは内部でIV を生成します。

    OAuth

    OAuth を使用してPaylocity で認証する必要があります。OAuth では認証するユーザーにブラウザでPaylocity との通信を要求します。詳しくは、ヘルプドキュメントのOAuth セクションを参照してください。

    Pay Entry API

    Pay Entry API はPaylocity API の他の部分と完全に分離されています。個別のクライアントID とシークレットを使用し、アカウントへのアクセスを許可するにはPaylocity から明示的にリクエストする必要があります。 Pay Entry API を使用すると、個々の従業員の給与情報を自動的に送信できます。 Pay Entry API によって提供されるものの性質が非常に限られているため、CData では個別のスキーマを提供しないことを選択しましたが、UsePayEntryAPI 接続プロパティを介して有効にできます。

    UsePayEntryAPI をtrue に設定する場合は、CreatePayEntryImportBatch、MergePayEntryImportBatch、Input_TimeEntry、およびOAuth ストアドプロシージャのみ利用できることに注意してください。 製品のその他の機能を使用しようとするとエラーが発生します。また、OAuthAccessToken を個別に保存する必要があります。これは、この接続プロパティを使用するときに異なるOAuthSettingsLocation を設定することを意味します。

  4. 「 Save & Test」をクリック。
  5. Paylocity ページの「Permissions」タブに移動し、ユーザーごとの権限を更新します。

パーソナルアクセストークンの追加

パーソナルアクセストークン（PAT）は、Agentforce からConnect AI への接続を認証するために使用されます。アクセスの粒度を維持するため、サービスごとに個別のPAT を作成するのがベストプラクティスです。

  1. Connect AI アプリの右上にある歯車アイコン（）をクリックして、設定ページを開きます。
  2. 設定ページで、アクセストークンセクションに移動し、PATを作成をクリックします。
  3. PATに名前を付けて、作成をクリックします。
  4. パーソナルアクセストークンは作成時のみ表示されるため、必ずコピーして安全に保管してください。

接続が設定され、PAT が生成されたので、Agentforce からSalesforce に接続する準備が整いました。

ステップ2: Agentforce を有効化し、Experience Cloud サイトを公開

  1. Salesforce の設定ページで「Einstein Setup」に移動し、「Turn on Einstein」トグルを有効化します。
  2. Agentforce Agents に移動し、「Agentforce」をオンにします。
  3. 次に、Coral Cloud をホストする「Experience Cloud サイト」を公開します。 設定で「すべてのサイト」を検索し、「Builder（for coral-cloud）」を選択して「Publish」をクリックし、「Confirm」をクリックします。
  4. Experience Builder メニューをクリック
  5. Salesforce Settings をクリック
  6. ブラウザを更新して設定を再読み込みします。Experience Site のブラウザタブは閉じてもOKです。

ステップ3: EinsteinServiceAgentユーザーとプロファイルを設定する

  1. 設定のクイック検索で「Users」を検索して選択
  2. EinsteinServiceAgent ユーザーの横にある編集をクリック
  3. プロファイルを「Einstein Agent User」に更新
  4. その他の設定はそのまま
  5. 「保存」をクリック

ステップ4: エージェントを作成し、外部接続をアタッチ

  1. 設定で「Agentforce Agents」を検索し、「新しいエージェント」をクリックします。
  2. 「Create with Gen AI」をクリック
  3. 概要、名前、役割、会社情報を入力します。Agent User = EinsteinServiceAgent Userと設定しましょう。
  4. 「Keep a record of conversations with enhanced event logs」にチェックを入れ、「Save」をクリックします。
  5. 「Create」をクリック
  6. 「Connections」を追加し、「Needs Setup」を選択して「External App」をクリック
  7. Connection Type = API を設定し、「Integratoin Name」に任意の名前を入力します。
  8. 「新しい接続アプリ」をクリック（次のステップで設定します）

ステップ5: Salesforce接続アプリを作成・設定する（クライアント認証情報）

  1. 接続アプリの作成を許可します。
  2. 「New Connected App」をクリックし、「Name」と「Admin Email」を入力
  3. 「Enable OAuth Settings」をオンにしてCallback URL: https://login.salesforce.com に設定しましょう。
  4. 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)
  5. 以下の選択を解除してください。
    • PKCE
    • Require Secret for Web Server Flow
    • Require Secret for Refresh Token Flow
  6. 以下を選択してください。
    • Enable Client Credentials Flow
    • Issue JWT-based access tokens for named users.
    • Enable the Save option
  7. アプリの「Manage」ページで「Edit Policies」をクリック
  8. 「Permitted Users」は、自社のポリシーに合わせて設定してください。
    • 「Client Credentials Flow」でRun As: choose a user with at least API Only access を選択します。
    • 「JWT-based Access Tokens」を有効化し、Token Timeout: 30 minutesを設定して「保存」をクリックします。
  9. 最後に、「Agent Connection」に戻り、今作成したアプリを選択して「Save」をクリックします。
  10. 「App Manager」から「Consumer Key」と「Consumer Secret」を取得
  11. 作成したアプリを見つけて、「Manage Consumer Details」を表示します。
  12. 「Consumer Key」と「Consumer Secret」をコピー

ステップ6：連携に必要な残りのID を入手

  1. 「My Domain」: 設定でMy Domain に移動し、トークンリクエスト用の...my.salesforce.com URLをコピーします。
  2. 「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 経由でPaylocity とAgentforce を連携してAgentforce がPaylocity のデータを取得するPython スクリプトを用意しました。Python スクリプトは3つの内容で構成されています。

  1. Salesforce / Einstein Agent関数
    • get_salesforce_token() - OAuth2トークンを取得
    • start_agent_session() - エージェントセッションを開始
    • post_message_to_agent() - エージェントにメッセージを送信
  2. MCP関数
    • query_mcp() - カタログを取得し、MCPツールからサンプルクエリを実行
  3. 全体のオーケストレーション
    • 「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 [Paylocity].[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以上の外部システムに接続してみましょう！

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

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

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

この記事の内容

関連記事