 |
VOOZH | about |
|
VOOZH | about |
Agentforce is built for enterprise teams. Its declarative tooling and API-first architecture make it straightforward to compose assistants and automate CRM-centric workflows. But when agents need to work with data beyond Presto data, many implementations fall back on periodic syncs or bespoke middleware to replicate external systems into local stores. That adds complexity, creates governance and maintenance overhead, introduces sync delays, and limits the real-time potential of your agents.
CData Connect AI fills this gap with direct, live connectivity to hundreds of enterprise apps, databases, ERPs, and finance platforms. Using CData's remote Model Context Protocol (MCP) Server, Agentforce agents can securely read, write, and act on fresh, contextual data at runtime; no replication required. The result is grounded responses, real-time decisions, and cross-system automations with fewer moving parts and stronger control.
This article outlines the steps required to configure Presto connectivity in Connect AI, register the MCP server in Agentforce, and build a workflow that queries Presto data data.
Accessing and integrating live data from Trino and Presto SQL engines has never been easier with CData. Customers rely on CData connectivity to:
Presto and Trino allow users to access a variety of underlying data sources through a single endpoint. When paired with CData connectivity, users get pure, SQL-92 access to their instances, allowing them to integrate business data with a data warehouse or easily access live data directly from their preferred tools, like Power BI and Tableau.
In many cases, CData's live connectivity surpasses the native import functionality available in tools. One customer was unable to effectively use Power BI due to the size of the datasets needed for reporting. When the company implemented the CData Power BI Connector for Presto they were able to generate reports in real-time using the DirectQuery connection mode.
Before you begin, you'll need a few credentials mentioned below:
Note: You can create the base64 encoded version of MCP_AUTH using any Base64 encoding tool.
Connectivity to Presto from Agentforce is made possible through CData Connect AI Remote MCP. To interact with Presto data from Agentforce, we start by creating and configuring a Presto connection in CData Connect AI.
Set the Server and Port connection properties to connect, in addition to any authentication properties that may be required.
To enable TLS/SSL, set UseSSL to true.
In order to authenticate with LDAP, set the following connection properties:
In order to authenticate with KERBEROS, set the following connection properties:
A Personal Access Token (PAT) is used to authenticate the connection to Connect AI from Agentforce. It is best practice to create a separate PAT for each service to maintain granularity of access.
With the connection configured and a PAT generated, we are ready to connect to Presto data from Agentforce.
Install packages:
pip install requests asyncio langchain-mcp-adapters
Create config.py (replace placeholders with your actual values). It should define:
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"
The Python script has three main sections:
Open a terminal in the project directory where the script is and run:
python 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():
"""Fetch a fresh Salesforce OAuth token using client credentials."""
print(" Requesting fresh Salesforce token...")
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(" Got access token.")
return token_data["access_token"], token_data["instance_url"]
def start_agent_session(access_token):
"""Start a session with 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" Agent session started: {session_data['sessionId']}")
return session_data["sessionId"]
def post_message_to_agent(access_token, session_id, message_text):
"""Send a message to the Salesforce Einstein Agent session."""
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 part ---------------- #
async def query_mcp():
"""Query MCP server for catalogs and a sample query."""
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("Discovered MCP tools:", [t.name for t in tools])
# List catalogs
get_catalogs_tool = next(t for t in tools if t.name == "getCatalogs")
catalogs = await get_catalogs_tool.ainvoke({})
print("Catalogs:", catalogs)
# Run a sample query
query_tool = next(t for t in tools if t.name == "queryData")
query_result = await query_tool.ainvoke({
"query": "SELECT * FROM [Presto].[PUBLIC].[EMPLOYEES] LIMIT 5"
})
print("Query result:", query_result)
return catalogs, query_result
# ---------------- Main ---------------- #
async def main():
# Step 1: MCP
catalogs, query_result = await query_mcp()
# Step 2: Salesforce token and agent
access_token, _ = get_salesforce_token()
session_id = start_agent_session(access_token)
# Step 3: Post MCP results to agent with instructions
context_message = (
f"You are a helpful assistant.\n"
f"The following MCP data was retrieved:\n\n"
f"Catalogs: {catalogs}\n"
f"Query result: {query_result}\n\n"
f"Please answer the user's question based on this data: "
f"List all available catalogs and summarize the query results."
)
response = post_message_to_agent(access_token, session_id, context_message)
print("\n Agent response:")
print(response)
if __name__ == "__main__":
asyncio.run(main())
Agentforce lets you orchestrate AI agents and automate CRM workflows. With live connectivity through CData Connect AI and the remote MCP Server, your agents can operate on real-time, contextual data across your stack CRMs, ERPs, databases, finance platforms, and more.
Start your free trial today to see how CData can empower Agentforce with live, secure access to hundreds of external systems.
Learn more about CData Connect AI or sign up for free trial access:
Free Trial