 |
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 Workday 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 Workday connectivity in Connect AI, register the MCP server in Agentforce, and build a workflow that queries Workday data data.
CData provides the easiest way to access and integrate live data from Workday. Customers use CData connectivity to:
Users frequently integrate Workday with analytics tools such as Tableau, Power BI, and Excel, and leverage our tools to replicate Workday data to databases or data warehouses. Access is secured at the user level, based on the authenticated user's identity and role.
For more information on configuring Workday to work with CData, refer to our Knowledge Base articles: Comprehensive Workday Connectivity through Workday WQL and Reports-as-a-Service & Workday + CData: Connection & Integration Best Practices.
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 Workday from Agentforce is made possible through CData Connect AI Remote MCP. To interact with Workday data from Agentforce, we start by creating and configuring a Workday connection in CData Connect AI.
To connect to Workday, users need to find the Tenant and BaseURL and then select their API type.
To obtain the BaseURL and Tenant properties, log into Workday and search for "View API Clients." On this screen, you'll find the Workday REST API Endpoint, a URL that includes both the BaseURL and Tenant.
The format of the REST API Endpoint is: https://domain.com/subdirectories/mycompany, where:
The value you use for the ConnectionType property determines which Workday API you use. See our Community Article for more information on Workday connectivity options and best practices.
| API | ConnectionType Value |
|---|---|
| WQL | WQL |
| Reports as a Service | Reports |
| REST | REST |
| SOAP | SOAP |
Your method of authentication depends on which API you are using.
See the Help documentation for more information on configuring OAuth with Workday.
👁 Configuring a connection (Salesforce is shown)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 Workday 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 [Workday].[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