AI导哈

Salesforce MCP服务器

一个高度可定制的模型上下文协议（MCP）服务器，用于将Salesforce API与GenAI应用程序集成。

  • by LokiMCPUniverse
  • Python
  • 本地部署
  • 8 星标
  • 更新于
在 GitHub 查看 LokiMCPUniverse/salesforce-mcp-server

使用说明

Salesforce MCP服务器

Salesforce Mcp服务器

一个高度可定制的模型上下文协议（MCP）服务器，用于将Salesforce API与GenAI应用程序集成。

特性

  • 全面的Salesforce API覆盖范围:

    • 支持自动分页的SOQL查询
    • SOSL（Salesforce对象搜索语言）搜索
    • 记录CRUD操作（创建、读取、更新、删除）
    • 对象描述的元数据API访问
    • 大型数据量的批量API操作
    • Apex代码执行能力
    • 报告支持过滤的API
    • 组织限制和API使用监控

  • 灵活的身份验证:

    • OAuth 2.0 Web服务器流
    • OAuth 2.0 JWT承载流
    • 用户名密码流
    • 互联应用支持

  • 企业就绪:

    • 多组织支持
    • 速率限制和重试逻辑
    • 全面的错误处理
    • 审核日志记录
    • 现场级安全尊重

安装

pip install salesforce-mcp-server

或者从源代码安装：

git clone https://github.com/asklokesh/salesforce-mcp-server.git
cd salesforce-mcp-server
pip install -e .

配置

创建一个 .env 文件或设置环境变量：

# Salesforce Credentials
SALESFORCE_USERNAME=your_username@company.com
SALESFORCE_PASSWORD=your_password
SALESFORCE_SECURITY_TOKEN=your_security_token
SALESFORCE_DOMAIN=login  # or test, or your custom domain

# OR use OAuth
SALESFORCE_CLIENT_ID=your_connected_app_client_id
SALESFORCE_CLIENT_SECRET=your_connected_app_client_secret
SALESFORCE_REDIRECT_URI=http://localhost:8080/callback

# Optional Settings
SALESFORCE_API_VERSION=59.0
SALESFORCE_SANDBOX=false
SALESFORCE_MAX_RETRIES=3
SALESFORCE_TIMEOUT=30

需求

  • Python 3.10或更新版本
  • MCP SDK 1.27+

快速开始

运行服务器

服务器构建在MCP Python SDK的FastMCP API上，并使用stdio 运输。安装软件包，通过环境配置凭据 变量（请参阅配置），并使用以下方式启动：

salesforce-mcp
# or equivalently
python -m salesforce_mcp.server

程序化使用

from salesforce_mcp import mcp

# `mcp` is a configured `FastMCP` instance with every Salesforce tool
# already registered. Run it over stdio:
if __name__ == "__main__":
    mcp.run(transport="stdio")

Claude桌面配置

添加到您的Claude Desktop配置中：

{
  "mcpServers": {
    "salesforce": {
      "command": "salesforce-mcp",
      "env": {
        "SALESFORCE_USERNAME": "your_username@company.com",
        "SALESFORCE_PASSWORD": "your_password",
        "SALESFORCE_SECURITY_TOKEN": "your_security_token"
      }
    }
  }
}

可用工具

1.查询记录

执行SOQL查询以检索数据：

{
  "tool": "salesforce_query",
  "arguments": {
    "query": "SELECT Id, Name, Email FROM Contact WHERE LastModifiedDate = TODAY",
    "include_deleted": false
  }
}

2.获取记录

按ID检索特定记录：

{
  "tool": "salesforce_get_record",
  "arguments": {
    "object_type": "Account",
    "record_id": "001XX000003DHPh",
    "fields": ["Name", "Industry", "AnnualRevenue"]
  }
}

3.创建记录

创建新记录：

{
  "tool": "salesforce_create_record",
  "arguments": {
    "object_type": "Contact",
    "data": {
      "FirstName": "John",
      "LastName": "Doe",
      "Email": "john.doe@example.com",
      "AccountId": "001XX000003DHPh"
    }
  }
}

4.更新记录

更新现有记录：

{
  "tool": "salesforce_update_record",
  "arguments": {
    "object_type": "Contact",
    "record_id": "003XX000004TMM2",
    "data": {
      "Title": "Senior Developer",
      "Department": "Engineering"
    }
  }
}

5.删除记录

删除记录：

{
  "tool": "salesforce_delete_record",
  "arguments": {
    "object_type": "Contact",
    "record_id": "003XX000004TMM2"
  }
}

