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

   
   cd backend/python/mcp_servers/mcp-salesforce-server
   
   

2. Install dependencies

   
   pip install uv
   
   uv pip install .
   
   

3. Configure environment variables

   
   cp env.example .env
   
   # Edit .env with your Salesforce credentials
   
   

Environment Configuration

Create a .env file with the following variables:

# 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:

python test_config.py

Local Development:

python salesforce_mcp_server.py

Docker:

docker-compose up --build

Direct Docker:

docker build -t salesforce-mcp-server .

docker run -p 8080:8080 --env-file .env salesforce-mcp-server

Health Check

Check server health:

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

# 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

# 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

{

    "last_name": "Doe",

    "first_name": "John",

    "company": "Tech Corp",

    "email": john.doe@techcorp.com,

    "phone": "+1234567890",

    "status": "New",

    "lead_source": "Website",

    "industry": "Technology"

}

Contact

{

    "last_name": "Smith",

    "first_name": "Jane",

    "email": jane.smith@company.com,

    "phone": "+1234567890",

    "title": "Sales Manager",

    "department": "Sales"

}

Account

{

    "name": "Acme Corporation",

    "type": "Customer",

    "industry": "Technology",

    "phone": "+1234567890",

    "website": https://www.acme.com

}

Opportunity

{

    "name": "Enterprise Deal",

    "stage_name": "Prospecting",

    "close_date": "2024-12-31",

    "amount": 50000.00,

    "type": "New Business",

    "probability": 25.0

}

Task

{

    "subject": "Follow up call",

    "description": "Call prospect about proposal",

    "status": "Not Started",

    "priority": "High",

    "activity_date": "2024-01-15"

}

Case

{

    "subject": "Technical Support Request",

    "description": "User cannot access the system",

    "status": "New",

    "priority": "Medium",

    "origin": "Web"

}

Campaign

{

    "name": "Q4 Product Launch",

    "type": "Email",

    "status": "Planned",

    "start_date": "2024-10-01",

    "end_date": "2024-12-31",

    "budget": 50000.00

}

Quote

{

    "name": "Enterprise Quote #001",

    "status": "Draft",

    "expiration_date": "2024-12-31",

    "subtotal": 100000.00,

    "total_price": 110000.00

}

Order

{

    "account_id": "001O400001BrpgiIAB",

    "effective_date": "2024-01-15",

    "status": "Draft",

    "total_amount": 50000.00,

    "type": "New"

}

Product

{

    "name": "Enterprise CRM License",

    "product_code": "CRM-ENT-001",

    "family": "Software",

    "is_active": true,

    "description": "Enterprise CRM software license"

}

Asset

{

    "name": "CRM Software License",

    "account_id": "001O400001BrpgiIAB",

    "product2_id": "01tO400000ABC123",

    "status": "Installed",

    "purchase_date": "2024-01-15"

}

Opportunity Line Item

{

    "opportunity_id": "006O400000ABC123",

    "product2_id": "01tO400000ABC123",

    "quantity": 10,

    "unit_price": 1000.00,

    "total_price": 10000.00

}

Pricebook2

{

    "name": "Standard Price Book",

    "is_active": true,

    "description": "Standard pricing for all products"

}

PricebookEntry

{

    "pricebook2_id": "01sO400000ABC123",

    "product2_id": "01tO400000ABC123",

    "unit_price": 1000.00,

    "is_active": true

}

Opportunity Contact Role

{

    "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

# 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:

pytest

Run with coverage:

pytest --cov=. --cov-report=html

Integration Tests

Run comprehensive integration tests:

python tests/run_integration_tests.py

Run specific integration test categories:

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):

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:

black .

Lint code:

flake8 .

Type checking:

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

3. Rate Limit Exceeded

   • Check Salesforce API limits

   • Implement request throttling

   • Use bulk operations for large datasets

4. Invalid Object Type

   • Verify object name spelling

   • Check if object is available in your Salesforce org

   • Ensure proper permissions

5. 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:

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