# Directory Structure
```
├── .env
├── .env.example
├── .gitignore
├── agentforce_client.py
├── agentforce_mcp_server.py
├── github_push.sh
├── README.md
├── requirements.txt
├── session_manager.py
├── setup.sh
└── test_agentforce.py
```
# Files
--------------------------------------------------------------------------------
/.env:
--------------------------------------------------------------------------------
```
SALESFORCE_ORG_ID="PLACEHOLDER_FOR_ORG_ID"
SALESFORCE_AGENT_ID="PLACEHOLDER_FOR_AGENT_ID"
SALESFORCE_CLIENT_ID="PLACEHOLDER_FOR_CLIENT_ID"
SALESFORCE_CLIENT_SECRET="PLACEHOLDER_FOR_CLIENT_SECRET"
SALESFORCE_SERVER_URL="PLACEHOLDER_FOR_SERVER_URL"
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# Salesforce Agentforce API configuration
SALESFORCE_ORG_ID="your_org_id_here"
SALESFORCE_AGENT_ID="your_agent_id_here"
SALESFORCE_CLIENT_ID="your_client_id_here"
SALESFORCE_CLIENT_SECRET="your_client_secret_here"
SALESFORCE_SERVER_URL="your_server_url_here"
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Environment variables
.env
# Virtual environment
venv/
# Python cache files
__pycache__/
*.py[cod]
*$py.class
*.so
.Python
.pytest_cache/
# Distribution / packaging
dist/
build/
*.egg-info/
# Logs
*.log
# OS generated files
.DS_Store
.DS_Store?
._*
.Spotlight-V100
.Trashes
ehthumbs.db
Thumbs.db
# IDE specific files
.idea/
.vscode/
*.swp
*.swo
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# Agentforce MCP Server
This MCP server provides tools to interact with the Salesforce Agentforce API. It allows authentication, session creation, and message exchange with Agentforce agents.
## Getting Started After Cloning
If you've just cloned this repository, you can use the interactive setup script to quickly configure and run the server:
```bash
chmod +x setup.sh
./setup.sh
```
The setup script will:
1. Check your Python version
2. Install required dependencies
3. Guide you through entering your Salesforce credentials
4. Test your connection to Salesforce
5. Offer to start the server
6. Provide instructions for configuring Claude Desktop
Alternatively, you can follow these manual steps:
1. **Install dependencies**:
```bash
pip install -r requirements.txt
```
2. **Set up your environment variables**:
```bash
cp .env.example .env
```
3. **Collect your Salesforce credentials**:
- **SALESFORCE_ORG_ID**: Your 18-character Salesforce Org ID
- **SALESFORCE_AGENT_ID**: The 18-character Agent ID from your Agentforce agent
- **SALESFORCE_CLIENT_ID**: The Consumer Key from your Connected App
- **SALESFORCE_CLIENT_SECRET**: The Consumer Secret from your Connected App
- **SALESFORCE_SERVER_URL**: Your Salesforce My Domain URL without https:// prefix
4. **Edit your .env file** with the collected credentials:
```
SALESFORCE_ORG_ID="00D5f000000J2PKEA0"
SALESFORCE_AGENT_ID="0XxHn000000x9F1KAI"
SALESFORCE_CLIENT_ID="3MVG9OGq41FnYVsFgnaG0AzJDWnoy37Bb18e0R.GgDJu2qB9sqppVl7ehWmJhGvPSLrrA0cBNhDJdsbZXnv52"
SALESFORCE_CLIENT_SECRET="210117AC36E9E4C8AFCA02FF062B8A677BACBFFB71D2BB1162D60D316382FADE"
SALESFORCE_SERVER_URL="example.my.salesforce.com"
```
(Note: These are fictional example values. Replace with your actual credentials.)
5. **Make the server script executable**:
```bash
chmod +x agentforce_mcp_server.py
```
6. **Run the server**:
```bash
python agentforce_mcp_server.py
```
For detailed instructions on finding your Salesforce credentials, see the [Setting Up Salesforce](#setting-up-salesforce) section below.
## Setup
1. Ensure you have Python 3.10 or higher installed.
2. Install the required dependencies:
```bash
pip install -r requirements.txt
```
3. Make the server script executable:
```bash
chmod +x agentforce_mcp_server.py
```
## Configuration
The server uses environment variables for configuration. These are loaded from the `.env` file.
1. Copy the example environment file to create your own:
```bash
cp .env.example .env
```
2. Edit the `.env` file and fill in your values:
```
SALESFORCE_ORG_ID="your_org_id_here"
SALESFORCE_AGENT_ID="your_agent_id_here" # The 18-character Agent ID you found in Salesforce
SALESFORCE_CLIENT_ID="your_client_id_here" # The Consumer Key from your Connected App
SALESFORCE_CLIENT_SECRET="your_client_secret_here" # The Consumer Secret from your Connected App
SALESFORCE_SERVER_URL="your_server_url_here" # Your My Domain URL (e.g., example.my.salesforce.com)
```
## Salesforce Configuration
To use the Agentforce API, you need to:
1. Create a Connected App in your Salesforce org
2. Find your Agentforce Agent ID
3. Note your Salesforce My Domain URL
For detailed instructions on these steps, see the [Setting Up Salesforce](#setting-up-salesforce) section below.
## Running the Server
Run the server using:
```bash
python agentforce_mcp_server.py
```
## Available Tools
The MCP server exposes the following tools:
### 1. `authenticate`
Authenticates with the Agentforce API using a client email.
Parameters:
- `client_email`: Email of the client for authentication
### 2. `create_agent_session`
Creates a session with the configured Agentforce agent.
Parameters:
- `client_email`: Email of the authenticated client
### 3. `send_message_to_agent`
Sends a message to the Agentforce agent and returns the response.
Parameters:
- `client_email`: Email of the authenticated client
- `message`: Message to send to the agent
### 4. `get_session_status`
Gets the status of the current session, including authentication status, session ID, and sequence ID.
Parameters:
- `client_email`: Email of the authenticated client
### 5. `complete_agentforce_conversation`
Convenience method that handles the complete flow - authentication, session creation, and message sending.
Parameters:
- `client_email`: Email of the client for authentication
- `user_query`: Message to send to the agent
## Using with Claude for Desktop
To use this server with Claude for Desktop, update your `claude_desktop_config.json` file:
```json
{
"mcpServers": {
"agentforce": {
"command": "python",
"args": [
"/path/to/your/agentforce_mcp_server.py"
]
}
}
}
```
Replace the path with the absolute path to the server script on your machine.
### Path Locations by Platform
#### macOS
- Configuration file: `~/Library/Application Support/Claude/claude_desktop_config.json`
- Example path: `/Users/yourusername/Projects/agentforce-mcp-server/agentforce_mcp_server.py`
#### Windows
- Configuration file: `%APPDATA%\Claude\claude_desktop_config.json`
- Example path: `C:\Users\yourusername\Projects\agentforce-mcp-server\agentforce_mcp_server.py`
## Setting Up Salesforce
### Creating a Connected App
To use the Agentforce API, you need to create a Connected App in your Salesforce org:
1. Log in to your Salesforce org as an administrator
2. Go to **Setup**
3. In the Quick Find box, search for "App Manager" and click on it
4. Click the **New Connected App** button
5. Fill in the basic information:
- **Connected App Name**: Agentforce MCP Integration (or any name you prefer)
- **API Name**: Agentforce_MCP_Integration (this will be auto-filled)
- **Contact Email**: Your email address
6. Check **Enable OAuth Settings**
7. Set the **Callback URL** to `https://localhost/oauth/callback` (this is not used but required)
8. Under **Selected OAuth Scopes**, add:
- Manage user data via APIs (api)
- Perform requests at any time (refresh_token, offline_access)
9. Click **Save**
10. After saving, you'll be redirected to the Connected App detail page
11. Note the **Consumer Key** (this is your client ID) and click **Click to reveal** next to **Consumer Secret** to get your client secret
### Finding Your Agent ID
To find your Agentforce Agent ID:
1. Log in to your Salesforce org
2. Navigate to **Einstein Agent Builder**
3. Select the agent you want to use
4. Look at the URL in your browser - it will contain the Agent ID in the format: `https://your-salesforce-instance.lightning.force.com/lightning/r/Agent__c/0XxXXXXXXXXXXXXX/view`
5. The Agent ID is that 18-character ID (`0XxXXXXXXXXXXXXX`) in the URL
### Finding Your Salesforce My Domain URL
To find your Salesforce My Domain URL:
1. Log in to your Salesforce org
2. Go to **Setup**
3. In the Quick Find box, search for "My Domain" and click on it
4. You'll see your domain in the format `DOMAIN-NAME.my.salesforce.com`
5. Use this URL without the "https://" prefix in your .env file
### Finding Your Org ID
To find your Salesforce Org ID:
1. Log in to your Salesforce org
2. Go to **Setup**
3. In the Quick Find box, search for "Company Information" and click on it
4. Look for the "Organization ID" field - this is your Salesforce Org ID
5. It will be a 15 or 18-character alphanumeric string
## Notes
- The server automatically manages sequence IDs for message exchanges
- Authentication and session state are maintained for each client email
- All API interactions are logged for debugging purposes
## Troubleshooting
If you encounter issues:
1. **Authentication failures**: Verify your Connected App settings and ensure the client ID and secret are correct
2. **Session creation errors**: Check your Agent ID and make sure it's the 18-character version
3. **Connection issues**: Verify your Salesforce My Domain URL is correct (without "https://" prefix)
4. **Permission errors**: Make sure your Connected App has the proper OAuth scopes enabled
## Testing the Setup
You can test your setup using the included test script:
```bash
python test_agentforce.py
```
This will attempt to authenticate, create a session, and exchange messages with your Agentforce agent.
## Contributing and GitHub Push
This repository includes a helpful script that simplifies the process of pushing your changes to GitHub:
```bash
chmod +x github_push.sh
./github_push.sh
```
The `github_push.sh` script will:
1. Check if git is installed on your system
2. Verify that sensitive files like `.env` won't be pushed (they're in `.gitignore`)
3. Prompt you for your GitHub repository URL
4. Initialize a git repository if needed, or update the remote URL
5. Add all files and display them for your review
6. Commit the changes with a descriptive message
7. Push the changes to GitHub
This makes it easy to share your customizations or contribute back to the project while ensuring that sensitive information stays secure.
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
httpx>=0.24.0
python-dotenv>=1.0.0
mcp>=1.2.0
```
--------------------------------------------------------------------------------
/test_agentforce.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
import httpx
import asyncio
import os
from dotenv import load_dotenv
from agentforce_client import AgentforceClient
async def test_agentforce_api():
"""Test the Agentforce API directly"""
# Load environment variables
load_dotenv()
# Create client
client = AgentforceClient(
server_url=os.getenv("SALESFORCE_SERVER_URL"),
client_id=os.getenv("SALESFORCE_CLIENT_ID"),
client_secret=os.getenv("SALESFORCE_CLIENT_SECRET"),
agent_id=os.getenv("SALESFORCE_AGENT_ID")
)
# Test email to use
test_email = "[email protected]"
print(f"Testing Agentforce API with {test_email}")
# Step 1: Get access token
print("\n1. Getting access token...")
token_response = await client.get_access_token(test_email)
if not token_response or not token_response.get('access_token'):
print("❌ Failed to get access token")
return
print(f"✅ Got access token: {token_response['access_token'][:10]}...")
# Step 2: Create session
print("\n2. Creating session...")
session_id = await client.create_session(
token=token_response['access_token'],
instance_url=token_response['instance_url']
)
if not session_id:
print("❌ Failed to create session")
return
print(f"✅ Created session with ID: {session_id}")
# Step 3: Send message
print("\n3. Sending test message...")
message_response = await client.send_message(
session_id=session_id,
token=token_response['access_token'],
message="Hello, can you help me find a hotel near CDG airport?",
sequence_id=1
)
print(f"✅ Got response: {message_response['agent_response'][:100]}...")
print(f"Next sequence ID: {message_response['sequence_id']}")
# Step 4: Send follow-up message
print("\n4. Sending follow-up message...")
followup_response = await client.send_message(
session_id=session_id,
token=token_response['access_token'],
message="I need a room for 2 people for tonight",
sequence_id=message_response['sequence_id']
)
print(f"✅ Got response: {followup_response['agent_response'][:100]}...")
print("\nTest completed successfully!")
if __name__ == "__main__":
asyncio.run(test_agentforce_api())
```
--------------------------------------------------------------------------------
/github_push.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Colors for terminal output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
NC='\033[0m' # No Color
echo -e "${BLUE}=====================================================${NC}"
echo -e "${BLUE} Agentforce MCP Server - GitHub Push Script ${NC}"
echo -e "${BLUE}=====================================================${NC}"
# Check if git is installed
if ! command -v git &> /dev/null; then
echo -e "${RED}Git is not installed. Please install git and try again.${NC}"
exit 1
fi
# Get the current directory
PROJECT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
cd "$PROJECT_DIR"
# Check if .env file exists and warn the user
if [ -f ".env" ]; then
echo -e "${YELLOW}WARNING: .env file detected. This file contains sensitive information and should NOT be pushed to GitHub.${NC}"
echo -e "${YELLOW}The .env file has been added to .gitignore, but please verify it won't be included in your commit.${NC}"
echo ""
fi
# Ask for GitHub repository URL
echo -e "${BLUE}Please enter your GitHub repository URL (e.g., https://github.com/username/repo.git):${NC}"
read -r REPO_URL
if [ -z "$REPO_URL" ]; then
echo -e "${RED}No repository URL provided. Exiting.${NC}"
exit 1
fi
# Check if the directory is already a git repository
if [ ! -d ".git" ]; then
echo -e "${YELLOW}Initializing Git repository...${NC}"
git init
echo -e "${GREEN}Git repository initialized.${NC}"
else
echo -e "${GREEN}Git repository already exists.${NC}"
fi
# Check if remote already exists
if git remote | grep -q "origin"; then
echo -e "${YELLOW}Remote 'origin' already exists. Updating to new URL...${NC}"
git remote set-url origin "$REPO_URL"
else
echo -e "${YELLOW}Adding remote 'origin'...${NC}"
git remote add origin "$REPO_URL"
fi
echo -e "${GREEN}Remote 'origin' set to: $REPO_URL${NC}"
# Add all files to git
echo -e "${YELLOW}Adding files to git...${NC}"
git add .
# Show status
echo -e "${YELLOW}Git status:${NC}"
git status
# Confirm with user before committing
echo -e "${BLUE}Review the files above. Are you sure you want to commit these files? (y/n)${NC}"
read -r CONFIRM
if [ "$CONFIRM" != "y" ] && [ "$CONFIRM" != "Y" ]; then
echo -e "${RED}Commit aborted by user.${NC}"
exit 1
fi
# Commit
echo -e "${YELLOW}Committing changes...${NC}"
git commit -m "Initial commit of Agentforce MCP Server"
# Push to GitHub
echo -e "${YELLOW}Pushing to GitHub...${NC}"
git push -u origin master || git push -u origin main
# Final message
echo -e "${GREEN}Push completed! Your code should now be on GitHub.${NC}"
echo -e "${BLUE}=====================================================${NC}"
echo -e "${YELLOW}Next steps:${NC}"
echo -e "1. Ensure your .env file is not published (it should be in .gitignore)"
echo -e "2. Share the .env.example file with users so they know what environment variables to set"
echo -e "3. Update the README.md if needed with specific instructions for your repository"
echo -e "${BLUE}=====================================================${NC}"
```
--------------------------------------------------------------------------------
/session_manager.py:
--------------------------------------------------------------------------------
```python
from typing import Dict, Any, Optional
import logging
logger = logging.getLogger(__name__)
class SessionManager:
"""Manages Agentforce sessions and sequence IDs"""
def __init__(self):
self.sessions: Dict[str, Dict[str, Any]] = {}
def store_auth_info(self, client_email: str, token_info: Dict[str, str]) -> None:
"""Store authentication information for a client"""
self.sessions[client_email] = {
'access_token': token_info.get('access_token'),
'instance_url': token_info.get('instance_url'),
'session_id': None,
'last_sequence_id': 0
}
def store_session_id(self, client_email: str, session_id: str) -> None:
"""Store session ID for a client"""
if client_email in self.sessions:
self.sessions[client_email]['session_id'] = session_id
self.sessions[client_email]['last_sequence_id'] = 0
else:
logger.error(f"❌ Client email {client_email} not found in sessions")
def update_sequence_id(self, client_email: str, sequence_id: int) -> None:
"""Update sequence ID for a client"""
if client_email in self.sessions:
self.sessions[client_email]['last_sequence_id'] = sequence_id
else:
logger.error(f"❌ Client email {client_email} not found in sessions")
def get_next_sequence_id(self, client_email: str) -> int:
"""Get next sequence ID for a client"""
if client_email in self.sessions:
return self.sessions[client_email]['last_sequence_id'] + 1
else:
logger.error(f"❌ Client email {client_email} not found in sessions")
return 1 # Default to 1 if not found
def get_access_token(self, client_email: str) -> Optional[str]:
"""Get access token for a client"""
if client_email in self.sessions:
return self.sessions[client_email].get('access_token')
else:
logger.error(f"❌ Client email {client_email} not found in sessions")
return None
def get_instance_url(self, client_email: str) -> Optional[str]:
"""Get instance URL for a client"""
if client_email in self.sessions:
return self.sessions[client_email].get('instance_url')
else:
logger.error(f"❌ Client email {client_email} not found in sessions")
return None
def get_session_id(self, client_email: str) -> Optional[str]:
"""Get session ID for a client"""
if client_email in self.sessions:
return self.sessions[client_email].get('session_id')
else:
logger.error(f"❌ Client email {client_email} not found in sessions")
return None
def get_session_status(self, client_email: str) -> str:
"""Get session status for a client"""
if client_email not in self.sessions:
return "No active session. You need to authenticate first."
session_info = self.sessions[client_email]
status = f"Client Email: {client_email}\n"
status += f"Authenticated: {'Yes' if session_info.get('access_token') else 'No'}\n"
status += f"Session ID: {session_info.get('session_id') or 'Not created'}\n"
status += f"Last Sequence ID: {session_info.get('last_sequence_id', 0)}\n"
return status
def is_authenticated(self, client_email: str) -> bool:
"""Check if a client is authenticated"""
return client_email in self.sessions and self.sessions[client_email].get('access_token') is not None
def has_session(self, client_email: str) -> bool:
"""Check if a client has a session"""
return client_email in self.sessions and self.sessions[client_email].get('session_id') is not None
```
--------------------------------------------------------------------------------
/agentforce_mcp_server.py:
--------------------------------------------------------------------------------
```python
#!/usr/bin/env python3
from mcp.server.fastmcp import FastMCP
from typing import Dict, Optional, List, Any
import os
import logging
from dotenv import load_dotenv
from agentforce_client import AgentforceClient
from session_manager import SessionManager
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
# Load environment variables from .env file
load_dotenv()
# Initialize FastMCP server
mcp = FastMCP("agentforce-mcp-server")
# Initialize the Agentforce client
agentforce_client = AgentforceClient(
server_url=os.getenv("SALESFORCE_SERVER_URL"),
client_id=os.getenv("SALESFORCE_CLIENT_ID"),
client_secret=os.getenv("SALESFORCE_CLIENT_SECRET"),
agent_id=os.getenv("SALESFORCE_AGENT_ID")
)
# Initialize the session manager
session_manager = SessionManager()
@mcp.tool()
async def authenticate(client_email: str) -> str:
"""Authenticate with the Agentforce API and store the token.
Args:
client_email: Email of the client for authentication
"""
token_response = await agentforce_client.get_access_token(client_email)
if not token_response or not token_response.get('access_token'):
return "Failed to authenticate. Please check the client email and try again."
# Store token information
session_manager.store_auth_info(client_email, token_response)
return f"Successfully authenticated as {client_email}"
@mcp.tool()
async def create_agent_session(client_email: str) -> str:
"""Create a session with an Agentforce agent.
Args:
client_email: Email of the authenticated client
"""
if not session_manager.is_authenticated(client_email):
return "You need to authenticate first using the authenticate tool."
token = session_manager.get_access_token(client_email)
instance_url = session_manager.get_instance_url(client_email)
session_id = await agentforce_client.create_session(token, instance_url)
if not session_id:
return "Failed to create session. Please try again."
# Store session ID
session_manager.store_session_id(client_email, session_id)
return f"Successfully created session with agent. Session ID: {session_id}"
@mcp.tool()
async def send_message_to_agent(client_email: str, message: str) -> str:
"""Send a message to the Agentforce agent and get the response.
Args:
client_email: Email of the authenticated client
message: Message to send to the agent
"""
if not session_manager.is_authenticated(client_email):
return "You need to authenticate first using the authenticate tool."
if not session_manager.has_session(client_email):
return "You need to create a session first using the create_agent_session tool."
token = session_manager.get_access_token(client_email)
session_id = session_manager.get_session_id(client_email)
next_sequence_id = session_manager.get_next_sequence_id(client_email)
response = await agentforce_client.send_message(
session_id=session_id,
token=token,
message=message,
sequence_id=next_sequence_id
)
# Update last sequence ID
session_manager.update_sequence_id(client_email, response['sequence_id'])
return response['agent_response']
@mcp.tool()
async def get_session_status(client_email: str) -> str:
"""Get the status of the current session.
Args:
client_email: Email of the authenticated client
"""
return session_manager.get_session_status(client_email)
@mcp.tool()
async def complete_agentforce_conversation(client_email: str, user_query: str) -> str:
"""Complete full conversation flow with Agentforce - authenticate, create session, and send message.
Args:
client_email: Email of the client for authentication
user_query: Message to send to the agent
"""
# Step 1: Authenticate
if not session_manager.is_authenticated(client_email):
auth_result = await authenticate(client_email)
if not "Successfully" in auth_result:
return f"Authentication failed: {auth_result}"
logger.info(f"Authentication successful for {client_email}")
# Step 2: Create session
if not session_manager.has_session(client_email):
session_result = await create_agent_session(client_email)
if not "Successfully" in session_result:
return f"Session creation failed: {session_result}"
logger.info("Session created successfully")
# Step 3: Send message
response = await send_message_to_agent(client_email, user_query)
return response
if __name__ == "__main__":
logger.info("Starting Agentforce MCP Server...")
logger.info(f"Using Agent ID: {os.getenv('SALESFORCE_AGENT_ID')}")
# Initialize and run the server
mcp.run(transport='stdio')
```
--------------------------------------------------------------------------------
/agentforce_client.py:
--------------------------------------------------------------------------------
```python
import httpx
import json
import logging
import uuid
from typing import Dict, Any, Optional
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class AgentforceClient:
"""Client for interacting with the Salesforce Agentforce API"""
def __init__(self, server_url: str, client_id: str, client_secret: str, agent_id: str):
self.server_url = server_url
self.client_id = client_id
self.client_secret = client_secret
self.agent_id = agent_id
self.token_url = f"https://{server_url}/services/oauth2/token"
self.api_url = "https://api.salesforce.com"
async def get_access_token(self, client_email: str) -> Optional[Dict[str, str]]:
"""Get an access token for the Agentforce API"""
try:
payload = {
'grant_type': 'client_credentials',
'client_id': self.client_id,
'client_secret': self.client_secret,
'client_email': client_email
}
headers = {
'Content-Type': 'application/x-www-form-urlencoded'
}
async with httpx.AsyncClient() as client:
response = await client.post(self.token_url, data=payload, headers=headers)
response.raise_for_status()
token_data = response.json()
logger.info(f"✅ Token retrieved for {client_email}")
return {
'client_email': client_email,
'access_token': token_data.get('access_token'),
'instance_url': token_data.get('instance_url')
}
except Exception as e:
logger.error(f"❌ Error retrieving token: {str(e)}")
return None
async def create_session(self, token: str, instance_url: str) -> Optional[str]:
"""Create a session with the Agentforce API"""
try:
session_url = f"{self.api_url}/einstein/ai-agent/v1/agents/{self.agent_id}/sessions"
random_uuid = str(uuid.uuid4())
payload = {
'externalSessionKey': random_uuid,
'instanceConfig': {
'endpoint': instance_url
},
'streamingCapabilities': {
'chunkTypes': ['Text']
},
'bypassUser': True
}
headers = {
'Authorization': f'Bearer {token}',
'Content-Type': 'application/json'
}
async with httpx.AsyncClient() as client:
response = await client.post(session_url, json=payload, headers=headers)
response.raise_for_status()
session_data = response.json()
session_id = session_data.get('sessionId')
if session_id:
logger.info(f"✅ Session Created, ID: {session_id}")
return session_id
else:
logger.error("❌ No session ID in response")
return None
except Exception as e:
logger.error(f"❌ Error creating session: {str(e)}")
return None
async def send_message(self, session_id: str, token: str, message: str, sequence_id: int = 1) -> Dict[str, Any]:
"""Send a message to the Agentforce API"""
try:
message_url = f"{self.api_url}/einstein/ai-agent/v1/sessions/{session_id}/messages"
payload = {
'message': {
'sequenceId': sequence_id,
'type': 'Text',
'text': message
}
}
headers = {
'Authorization': f'Bearer {token}',
'Content-Type': 'application/json'
}
logger.info(f"📤 Sending request with body: {json.dumps(payload)}")
async with httpx.AsyncClient() as client:
response = await client.post(message_url, json=payload, headers=headers, timeout=120.0)
response.raise_for_status()
response_data = response.json()
result = {
'session_id': session_id,
'sequence_id': sequence_id + 1, # Increment sequence ID for next message
'agent_response': None
}
if 'messages' in response_data and response_data['messages']:
result['agent_response'] = response_data['messages'][0].get('message')
else:
result['agent_response'] = 'No response message received'
return result
except Exception as e:
logger.error(f"❌ Error sending message: {str(e)}")
return {
'session_id': session_id,
'sequence_id': sequence_id,
'agent_response': f"Error sending message: {str(e)}"
}
```
--------------------------------------------------------------------------------
/setup.sh:
--------------------------------------------------------------------------------
```bash
#!/bin/bash
# Colors for terminal output
RED='\033[0;31m'
GREEN='\033[0;32m'
YELLOW='\033[1;33m'
BLUE='\033[0;34m'
CYAN='\033[0;36m'
NC='\033[0m' # No Color
# Get the absolute path of the installation directory
INSTALL_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
echo "Installation directory: $INSTALL_DIR"
clear
echo -e "${BLUE}======================================================${NC}"
echo -e "${BLUE} Agentforce MCP Server - Setup Script ${NC}"
echo -e "${BLUE}======================================================${NC}"
echo
# Update Homebrew if it exists
if command -v brew &> /dev/null; then
echo -e "${YELLOW}Updating Homebrew...${NC}"
brew update
echo -e "${GREEN}Homebrew updated successfully.${NC}"
# Check if Python is installed via Homebrew
if ! brew list [email protected] &> /dev/null; then
echo -e "${YELLOW}Installing Python 3.10 via Homebrew...${NC}"
brew install [email protected]
echo -e "${GREEN}Python 3.10 installed successfully.${NC}"
else
echo -e "${GREEN}Python 3.10 is already installed.${NC}"
fi
fi
# Check Python version
echo -e "${YELLOW}Checking Python version...${NC}"
if command -v python3 &> /dev/null; then
PYTHON_CMD="python3"
elif command -v python &> /dev/null; then
PYTHON_CMD="python"
else
echo -e "${RED}Python not found. Please install Python 3.10 or higher and try again.${NC}"
exit 1
fi
# Get Python version
PYTHON_VERSION=$($PYTHON_CMD --version | cut -d " " -f 2)
PYTHON_MAJOR=$(echo $PYTHON_VERSION | cut -d. -f1)
PYTHON_MINOR=$(echo $PYTHON_VERSION | cut -d. -f2)
echo -e "${GREEN}Found Python $PYTHON_VERSION${NC}"
# Check Python version is at least 3.10
if [ "$PYTHON_MAJOR" -lt 3 ] || ([ "$PYTHON_MAJOR" -eq 3 ] && [ "$PYTHON_MINOR" -lt 10 ]); then
echo -e "${RED}Python 3.10 or higher is required. You have Python $PYTHON_VERSION.${NC}"
echo -e "${RED}Please install a newer version of Python and try again.${NC}"
exit 1
fi
# Create and activate virtual environment
VENV_DIR="$INSTALL_DIR/.venv"
echo -e "${YELLOW}Creating virtual environment in $VENV_DIR...${NC}"
# Remove existing venv if it exists
if [ -d "$VENV_DIR" ]; then
echo -e "${YELLOW}Removing existing virtual environment...${NC}"
rm -rf "$VENV_DIR"
fi
# Create a new virtual environment
$PYTHON_CMD -m venv "$VENV_DIR"
if [ $? -ne 0 ]; then
echo -e "${RED}Failed to create virtual environment. Please check if you have venv module installed.${NC}"
echo -e "${YELLOW}You can install it with: pip install --user virtualenv${NC}"
exit 1
fi
echo -e "${GREEN}Virtual environment created successfully.${NC}"
# Activate the virtual environment
echo -e "${YELLOW}Activating virtual environment...${NC}"
source "$VENV_DIR/bin/activate"
if [ $? -ne 0 ]; then
echo -e "${RED}Failed to activate virtual environment.${NC}"
exit 1
fi
echo -e "${GREEN}Virtual environment activated successfully.${NC}"
# Upgrade pip in the virtual environment
echo -e "${YELLOW}Upgrading pip...${NC}"
pip install --upgrade pip
if [ $? -ne 0 ]; then
echo -e "${RED}Failed to upgrade pip. Continuing anyway...${NC}"
fi
# Install dependencies
echo -e "${YELLOW}Installing required dependencies...${NC}"
pip install -r "$INSTALL_DIR/requirements.txt"
if [ $? -ne 0 ]; then
echo -e "${RED}Failed to install dependencies. Please check the error message above and try again.${NC}"
exit 1
fi
echo -e "${GREEN}Dependencies installed successfully.${NC}"
# Make the server script executable
chmod +x "$INSTALL_DIR/agentforce_mcp_server.py"
# Check if .env file already exists
if [ -f "$INSTALL_DIR/.env" ]; then
echo -e "${YELLOW}An .env file already exists.${NC}"
read -p "Do you want to overwrite it? (y/n): " -n 1 -r
echo
if [[ ! $REPLY =~ ^[Yy]$ ]]; then
echo -e "${YELLOW}Keeping existing .env file.${NC}"
echo -e "${YELLOW}If you encounter issues, you can run this script again to recreate the file.${NC}"
else
# We'll create a new .env file below
echo -e "${YELLOW}Creating new .env file...${NC}"
fi
else
# Create .env file
echo -e "${YELLOW}Setting up your environment variables...${NC}"
fi
if [[ ! -f "$INSTALL_DIR/.env" || $REPLY =~ ^[Yy]$ ]]; then
# Get Salesforce credentials
echo -e "${CYAN}Please enter your Salesforce credentials:${NC}"
echo -e "${YELLOW}(Press Enter to skip any field and use empty value)${NC}"
read -p "Salesforce Org ID: " ORG_ID
read -p "Salesforce Agent ID: " AGENT_ID
read -p "Salesforce Client ID (Consumer Key): " CLIENT_ID
read -p "Salesforce Client Secret (Consumer Secret): " CLIENT_SECRET
read -p "Salesforce Server URL (e.g., example.my.salesforce.com): " SERVER_URL
# Create .env file
cat > "$INSTALL_DIR/.env" << EOF
SALESFORCE_ORG_ID="${ORG_ID}"
SALESFORCE_AGENT_ID="${AGENT_ID}"
SALESFORCE_CLIENT_ID="${CLIENT_ID}"
SALESFORCE_CLIENT_SECRET="${CLIENT_SECRET}"
SALESFORCE_SERVER_URL="${SERVER_URL}"
EOF
echo -e "${GREEN}.env file created successfully.${NC}"
fi
# Check if we want to test the setup
echo
read -p "Do you want to test the setup now? (y/n): " -n 1 -r TEST_SETUP
echo
if [[ $TEST_SETUP =~ ^[Yy]$ ]]; then
echo -e "${YELLOW}Testing Agentforce connection...${NC}"
echo -e "${YELLOW}This will attempt to authenticate and create a session.${NC}"
python "$INSTALL_DIR/test_agentforce.py"
if [ $? -ne 0 ]; then
echo -e "${RED}Test failed. Please check your credentials and try again.${NC}"
else
echo -e "${GREEN}Test completed. Your setup is working!${NC}"
fi
fi
# Determine the MCP configuration paths based on OS
PYTHON_PATH="$VENV_DIR/bin/python"
SERVER_SCRIPT="$INSTALL_DIR/agentforce_mcp_server.py"
# For Windows, adjust paths
if [[ "$OSTYPE" == "msys" || "$OSTYPE" == "win32" ]]; then
# Convert paths to Windows format
PYTHON_PATH=$(echo "$PYTHON_PATH" | sed 's/\//\\/g')
SERVER_SCRIPT=$(echo "$SERVER_SCRIPT" | sed 's/\//\\/g')
fi
# Generate the Claude Desktop configuration JSON
CONFIG_JSON=$(cat << EOF
{
"mcpServers": {
"agentforce": {
"command": "$PYTHON_PATH",
"args": [
"$SERVER_SCRIPT"
]
}
}
}
EOF
)
# Check if we want to start the server
echo
read -p "Do you want to start the MCP server now? (y/n): " -n 1 -r START_SERVER
echo
if [[ $START_SERVER =~ ^[Yy]$ ]]; then
echo -e "${YELLOW}Starting Agentforce MCP Server...${NC}"
echo -e "${YELLOW}Press Ctrl+C to stop the server.${NC}"
python "$INSTALL_DIR/agentforce_mcp_server.py"
else
# Determine OS-specific config path
if [[ "$OSTYPE" == "darwin"* ]]; then
CONFIG_PATH="$HOME/Library/Application Support/Claude/claude_desktop_config.json"
CONFIG_DIR="$HOME/Library/Application Support/Claude"
elif [[ "$OSTYPE" == "msys" || "$OSTYPE" == "win32" ]]; then
CONFIG_PATH="$APPDATA\\Claude\\claude_desktop_config.json"
CONFIG_DIR="$APPDATA\\Claude"
else
# Linux or other
CONFIG_PATH="$HOME/.config/Claude/claude_desktop_config.json"
CONFIG_DIR="$HOME/.config/Claude"
fi
# Provide instructions for setting up Claude Desktop
echo -e "${BLUE}======================================================${NC}"
echo -e "${BLUE} Claude Desktop Configuration Guide ${NC}"
echo -e "${BLUE}======================================================${NC}"
echo
echo -e "${CYAN}To use this server with Claude Desktop:${NC}"
echo
echo -e "1. Locate your Claude Desktop configuration file:"
echo -e " ${YELLOW}macOS:${NC} ~/Library/Application Support/Claude/claude_desktop_config.json"
echo -e " ${YELLOW}Windows:${NC} %APPDATA%\\Claude\\claude_desktop_config.json"
echo
echo -e "2. Add the following to your configuration:"
echo
echo -e "${YELLOW}"
echo "$CONFIG_JSON" | jq .
echo -e "${NC}"
echo
# Ask if user wants to automatically update the config
read -p "Do you want to automatically update your Claude Desktop configuration? (y/n): " -n 1 -r AUTO_CONFIG
echo
if [[ $AUTO_CONFIG =~ ^[Yy]$ ]]; then
# Create config directory if it doesn't exist
mkdir -p "$CONFIG_DIR"
# Check if config file exists
if [ -f "$CONFIG_PATH" ]; then
# Backup existing config
cp "$CONFIG_PATH" "${CONFIG_PATH}.backup"
echo -e "${YELLOW}Existing config backed up to ${CONFIG_PATH}.backup${NC}"
# Update existing config
if command -v jq &> /dev/null; then
# Use jq to merge configs if available
jq -s '.[0] * .[1]' "$CONFIG_PATH" <(echo "$CONFIG_JSON") > "${CONFIG_PATH}.tmp"
mv "${CONFIG_PATH}.tmp" "$CONFIG_PATH"
else
# Simple replacement if jq is not available
echo "$CONFIG_JSON" > "$CONFIG_PATH"
fi
else
# Create new config file
echo "$CONFIG_JSON" > "$CONFIG_PATH"
fi
echo -e "${GREEN}Claude Desktop configuration updated successfully.${NC}"
echo -e "${YELLOW}Please restart Claude Desktop to apply the changes.${NC}"
fi
echo -e "3. Restart Claude Desktop"
echo
echo -e "4. Start a conversation with Claude and look for the hammer icon in the input box"
echo -e " This indicates that MCP tools are available."
echo
echo -e "${GREEN}You can start the server at any time by running:${NC}"
echo -e "${YELLOW}cd $INSTALL_DIR && source .venv/bin/activate && python agentforce_mcp_server.py${NC}"
echo
echo -e "${BLUE}======================================================${NC}"
fi
echo -e "\n${GREEN}Setup complete!${NC}"
# Deactivate virtual environment if we're not starting the server
if [[ ! $START_SERVER =~ ^[Yy]$ ]]; then
deactivate 2>/dev/null || true
fi
```