Salesforce+Pardot MCP服务器
用户通过以下方式连接 MCP OAuth 2.1 --Claude Desktop自动处理身份验证,无需配置。
运作原理
1. In Claude Desktop: Settings → Connectors → Add custom connector
2. Enter server URL: https://your-server.up.railway.app/sse
3. Claude Desktop opens a Salesforce login popup
4. User logs in → Done. All tools appear automatically.Claude Desktop通过PKCE安全的OAuth自动处理令牌管理(获取、存储、刷新)。服务器上的一个互联应用程序处理所有Salesforce组织中的所有用户。
可用工具(17个只读+5个写入)
服务器运行在 默认情况下为只读模式写入工具(更新/创建)仅在以下情况下注册 ENABLE_WRITE_TOOLS=true 已设置。
Salesforce工具(sf_*)
| 工具 | 模式 | 描述 |
|---|---|---|
sf_query | read | 运行任意SOQL SELECT查询(强制只读) |
sf_get_leads | read | 使用过滤器获取潜在客户(状态、创建时间、潜在客户来源) |
sf_get_contacts | read | 使用筛选器(姓名、电子邮件、帐户ID)获取联系人 |
sf_update_lead | 写 | 更新潜在客户字段(受保护的字段被阻止) |
sf_update_contact | 写 | 更新联系人字段(受保护的字段被阻止) |
sf_create_lead | 写 | 创建新的潜在客户(需要姓氏+公司) |
sf_pipeline_report | read | 按阶段汇总的开放机会 |
sf_get_tasks | read | 使用筛选器获取任务(谁/什么ID、状态、日期范围、主题) |
sf_get_events | read | 使用筛选器获取事件(谁/什么ID,日期时间范围) |
sf_get_activity_history | read | 按日期排序的组合任务+事件记录 |
Pardot工具(pardot_*)
| 工具 | 模式 | 描述 |
|---|---|---|
pardot_get_prospects | 阅读 | 使用过滤器(电子邮件、分数、活动)获取潜在客户 |
pardot_get_prospect_by_email | 阅读 | 通过电子邮件地址查找单个潜在客户 |
pardot_update_prospect | 写 | 更新潜在客户字段(受保护的字段被阻止) |
pardot_get_campaigns | read | 列出所有活动 |
pardot_get_lists | read | 列出所有列表 |
pardot_get_forms | read | 列出所有表单 |
pardot_add_prospect_to_list | 写 | 将潜在客户添加到列表中 |
pardot_get_visitor_activities | read | 通过类型丰富(标签+类别)和友好的名称过滤获取访问者活动(form_submit, email_open, bounce等等) |
pardot_get_form_handlers | read | 列出所有表单处理程序 |
pardot_get_emails | read | 列出电子邮件模板并发送 |
pardot_get_lifecycle_history | read | 获取潜在客户的生命周期阶段进展 |
pardot_set_business_unit | read | 为当前会话设置Pardot业务部门ID |
访客活动类型
pardot_get_visitor_activities 支持按友好名称过滤(activity_type_name)或数字代码(activity_type):
| 类别 | 名称 | 代码 |
|---|---|---|
| 网络 | click, view/form_view/page_view, error/form_error, success/form_submit/form_success, session, site_search, visit, custom_redirect | 1, 2, 3, 4, 5, 7, 20, 21 |
| 电子邮件 | email_sent, email_open, unsubscribe, bounce, spam, email_preference, opt_in, third_party_click | 6, 11, 12, 13, 14, 15, 16, 17 |
| 机会 | opportunity_created, opportunity_won, opportunity_lost, opportunity_reopen, opportunity_linked, opportunity_unlinked | 8, 9, 10, 18, 19, 38 |
每个返回的活动都丰富了 activityLabel 和 category 领域。
安全
| 功能 | 详细信息 |
|---|---|
| 默认情况下为只读 | 写入工具禁用,除非 ENABLE_WRITE_TOOLS=true 已设置 |
| 认证 | 承载令牌——来自MCP OAuth 2.1流的会话令牌 |
| MCP OAuth(PKCE S256) | 具有强制PKCE的授权码流,定时安全验证 |
| 重定向URI验证 | 只有 https:// 允许(+ http://localhost 对于dev);强制实施DCR注册URI |
| 会话TTL | 会话令牌在24小时后过期(可通过配置 SESSION_TTL_SECONDS) |
| SKIP_AUTH限制 | SKIP_AUTH 仅在stdio模式(本地)下工作,对远程SSE忽略 |
| 速率限制 | 每个令牌每分钟60个请求(滑动窗口) |
| DCR速率限制 | 每个IP每分钟10个客户端注册请求 |
| 内存限制 | 身份验证码(500)、注册客户端(200)、刷新令牌(1000)上限以防止DoS |
| 安全标头 | HSTS、X-内容类型选项、X-帧选项、CSP default-src 'none',缓存控制 no-store |
| SOQL注入保护 | 用户输入在包含在查询中之前已转义 |
| 只读执行 | sf_query 仅接受SELECT语句 |
| SF保护字段 | OwnerId, IsConverted, IsDeleted, MasterRecordId 无法更新(不区分大小写) |
| Pardot保护字段 | email, score, grade, isDoNotEmail, isDoNotCall, salesforceId, crmContactFid, crmLeadFid 无法更新 |
| Pardot ID验证 | 仅对潜在客户/列表ID进行数字验证可防止路径注入 |
| 截断错误 | API错误消息上限为200个字符,以防止组织数据泄露 |
| 审核日志记录 | 每个请求记录SHA-256密钥指纹 |
| 令牌加密 | 使用Fernet静态加密的每用户OAuth令牌(AES-128-CBC) |
| HMAC缓存密钥 | 客户端缓存使用HMAC哈希密钥来防止内存中的原始令牌暴露 |
| 实例URL验证 | 只有 *.salesforce.com, *.force.com, *.salesforce.mil, *.cloudforce.com 接受 |
| 输入净化 | 客户端名称已清理(控制字符已删除,长度有限) |
| 集管注入保护 | 在HTTP标头中使用之前,Pardot业务部门ID已验证为字母数字 |
先决条件
- Python 3.10+ (使用
str | None联合语法) - Salesforce互联应用 启用OAuth
创建互联应用
- 在Salesforce中: 设置→ 应用程序管理器→ 新连接的应用程序
- 启用OAuth设置
- 将回调URL设置为
https://your-server.up.railway.app/oauth/callback - 添加OAuth作用域:
api,refresh_token,pardot_api - 保存并记录 消费者密钥 和 消费者秘密
设置
本地开发
git clone https://github.com/DaniilMai/salesforce-pardot-mcp.git
cd salesforce-pardot-mcp
python3 -m venv .venv
source .venv/bin/activate
pip install -r requirements.txt
cp .env.example .env
# Edit .env with your Connected App credentials
python server.py
# → Listening on http://0.0.0.0:8000 (SSE)
# → Health check: http://localhost:8000/health铁路部署
- 将仓库推送到GitHub(或直接将Railway连接到仓库)
- 从仓库创建新的铁路服务
- 将环境变量(见下表)设置为铁路服务变量
- 将健康检查路径设置为
/health - 部署
包括 railway.toml 和 Dockerfile 处理剩下的。
# Or build and run manually:
docker build -t sf-mcp .
docker run -p 8000:8000 --env-file .env sf-mcp连接克劳德桌面
- 打开克劳德桌面: 设置→ 连接器→ 添加自定义连接器
- 输入URL:
https://your-server.up.railway.app/sse - 将OAuth客户端ID/密钥留空(动态客户端注册会处理它)
- 点击添加→ Salesforce登录弹出窗口→ Done
环境变量
| 变量 | 必填 | 描述 |
|---|---|---|
SF_OAUTH_CLIENT_ID | 是 | 连接的应用程序消费者密钥 |
SF_OAUTH_CLIENT_SECRET | 是 | 互联应用消费者机密 |
SF_OAUTH_REDIRECT_URI | 是 | OAuth回调URL(例如。 https://your-server/oauth/callback) |
SF_OAUTH_LOGIN_URL | no | salesforce登录URL(默认值: https://login.salesforce.com) |
ENCRYPTION_KEY | 是 | 用于令牌加密的Fernet密钥(见下文) |
PORT | 无 | 服务器端口(默认值: 8000) |
ENABLE_WRITE_TOOLS | 否 | 设置为 true 启用写入工具(默认:禁用,只读模式) |
SESSION_TTL_SECONDS | 否 | 会话令牌生存期(秒)(默认值: 86400 --24小时) |
生成加密密钥:
python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"看 .env.example 查看带有注释的完整模板。
运行测试
# Unit tests (no Salesforce connection needed)
python -m pytest tests/test_security.py -v
# MCP OAuth tests (PKCE, token exchange, security hardening)
python -m pytest tests/test_mcp_oauth.py -v
# Integration tests (starts server subprocess)
python -m pytest tests/test_integration.py -v
# All tests via Docker
docker build -f Dockerfile.test -t sf-pardot-mcp-tests .
docker run --rm sf-pardot-mcp-tests项目结构
server.py # Entry point — FastMCP + SSE + health + OAuth routes + security headers
auth.py # Bearer token middleware (session tokens from MCP OAuth)
user_context.py # ContextVar for per-request user identity
token_store.py # Fernet-encrypted per-user OAuth token storage (HMAC-keyed)
oauth.py # Shared OAuth utilities (SF config, instance URL validation, BUID detection)
mcp_oauth.py # MCP OAuth 2.1 Authorization Server (RFC 9728, RFC 8414, RFC 7591)
tools/
__init__.py # Re-exports ALL_TOOLS list (17 read + 5 write, write opt-in)
salesforce.py # 10 Salesforce tools (SOQL, CRUD, pipeline, activities)
pardot.py # 12 Pardot tools (prospects, campaigns, activities, emails, config)
tests/
test_security.py # Unit tests — SOQL injection, field protection, auth, rate limiting, activity types
test_mcp_oauth.py # MCP OAuth tests — PKCE, token exchange, DCR, redirect validation
test_integration.py # Integration tests — server startup, health, SSE
Dockerfile # Production container
Dockerfile.test # Test runner container
railway.toml # Railway deployment config
requirements.txt # Python dependencies
.env.example # Environment variable template许可证
麻省理工学院——见 许可证 了解详情。
