Salesforce MCP Server
README.md
# Salesforce MCP Server
A Model Context Protocol (MCP) server that provides tools for interacting with Salesforce via REST API, Bulk API, and SOQL using a tier 3 architecture.
**Version**: 0.1.0
**Last Updated**: January 2025
**Test Coverage**: 90%+
**Integration Tests**: 100+ test cases
## Features
### Supported Salesforce Objects
- **Lead**: Create, read, update, delete, and query leads
- **Contact**: Create, read, update, delete, and query contacts
- **Account**: Create, read, update, delete, and query accounts
- **Opportunity**: Create, read, update, delete, and query opportunities
- **Task**: Create, read, update, delete, and query tasks
- **Case**: Create, read, update, delete, and query cases
- **Campaign**: Create, read, update, delete, and query campaigns
- **Quote**: Create, read, update, delete, and query quotes
- **Order**: Create, read, update, delete, and query orders
- **Product**: Create, read, update, delete, and query products
- **Asset**: Create, read, update, delete, and query assets
- **OpportunityLineItem**: Create, read, update, delete, and query opportunity line items
- **Pricebook2**: Create, read, update, delete, and query pricebooks
- **PricebookEntry**: Create, read, update, delete, and query pricebook entries
- **OpportunityContactRole**: Create, read, update, delete, and query opportunity contact roles
### APIs Supported
- **REST API**: Standard CRUD operations for all objects
- **Bulk API**: Bulk operations for large datasets
- **SOQL**: Salesforce Object Query Language for complex queries
### Architecture
- **Presentation Layer (Controllers)**: Handle MCP tool definitions and responses
- **Business Logic Layer (Services)**: Handle business logic and validation
- **Data Access Layer (Repositories)**: Handle data access and external API calls
## Setup
### Prerequisites
- Python 3.10+
- Salesforce account with API access
- Connected App configured in Salesforce
### Installation
1. **Clone the repository**
```bash
cd backend/python/mcp_servers/mcp-salesforce-server
```
2. **Install dependencies**
```bash
pip install uv
uv pip install .
```
3. **Configure environment variables**
```bash
cp env.example .env
# Edit .env with your Salesforce credentials
```
### Environment Configuration
Create a `.env` file with the following variables:
```env
# Salesforce Authentication
SALESFORCE_CLIENT_ID=your_client_id_here
SALESFORCE_CLIENT_SECRET=your_client_secret_here
SALESFORCE_USERNAME=your_username_here
SALESFORCE_PASSWORD=your_password_here
SALESFORCE_SECURITY_TOKEN=your_security_token_here
# Salesforce Instance Configuration
SALESFORCE_INSTANCE_URL=https://your-instance.salesforce.com
SALESFORCE_LOGIN_URL=https://login.salesforce.com
SALESFORCE_SANDBOX_URL=https://test.salesforce.com
# API Configuration
SALESFORCE_API_VERSION=v58.0
SALESFORCE_USE_SANDBOX=false
# Server Configuration
HOST=0.0.0.0
PORT=8080
# Logging Configuration
LOG_LEVEL=INFO
# Optional: Custom timeout settings (in seconds)
SALESFORCE_REQUEST_TIMEOUT=30
SALESFORCE_BULK_TIMEOUT=300
# Azure Key Vault Configuration (optional)
AGENTOS_AKV_NAME=your_key_vault_name
AZURE_KEYVAULT_NAME=your_key_vault_name
AKV_NAME=your_key_vault_name
# MCP Server Configuration
MCP_SERVER_NAME=SalesforceServer
MCP_SERVER_INSTRUCTIONS=A powerful server for Salesforce operations and CRM management.
MCP_STATELESS_HTTP=true
MCP_JSON_RESPONSE=true
MCP_STREAM_PATH=/stream
MCP_SERVER_URL=http://localhost:8080/mcp-salesforce-server/
# Authentication & Security
MCP_TOKEN=your_optional_mcp_token
CORS_ALLOW_ORIGINS=*
# Environment Configuration
ENVIRONMENT=development
FASTAPI_PREFIX=salesforce
# Logging Configuration
LOG_LEVEL=INFO
LOG_FORMAT=json
LOG_FILE=./logs
LOG_MAX_SIZE=10485760
LOG_BACKUP_COUNT=5
```
## Usage
### Running the Server
**Test Configuration First:**
```bash
python test_config.py
```
**Local Development:**
```bash
python salesforce_mcp_server.py
```
**Docker:**
```bash
docker-compose up --build
```
**Direct Docker:**
```bash
docker build -t salesforce-mcp-server .
docker run -p 8080:8080 --env-file .env salesforce-mcp-server
```
### Health Check
Check server health:
```bash
curl http://localhost:8080/health
```
### MCP Tools Available
#### Lead Operations
- `create_lead(lead_data)`: Create a new lead
- `get_lead(lead_id, fields)`: Get lead by ID
- `update_lead(lead_id, lead_data)`: Update existing lead
- `delete_lead(lead_id)`: Delete lead
- `query_leads(where_conditions, limit)`: Query leads using SOQL
#### Contact Operations
- `create_contact(contact_data)`: Create a new contact
- `get_contact(contact_id, fields)`: Get contact by ID
- `update_contact(contact_id, contact_data)`: Update existing contact
- `delete_contact(contact_id)`: Delete contact
- `query_contacts(where_conditions, limit)`: Query contacts using SOQL
#### Account Operations
- `create_account(account_data)`: Create a new account
- `get_account(account_id, fields)`: Get account by ID
- `update_account(account_id, account_data)`: Update existing account
- `delete_account(account_id)`: Delete account
- `query_accounts(where_conditions, limit)`: Query accounts using SOQL
#### Opportunity Operations
- `create_opportunity(opportunity_data)`: Create a new opportunity
- `get_opportunity(opportunity_id, fields)`: Get opportunity by ID
- `update_opportunity(opportunity_id, opportunity_data)`: Update existing opportunity
- `delete_opportunity(opportunity_id)`: Delete opportunity
- `query_opportunities(where_conditions, limit)`: Query opportunities using SOQL
#### Task Operations
- `create_task(task_data)`: Create a new task
- `get_task(task_id, fields)`: Get task by ID
- `update_task(task_id, task_data)`: Update existing task
- `delete_task(task_id)`: Delete task
- `query_tasks(where_conditions, limit)`: Query tasks using SOQL
#### Case Operations
- `create_case(case_data)`: Create a new case
- `get_case(case_id, fields)`: Get case by ID
- `update_case(case_id, case_data)`: Update existing case
- `delete_case(case_id)`: Delete case
- `query_cases(where_conditions, limit)`: Query cases using SOQL
#### Campaign Operations
- `create_campaign(campaign_data)`: Create a new campaign
- `get_campaign(campaign_id, fields)`: Get campaign by ID
- `update_campaign(campaign_id, campaign_data)`: Update existing campaign
- `delete_campaign(campaign_id)`: Delete campaign
- `query_campaigns(where_conditions, limit)`: Query campaigns using SOQL
#### Quote Operations
- `create_quote(quote_data)`: Create a new quote
- `get_quote(quote_id, fields)`: Get quote by ID
- `update_quote(quote_id, quote_data)`: Update existing quote
- `delete_quote(quote_id)`: Delete quote
- `query_quotes(where_conditions, limit)`: Query quotes using SOQL
#### Order Operations
- `create_order(order_data)`: Create a new order
- `get_order(order_id, fields)`: Get order by ID
- `update_order(order_id, order_data)`: Update existing order
- `delete_order(order_id)`: Delete order
- `query_orders(where_conditions, limit)`: Query orders using SOQL
#### Product Operations
- `create_product(product_data)`: Create a new product
- `get_product(product_id, fields)`: Get product by ID
- `update_product(product_id, product_data)`: Update existing product
- `delete_product(product_id)`: Delete product
- `query_products(where_conditions, limit)`: Query products using SOQL
#### Asset Operations
- `create_asset(asset_data)`: Create a new asset
- `get_asset(asset_id, fields)`: Get asset by ID
- `update_asset(asset_id, asset_data)`: Update existing asset
- `delete_asset(asset_id)`: Delete asset
- `query_assets(where_conditions, limit)`: Query assets using SOQL
#### Opportunity Line Item Operations
- `create_opportunity_line_item(line_item_data)`: Create a new opportunity line item
- `get_opportunity_line_item(line_item_id, fields)`: Get opportunity line item by ID
- `update_opportunity_line_item(line_item_id, line_item_data)`: Update existing opportunity line item
- `delete_opportunity_line_item(line_item_id)`: Delete opportunity line item
- `query_opportunity_line_items(where_conditions, limit)`: Query opportunity line items using SOQL
#### Pricebook Operations
- `create_pricebook2(pricebook_data)`: Create a new pricebook
- `get_pricebook2(pricebook_id, fields)`: Get pricebook by ID
- `update_pricebook2(pricebook_id, pricebook_data)`: Update existing pricebook
- `delete_pricebook2(pricebook_id)`: Delete pricebook
- `query_pricebook2s(where_conditions, limit)`: Query pricebooks using SOQL
#### Pricebook Entry Operations
- `create_pricebook_entry(entry_data)`: Create a new pricebook entry
- `get_pricebook_entry(entry_id, fields)`: Get pricebook entry by ID
- `update_pricebook_entry(entry_id, entry_data)`: Update existing pricebook entry
- `delete_pricebook_entry(entry_id)`: Delete pricebook entry
- `query_pricebook_entries(where_conditions, limit)`: Query pricebook entries using SOQL
#### Opportunity Contact Role Operations
- `create_opportunity_contact_role(role_data)`: Create a new opportunity contact role
- `get_opportunity_contact_role(role_id, fields)`: Get opportunity contact role by ID
- `update_opportunity_contact_role(role_id, role_data)`: Update existing opportunity contact role
- `delete_opportunity_contact_role(role_id)`: Delete opportunity contact role
- `query_opportunity_contact_roles(where_conditions, limit)`: Query opportunity contact roles using SOQL
#### Health Check
- `health_check()`: Check server health and Salesforce connection
## SOQL Query Examples
### Basic Queries
```python
# Get all leads
query_leads()
# Get leads with specific conditions
query_leads("Status = 'New' AND Company LIKE '%Tech%'")
# Get contacts with limit
query_contacts("Email != null", limit=10)
# Get accounts by industry
query_accounts("Industry = 'Technology'", limit=50)
```
### Complex Queries
```python
# Get opportunities with related account data
query_opportunities("StageName = 'Closed Won' AND Amount > 10000")
# Get tasks by status
query_tasks("Status = 'In Progress'")
# Get cases by priority
query_cases("Priority = 'High' AND Status != 'Closed'")
```
## Data Models
### Lead
```python
{
"last_name": "Doe",
"first_name": "John",
"company": "Tech Corp",
"email": john.doe@techcorp.com,
"phone": "+1234567890",
"status": "New",
"lead_source": "Website",
"industry": "Technology"
}
```
### Contact
```python
{
"last_name": "Smith",
"first_name": "Jane",
"email": jane.smith@company.com,
"phone": "+1234567890",
"title": "Sales Manager",
"department": "Sales"
}
```
### Account
```python
{
"name": "Acme Corporation",
"type": "Customer",
"industry": "Technology",
"phone": "+1234567890",
"website": https://www.acme.com
}
```
### Opportunity
```python
{
"name": "Enterprise Deal",
"stage_name": "Prospecting",
"close_date": "2024-12-31",
"amount": 50000.00,
"type": "New Business",
"probability": 25.0
}
```
### Task
```python
{
"subject": "Follow up call",
"description": "Call prospect about proposal",
"status": "Not Started",
"priority": "High",
"activity_date": "2024-01-15"
}
```
### Case
```python
{
"subject": "Technical Support Request",
"description": "User cannot access the system",
"status": "New",
"priority": "Medium",
"origin": "Web"
}
```
### Campaign
```python
{
"name": "Q4 Product Launch",
"type": "Email",
"status": "Planned",
"start_date": "2024-10-01",
"end_date": "2024-12-31",
"budget": 50000.00
}
```
### Quote
```python
{
"name": "Enterprise Quote #001",
"status": "Draft",
"expiration_date": "2024-12-31",
"subtotal": 100000.00,
"total_price": 110000.00
}
```
### Order
```python
{
"account_id": "001O400001BrpgiIAB",
"effective_date": "2024-01-15",
"status": "Draft",
"total_amount": 50000.00,
"type": "New"
}
```
### Product
```python
{
"name": "Enterprise CRM License",
"product_code": "CRM-ENT-001",
"family": "Software",
"is_active": true,
"description": "Enterprise CRM software license"
}
```
### Asset
```python
{
"name": "CRM Software License",
"account_id": "001O400001BrpgiIAB",
"product2_id": "01tO400000ABC123",
"status": "Installed",
"purchase_date": "2024-01-15"
}
```
### Opportunity Line Item
```python
{
"opportunity_id": "006O400000ABC123",
"product2_id": "01tO400000ABC123",
"quantity": 10,
"unit_price": 1000.00,
"total_price": 10000.00
}
```
### Pricebook2
```python
{
"name": "Standard Price Book",
"is_active": true,
"description": "Standard pricing for all products"
}
```
### PricebookEntry
```python
{
"pricebook2_id": "01sO400000ABC123",
"product2_id": "01tO400000ABC123",
"unit_price": 1000.00,
"is_active": true
}
```
### Opportunity Contact Role
```python
{
"opportunity_id": "006O400000ABC123",
"contact_id": "003O400000ABC123",
"role": "Decision Maker",
"is_primary": true
}
```
## Security Features
### **Multi-Layer Security Architecture**
- **SOQL Injection Prevention**: Validates and sanitizes all SOQL queries with advanced pattern matching
- **Field Validation**: Ensures only valid fields are used in queries with comprehensive field mapping
- **Rate Limiting**: Respects Salesforce API rate limits with intelligent retry logic
- **Authentication**: Secure OAuth2 authentication with Salesforce and automatic token refresh
- **Error Handling**: Comprehensive error handling and logging with correlation IDs
### **Enhanced Security Features**
- **Azure Key Vault Integration**: Secure secrets management for production environments
- **Input Validation**: Pydantic v2 models with comprehensive field validation
- **CORS Configuration**: Configurable cross-origin resource sharing policies
- **Container Security**: Non-root user execution and minimal base image
- **Structured Logging**: JSON logging with security event tracking
- **Request Timeouts**: Configurable timeouts to prevent hanging connections
- **Retry Logic**: Exponential backoff for failed authentication attempts
## Data Quality Management
### Duplicate Prevention
The server includes built-in capabilities to help manage data quality:
- **Duplicate Detection**: Query tools can identify duplicate records by name, email, or other key fields
- **Bulk Operations**: Use bulk delete operations to clean up duplicate data efficiently
- **Data Validation**: Field validation ensures data consistency before creation
### Example Duplicate Cleanup
```python
# Query for duplicate accounts
query_accounts("Name = 'Duplicate Company Name'")
# Delete specific duplicate records
delete_account("001O400000ABC123")
delete_account("001O400000XYZ789")
# Query for duplicate leads
query_leads("Company = 'Duplicate Company' AND Email = 'duplicate@email.com'")
```
### Best Practices
- Regularly audit data for duplicates using SOQL queries
- Implement data validation rules in Salesforce
- Use bulk operations for large-scale data cleanup
- Monitor data quality metrics through regular queries
## Rate Limits
The server respects Salesforce API rate limits:
- **REST API**: 15,000 requests per 24 hours
- **Bulk API**: 10,000 requests per 24 hours
- **Query API**: 5,000 requests per 24 hours
## Error Handling
The server provides structured error responses for:
- Authentication failures
- Invalid object types
- Missing required fields
- SOQL injection attempts
- Rate limit exceeded
- Network errors
## Development
### Project Structure
```
mcp-salesforce-server/
├── config/
│ ├── logging_config.py # Logging configuration with correlation IDs
│ └── settings.py # Application settings with Pydantic v2
├── constants/
│ └── salesforce_constants.py # Non-sensitive configuration
├── controllers/
│ └── salesforce_controller.py # Presentation layer
├── models/
│ └── salesforce_models.py # Data models with validation
├── repositories/
│ └── salesforce_repository.py # Data access layer with retry logic
├── services/
│ └── salesforce_service.py # Business logic layer
├── tests/ # Comprehensive test suite
│ ├── conftest.py # Test configuration and fixtures
│ ├── run_tests.py # Test runner for all tests
│ ├── run_integration_tests.py # Integration test runner
│ ├── test_salesforce_mcp_server.py
│ ├── constants/ # Unit tests for constants
│ ├── controllers/ # Unit tests for controllers
│ ├── models/ # Unit tests for models
│ ├── repositories/ # Unit tests for repositories
│ ├── services/ # Unit tests for services
│ └── integration/ # Integration tests
│ ├── conftest.py # Integration test fixtures
│ ├── test_lead_integration.py
│ ├── test_all_objects_crud_integration.py
│ ├── test_end_to_end_workflows.py
│ └── README.md # Integration test documentation
├── logs/ # Log files directory
├── dependencies.py # Dependency injection container
├── salesforce_mcp_server.py # Main server file
├── pyproject.toml # Dependencies and project config
├── pytest.ini # Pytest configuration
├── Dockerfile # Container configuration
├── docker-compose.yml # Docker orchestration
├── env.example # Environment template
└── README.md # This file
```
### Testing
#### **Unit Tests**
Run all unit tests:
```bash
pytest
```
Run with coverage:
```bash
pytest --cov=. --cov-report=html
```
#### **Integration Tests**
Run comprehensive integration tests:
```bash
python tests/run_integration_tests.py
```
Run specific integration test categories:
```bash
python tests/run_integration_tests.py crud # CRUD operations
python tests/run_integration_tests.py lead # Lead workflows
python tests/run_integration_tests.py workflows # Business workflows
```
#### **Complete Test Suite**
Run all tests (unit + integration):
```bash
python tests/run_tests.py all
```
#### **Test Coverage**
- **90%+ Code Coverage** across all layers
- **100+ Integration Test Cases** for end-to-end validation
- **15+ Business Workflows** tested
- **50+ Error Scenarios** covered
### Code Quality
Format code:
```bash
black .
```
Lint code:
```bash
flake8 .
```
Type checking:
```bash
mypy .
```
## Troubleshooting
### Common Issues
1. **Configuration Errors**
- Run `python test_config.py` to check your configuration
- Ensure all required environment variables are set in `.env`
- Copy `env.example` to `.env` and fill in your credentials
2. **Authentication Failed**
- Verify Salesforce credentials in `.env`
- Check if security token is correct
- Ensure connected app is properly configured
2. **Rate Limit Exceeded**
- Check Salesforce API limits
- Implement request throttling
- Use bulk operations for large datasets
3. **Invalid Object Type**
- Verify object name spelling
- Check if object is available in your Salesforce org
- Ensure proper permissions
4. **SOQL Query Errors**
- Validate query syntax
- Check field names and relationships
- Ensure proper escaping of special characters
### Logs
Check server logs for detailed error information:
```bash
docker-compose logs salesforce-mcp-server
```
## Contributing
1. Fork the repository
2. Create a feature branch
3. Make your changes
4. Add tests
5. Submit a pull request
## License
This project is licensed under the MIT License.
## Changelog
### Recent Updates (Last 2 Weeks)
#### 🔧 **Enhanced Security & Configuration**
- **Azure Key Vault Integration**: Added support for Azure Key Vault secrets management
- **Enhanced Environment Configuration**: Improved settings validation with Pydantic v2
- **Security Token Management**: Enhanced Salesforce security token handling
- **CORS Configuration**: Improved cross-origin resource sharing settings
#### 🧪 **Comprehensive Testing Suite**
- **Integration Test Framework**: Added complete integration test suite with 100+ test cases
- **End-to-End Workflow Testing**: Business process validation across all Salesforce objects
- **90%+ Code Coverage**: Comprehensive unit and integration test coverage
- **Mock Strategy**: Advanced HTTP-level mocking for realistic API simulation
- **Test Automation**: Automated test runners for unit and integration tests
#### 🏗️ **Architecture Improvements**
- **Tier 3 Architecture**: Fully implemented presentation, business, and data layers
- **Dependency Injection**: Enhanced dependency container with proper error handling
- **Structured Logging**: JSON logging with correlation IDs and session tracking
- **Error Handling**: Comprehensive error handling across all architectural layers
#### 📊 **Enhanced Data Models**
- **Pydantic v2 Migration**: Updated to latest Pydantic with enhanced validation
- **Field Validation**: Email, phone, website, and business rule validation
- **Type Safety**: Improved type hints and validation across all models
- **Data Quality**: Enhanced duplicate detection and data integrity features
#### 🚀 **Performance & Scalability**
- **Request Timeouts**: Configurable timeout settings for API requests
- **Retry Logic**: Exponential backoff for failed authentication attempts
- **Session Management**: HTTP session reuse for improved performance
- **Bulk Operations**: Enhanced bulk API support for large datasets
#### 🔒 **Security Enhancements**
- **SOQL Injection Prevention**: Enhanced query sanitization and validation
- **Authentication Flow**: Improved OAuth 2.0 flow with automatic token refresh
- **Input Validation**: Comprehensive input validation and sanitization
- **Secrets Management**: Azure Key Vault integration for secure credential storage
#### 📈 **API & Integration**
- **13 Salesforce Objects**: Full CRUD support for all major Salesforce objects
- **65+ MCP Tools**: Comprehensive tool set for Salesforce operations
- **Health Monitoring**: Enhanced health checks and status endpoints
- **API Version Management**: Support for multiple Salesforce API versions (v58.0+)
#### 🐳 **Container & Deployment**
- **Docker Security**: Non-root user execution and security hardening
- **Health Checks**: Container health monitoring with curl-based checks
- **Environment Management**: Improved environment variable handling
- **Docker Compose**: Enhanced orchestration with proper environment passing
#### 📚 **Documentation & Developer Experience**
- **Comprehensive README**: Updated with latest features and setup instructions
- **Integration Test Documentation**: Detailed testing guides and examples
- **API Documentation**: Complete MCP tool documentation with examples
- **Troubleshooting Guide**: Enhanced troubleshooting and debugging information