收入情报MCP服务器
一个生产就绪的MCP服务器，为大规模面向客户的业务团队演示ML系统集成模式。该服务器模拟了一个真实世界的基于机器学习的收入智能平台，展示了如何构建与业务工作流集成的可观察、可维护的机器学习系统。
业务环境
现代收入团队（销售、客户成功、营销）需要实时机器学习见解来优先考虑潜在客户，防止流失，并最大限度地提高转化率。此服务器演示了如何构建生产机器学习系统，该系统：
- 与业务工作流程集成 通过MCP资源、工具和提示
- 提供可解释的预测 具有特征归因
- 实现监控和可观察性 通过预测测井
- 支持生产ML模式 如版本控制、漂移检测和健康检查
这是您可以在高增长的SaaS公司中找到的支持收入运营的系统类型，与Salesforce、HubSpot或自定义CRM等工具集成。
架构概述
┌─────────────────────────────────────────────────────────────┐
│ MCP Server Interface │
│ (Resources, Tools, Prompts for Claude Desktop/API) │
└────────────────┬────────────────────────────────────────────┘
│
┌────────┴────────┐
│ │
┌───────▼──────┐ ┌──────▼────────┐ ┌──────────────┐
│ Scoring │ │ Data Store │ │ Config │
│ Engine │ │ (CRM Data) │ │ (Thresholds,│
│ │ │ │ │ Weights) │
│ • Lead Score │ │ • Accounts │ │ │
│ • Churn Risk │ │ • Leads │ │ • Model v1.2.3│
│ • Conversion │ │ • Pred Logs │ │ • Features │
└──────────────┘ └───────────────┘ └──────────────┘关键部件：
- MCP服务器 (
server.py)-通过MCP协议公开资源、工具和提示 - 评分引擎 (
scoring.py)-具有特征归因的ML预测逻辑 - 数据存储 (
data_store.py)-内存数据访问层（模拟数据库/仓库） - 配置 (
config.py)-模型参数、阈值、特征权重 - 模拟数据 (
mock_data.py)-20个客户，30个具有真实信号的线索
演示生产机器学习模式
此服务器展示了基本的生产机器学习工程模式：
1. 模型版本控制和元数据跟踪
- 显式模型版本(
v1.2.3)印在每一个预测上 - 跟踪培训日期和绩效指标
- 通过MCP资源记录和访问功能重要性
2. 监测预测记录
- 每个预测都记录了完整的输入/输出元数据
- 启用审计跟踪、调试和性能分析
- 漂移检测和模型再训练管道基础
3. 可解释性特征归因
- 每个预测都包括特征级属性
- 显示哪些信号推动了得分（例如，“演示请求”贡献了20%）
- 对收入团队的信任和监管合规性至关重要
4. 漂移检测框架
- 健康检查工具监控预测量和分布
- 当模式偏离训练基线时发出警报
- 在降级之前启用主动模型再训练
5. 与业务系统集成
- 资源通过标准URI公开CRM数据（客户、潜在客户）
- 工具映射到收入团队的工作流程（评分线索、检测流失）
- 提示为常见分析任务提供模板
6. 健康监测和SLO
check_model_health工具提供实时系统状态- 跟踪正常运行时间、预测量、准确性、漂移状态
- SLA监控和事件响应基础
7. 结构化错误处理
- 具有结构化上下文的全面日志记录
- 对缺失数据的优雅降级
- 清除错误消息以进行故障排除
安装
先决条件
- Python 3.10+
- pip或uv用于包装管理
设置
# Clone or navigate to the project
cd revenue-intel-mcp
# Create virtual environment
python -m venv venv
source venv/bin/activate # On Windows: venv\Scripts\activate
# Install dependencies
pip install -e ".[dev]"
# Run tests to verify installation
pytest tests/ -v用法
运行服务器
使用克劳德桌面
添加到您的Claude桌面配置(claude_desktop_config.json):
{
"mcpServers": {
"revenue-intel": {
"command": "python",
"args": [
"C:/Users/User/git-repo/revenue-intel-mcp/server.py"
]
}
}
}重新启动Claude Desktop，服务器将可用。
独立测试
# Run the server directly (for testing with MCP inspector)
python server.py可用资源
访问CRM数据和模型元数据：
crm://accounts/{account_id}-获取帐户详细信息
Example: crm://accounts/acc_001
Returns: Account data with usage signals, MRR, plan tiercrm://accounts/list-列出所有帐户
Returns: Array of all 20 sample accountscrm://leads/{lead_id}-获取潜在客户详细信息
Example: crm://leads/lead_001
Returns: Lead data with engagement signals, company infomodels://lead_scorer/metadata-模型元数据
Returns: Version, training date, performance metrics, feature importance, drift status可用工具
执行机器学习预测和监控：
1. score_lead
根据公司属性和参与度信号为潜在客户打分。
{
"company_name": "Acme Corp",
"signals": {
"website_visits_30d": 45,
"demo_requested": true,
"whitepaper_downloads": 3,
"email_engagement_score": 85,
"linkedin_engagement": true,
"free_trial_started": true
},
"industry": "technology",
"employee_count": 500
}返回：分数（0-100）、等级（热/暖/冷）、特征归因、解释
2. get_conversion_insights
预测试用到付费的转换概率。
{
"account_id": "acc_002"
}回报：转换概率、参与信号、建议行动
3. detect_churn_risk
分析账户健康状况并识别流失风险。
{
"account_id": "acc_006"
}回报：风险评分、风险等级、下降信号、干预建议
4. check_model_health
监控机器学习系统的健康状况和性能。
{}返回：模型版本、正常运行时间、预测计数、漂移状态、准确性
5. log_prediction
手动记录预测以进行监控。
{
"prediction_data": {
"prediction_type": "lead_score",
"input_data": {...},
"prediction_result": {...}
}
}返回：日志ID、时间戳、成功状态
可用提示
常见工作流的预构建模板：
analyze-account-expansion-CS团队追加销售分析
- 论点： account_id - 用例：评估客户对层升级的准备情况
weekly-lead-report-销售领导渠道报告
- 论点： week_number （可选） - 用例：每周潜在客户质量和速度分析
explain-low-score-领导评分解释
- 论点： lead_id - 用例：了解潜在客户得分低的原因以及如何改进
示例提示尝试
连接到Claude Desktop后，请尝试以下操作：
领先得分
“给我这个线索打分：Acme公司，科技行业，500名员工。他们访问了我们的网站50次，要求演示，下载了3份白皮书，电子邮件参与度得分为90，在领英上参与，并开始免费试用。”
抖动检测
“检查账户acc_006的流失风险”
转换分析
“acc_002试用帐户的转换概率是多少？我们应该做些什么来提高它？”
模型健康状况
“检查潜在客户评分模型的健康状况”
数据探索
“显示所有试用帐户并分析哪些帐户最有可能转换”
结构化分析
“使用帐户acc_001的分析帐户扩展提示”
测试
运行综合测试套件：
# Run all tests
pytest tests/ -v
# Run specific test file
pytest tests/test_scoring.py -v
# Run with coverage
pytest tests/ --cov=. --cov-report=html测试覆盖范围：
- ✅ 领先评分（热/温/冷级别）
- ✅ 流失风险检测
- ✅ 转换概率计算
- ✅ 特征归因生成
- ✅ 预测测井
- ✅ 数据访问层
- ✅ 边缘情况（数据缺失、输入无效）
- ✅ 模拟数据完整性
项目结构
revenue-intel-mcp/
├── server.py # MCP server with resources, tools, prompts
├── scoring.py # ML scoring logic (lead score, churn, conversion)
├── models.py # Data models with type hints (Account, Lead, etc.)
├── data_store.py # Data access layer (get/store operations)
├── mock_data.py # Sample accounts, leads, prediction logs
├── config.py # Model config, thresholds, feature weights
├── tests/
│ ├── __init__.py
│ ├── test_scoring.py # Scoring logic tests
│ └── test_tools.py # Data access and integration tests
├── pyproject.toml # Python package configuration
├── .gitignore
└── README.md配置
中的关键配置 config.py:
- 模型版本:
v1.2.3 - 领先层阈值：热（≥70）、温（40-70）、冷（\<40）
- 特征权重：公司规模（20%）、参与度（40%）、行业（20%）和意向（20%）
- 行业契合度得分：技术（90）、SaaS（85）、金融（80）等。
- 流失风险阈值：临界（≥70）、高（50-70）、中（30-50）、低（\<30）
样品数据
20个账户 跨行业：
- 3个试用账户（探索产品）
- 3个有风险的账户（使用率下降）
- 14个活跃账户（不同层次：初学者、专业人士、企业）
30条线索 质量参差不齐：
- 热门线索：高参与度、要求演示、企业规模
- 热门线索：中等参与度，中端市场
- 冷线索：参与度低，小公司
生产部署说明
此演示使用内存存储。 对于生产部署：
数据层
- 替换
mock_data.py连接到：
- 雪花/BigQuery 用于历史数据和特征存储 - PostgreSQL/MySQL 用于操作CRM数据 - 瑞迪斯 用于实时特征缓存
模特服务
- 将评分逻辑部署为：
- FastAPI/烧瓶 用于REST API的服务 - AWS Lambda/云功能 适用于无服务器 - SageMaker/顶点AI 用于托管ML服务
监控
- 实施生产监控：
- Datadog/New Relic 用于应用程序度量 - 流量/权重和偏差 用于ML实验跟踪 - 格拉法纳/基巴纳 用于预测漂移仪表板 - PagerDuty 用于警报路由
MLOps管道
- 建立模型生命周期管理：
- 特征管线 （dbt，Airflow）用于数据新鲜度 - 培训管道 具有版本控制（Git、DVC） - A/B测试 模型评估框架 - 自动化再培训 基于漂移检测 - 影子部署 在推出之前进行验证
数据质量
- 添加全面的数据验证：
- 远大前程 用于输入数据质量检查 - 模式演变 处理Pydantic - 特征漂移监测 反对培训分配
安全与合规
- 实施安全控制：
- 身份验证/授权 用于API访问 - PII处理 数据匿名化 - 审核日志记录 监管合规（GDPR、SOC2） - 速率限制 DDoS防护
许可证
MIT许可证-可以随意用作您自己的机器学习系统的模板。
贡献
这是一个示范项目。对于生产使用，请根据您的具体情况调整模式：
- 数据基础设施（仓库、功能库、CRM）
- 机器学习框架（scikit-learn、XGBoost、PyTorch）
- 部署环境（云提供商、Kubernetes、无服务器）
- 监控和可观察性堆栈
______________________________________________________________________
内置： Python 3.10+| MCP SDK |类型提示|结构化日志| pytest
演示： 生产机器学习模式|业务集成|可观察性|可解释性