6.描述对象

获取Salesforce对象的元数据：

{
  "tool": "salesforce_describe_object",
  "arguments": {
    "object_type": "Account"
  }
}

7.批量操作

处理大量数据：

{
  "tool": "salesforce_bulk_create",
  "arguments": {
    "object_type": "Contact",
    "records": [
      {"FirstName": "Jane", "LastName": "Smith", "Email": "jane@example.com"},
      {"FirstName": "Bob", "LastName": "Johnson", "Email": "bob@example.com"}
    ],
    "batch_size": 200
  }
}

8.执行Apex

运行Apex代码：

{
  "tool": "salesforce_execute_apex",
  "arguments": {
    "apex_body": "System.debug('Hello from Apex!');"
  }
}

9.搜索记录（SOSL）

跨多个对象搜索：

{
  "tool": "salesforce_search",
  "arguments": {
    "search_query": "FIND {John} IN NAME FIELDS RETURNING Contact(Id, Name, Email), Account(Id, Name)"
  }
}

10.查询分页

处理大型查询结果：

{
  "tool": "salesforce_query_more",
  "arguments": {
    "next_records_url": "/services/data/v59.0/query/01gxx000000002ABC-2000"
  }
}

11.组织限制

获取API限制和使用：

{
  "tool": "salesforce_limits",
  "arguments": {}
}

12.运行报告

执行Salesforce报告：

{
  "tool": "salesforce_run_report",
  "arguments": {
    "report_id": "00O1i000004Gzr0EAC",
    "filters": {
      "reportMetadata": {
        "reportFilters": [
          {
            "column": "CREATED_DATE",
            "operator": "greaterThan",
            "value": "2023-01-01"
          }
        ]
      }
    }
  }
}

高级配置

多组织支持

通过设置前缀环境变量来配置其他组织。 例如，要暴露一个 sandbox org旁边的默认值：

SALESFORCE_USERNAME=prod@company.com
SALESFORCE_PASSWORD=prod_password
SALESFORCE_SECURITY_TOKEN=prod_token

SALESFORCE_SANDBOX_USERNAME=sandbox@company.com.sandbox
SALESFORCE_SANDBOX_PASSWORD=sandbox_password
SALESFORCE_SANDBOX_SECURITY_TOKEN=sandbox_token
SALESFORCE_SANDBOX_DOMAIN=test

每个工具都接受一个可选 org 论证；省略它的目标 默认组织由定义 SALESFORCE_DEFAULT_ORG （违约 "default").

身份验证类

这 salesforce_mcp.auth 模块暴露 UsernamePasswordAuth, OAuth2Auth, 和 JWTAuth 用于嵌入底层时 SalesforceClient 在 您自己的代码：

from salesforce_mcp import JWTAuth, SalesforceClient

auth = JWTAuth(
    client_id="your_client_id",
    username="your_username",
    private_key_file="path/to/private_key.pem",
    sandbox=False,
)
client = SalesforceClient(auth=auth)

速率限制

默认情况下启用速率限制，并通过环境变量进行调整：

SALESFORCE_RATE_LIMIT_ENABLED=true
SALESFORCE_RATE_LIMIT_REQUESTS_PER_SECOND=10
SALESFORCE_RATE_LIMIT_BURST_SIZE=20

集成示例

请参阅 examples/ 完整集成示例目录：

  • basic_usage.py -简单的查询和CRUD操作
  • bulk_operations.py -处理大量数据
  • genai_integration.py -与GenAI API集成
  • multi_org.py -管理多个Salesforce组织
  • oauth_flow.py -OAuth身份验证设置

错误处理

服务器提供详细的错误信息：

try:
    result = server.execute_tool("salesforce_query", {
        "query": "SELECT InvalidField FROM Account"
    })
except SalesforceError as e:
    print(f"Salesforce error: {e.error_code} - {e.message}")
    print(f"Fields available: {e.available_fields}")

安全最佳实践

  1. 从不提交凭据 -使用环境变量或安全保管库
  2. 尽可能使用OAuth -比用户名/密码更安全
  3. 实施现场级安全 -尊重Salesforce权限
  4. 启用审核日志记录 -跟踪所有API操作
  5. 使用IP限制 -限制对已知IP范围的访问

贡献

欢迎投稿！请阅读我们的投稿指南并提交pull请求。

许可证

MIT许可证-有关详细信息，请参阅许可证文件

相关 MCP