星战科技xingzhan.cloud
返回市场
销售力-MCP服务器

销售力-MCP服务器

作者：LokiMCPUniverse4 星标更新：2025-10-27
安装使用来源：spider==github.com

项目介绍

Salesforce MCP Server

<div align="center">

Salesforce MCP Server

GitHub stars GitHub forks GitHub watchers

License Issues Pull Requests [Last Commit](https://github.com/L LokiMCPUniverse/salesforce-mcp-server/commits)

Python MCP

Commit Activity Code Size Contributors

</div>

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

功能

  • 全面的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凭证
SALESFORCE_USERNAME=your_username@company.com
SALESFORCE_PASSWORD=your_password
SALESFORCE_SECURITY_TOKEN=your_security_token
SALESFORCE_DOMAIN=login  # 或者test，或者自定义域名

# 或者使用OAuth
SALESFORCE_CLIENT_ID=your_connected_app_client_id
SALESFORCE_CLIENT_SECRET=your_connected_app_client_secret
SALESFORCE_REDIRECT_URI=http://localhost:8080/callback

# 可选设置
SALESFORCE_API_VERSION=59.0
SALESFORCE_SANDBOX=false
SALESFORCE_MAX_RETRIES=3
SALESFORCE_TIMEOUT=30

快速开始

基本用法

from salesforce_mcp import SalesforceMCPServer

# 初始化服务器
server = SalesforceMCPServer()

# 启动服务器
server.start()

Claude Desktop配置

添加到你的Claude Desktop配置中：

{
  "mcpServers": {
    "salesforce": {
      "command": "python",
      "args": ["-m", "salesforce_mcp.server"],
      "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"
          }
        ]
      }
    }
  }
}

高级配置

多组织支持

from salesforce_mcp import SalesforceMCPServer, OrgConfig

# 配置多个组织
orgs = {
    "production": OrgConfig(
        username="prod@company.com",
        password="prod_password",
        security_token="prod_token",
        domain="login"
    ),
    "sandbox": OrgConfig(
        username="sandbox@company.com.sandbox",
        password="sandbox_password",
        security_token="sandbox_token",
        domain="test"
    )
}

server = SalesforceMCPServer(orgs=orgs, default_org="production")

自定义身份验证

from salesforce_mcp import SalesforceMCPServer, JWTAuth

# JWT承载流
jwt_auth = JWTAuth(
    client_id="your_client_id",
    username="your_username",
    private_key_file="path/to/private_key.pem",
    sandbox=False
)

server = SalesforceMCPServer(auth=jwt_auth)

速率限制

from salesforce_mcp import SalesforceMCPServer, RateLimitConfig

rate_limit = RateLimitConfig(
    requests_per_second=10,
    burst_size=20,
    wait_on_limit=True
)

server = SalesforceMCPServer(rate_limit=rate_limit)

集成示例

查看examples/目录中的完整集成示例：

  • basic_usage.py - 简单查询和CRUD操作
  • bulk_operations.py - 处理大量数据
  • genai_integration.py - 与生成式AI 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错误: {e.error_code} - {e.message}")
    print(f"可用字段: {e.available_fields}")

安全最佳实践

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

贡献

欢迎贡献！请阅读我们的贡献指南并提交拉取请求。

许可证

MIT许可证 - 查看LICENSE文件了解详情