内容
# AEP MCP 服务器
两个 MCP 服务器，允许 Claude（和任何兼容 MCP 的客户端）访问 Adobe Experience Platform、Adobe Journey Optimizer 和 Real-Time CDP。
| 服务器 | 文件 | 工具 | 描述 |
|--------|------|-------|-------------|
| **AEP**（自定义构建） | `server.py` | 115+ | 使用 FastMCP 构建的完整 AEP/AJO API 覆盖 |
| **RT-CDP**（Adobe 官方代理） | `rtcdp_proxy.py` | 18 | 代理 Adobe 官方的 `rtcdp-mcp.adobe.io` 服务器 |
---
## 快速入门
### 1. 安装依赖
```bash
pip install -r requirements.txt
```
### 2. 配置凭证
复制示例并填写来自 **Adobe Developer Console → 项目 → OAuth 服务器到服务器** 的凭证：
```bash
cp orgs.example.json orgs.json
# 使用您的 client_id、client_secret、org_id、sandbox 编辑 orgs.json
```
或使用环境变量（单组织备用）：
```bash
cp .env.example .env
# 编辑 .env
```
### 3. 注册 Claude 代码
复制 `.mcp.json.example` 到 `.mcp.json` 并更新路径：
```bash
cp .mcp.json.example .mcp.json
# 编辑 .mcp.json — 更新 /path/to/aep-mcp-server 到您的实际路径
```
或手动添加到 `.claude/mcp_servers.json`：
```json
{
"aep": {
"command": "python",
"args": ["/path/to/aep-mcp-server/server.py"]
},
"rtcdp": {
"command": "python",
"args": ["/path/to/aep-mcp-server/rtcdp_proxy.py"]
}
}
```
### 4. 安装预提交钩子（推荐）
阻止意外凭证提交：
```bash
python scripts/install_hooks.py
```
### 5. 运行
```bash
python server.py # AEP 服务器，默认组织配置文件
AEP_PROFILE=MyClient python server.py # 使用特定配置文件启动
```
---
## 多组织配置
`orgs.json` 支持多个命名配置文件 — 在跨不同 Adobe 组织或沙盒工作时非常有用：
```json
{
"default": "ClientA",
"profiles": {
"ClientA": {
"client_id": "...",
"client_secret": "...",
"api_key": "...",
"org_id": "XXXX@AdobeOrg",
"sandbox": "prod",
"sandboxes": {
"dev": "my-sandbox-dev",
"prod": "my-sandbox-prod"
}
},
"ClientB": {
"client_id": "...",
"client_secret": "...",
"api_key": "...",
"org_id": "YYYY@AdobeOrg",
"sandbox": "prod"
}
}
}
```
在运行时使用 `switch_org_profile` 工具或设置 `AEP_PROFILE` 切换组织。令牌缓存按配置文件维护。
---
## 服务器 1 — AEP MCP (`server.py`)
使用 [FastMCP](https://github.com/jlowin/fastmcp) 构建。所有工具都接受可选的 `sandbox` 参数；如果省略，则使用活动配置文件的默认沙盒。
### 数据集和批次
`list_datasets` · `get_dataset` · `create_dataset` · `get_batch` · `list_batches` · `get_sandbox` · `list_sandboxes` · `reset_sandbox`
### XDM 架构注册表
`list_schemas` · `get_schema` · `create_schema` · `enable_schema_for_profile` · `list_field_groups` · `get_field_group` · `create_field_group` · `list_classes` · `list_data_types` · `list_descriptors` · `create_descriptor`
### 实时客户资料和身份
`get_profile_by_identity` · `get_identity_cluster` · `list_identity_namespaces` · `get_identity_namespace` · `list_merge_policies` · `list_profile_export_jobs` · `get_profile_export_job`
### 分段/受众
`list_segments` · `get_segment` · `create_segment` · `delete_segment` · `list_segment_jobs` · `get_segment_job` · `create_segment_job` · `list_streaming_jobs`
### 查询服务
`list_queries` · `get_query` · `run_query` · `cancel_query` · `list_query_runs` · `list_query_templates` · `get_query_template` · `create_query_template` · `list_scheduled_queries`
### AJO — 旅程和活动
`list_journeys` · `get_journey` · `list_journey_versions` · `list_campaigns` · `get_campaign`
### AJO — 提供决策
`list_offers` · `get_offer` · `list_placements` · `list_collections` · `list_offer_activities` · `get_offer_activity` · `list_ranking_formulas` · `create_eligibility_rule`
### 流服务（源和目标）
`list_connections` · `get_connection` · `list_connection_specs` · `get_connection_spec` · `list_dataflows` · `get_dataflow` · `list_flow_runs` · `get_flow_run` · `list_source_connections` · `get_source_connection` · `list_target_connections` · `get_target_connection`
### 可观察性洞察和警报
`get_metrics` · `list_alert_subscriptions` · `get_alert_subscription` · `get_alerts_by_feature` · `list_alert_notifications` · `subscribe_alert`
### 数据卫生
`list_dataset_expirations` · `get_dataset_expiration` · `create_dataset_expiration` · `cancel_dataset_expiration` · `list_record_delete_orders` · `get_record_delete_order` · `create_record_delete_order` · `get_hygiene_quota`
### 数据准备（映射）
`list_mapping_sets` · `get_mapping_set` · `create_mapping_set` · `preview_mapping_set` · `list_mappings` · `get_mapping` · `validate_mapping_expression` · `list_data_prep_functions`
### 计算属性
`list_computed_attributes` · `get_computed_attribute` · `create_computed_attribute` · `update_computed_attribute`
### 访问控制（ABAC）
`list_roles` · `get_role` · `list_role_subjects` · `list_role_policies` · `list_policies` · `get_policy` · `get_effective_policies` · `list_available_permissions` · `list_users` · `get_user_roles`
### 客户旅程分析（CJA）
CJA 工具使用数据视图 ID 代替沙盒。
`cja_list_connections` · `cja_get_connection` · `cja_list_data_views` · `cja_get_data_view` · `cja_list_dimensions` · `cja_list_metrics` · `cja_list_projects` · `cja_get_project` · `cja_create_project` · `cja_update_project` · `cja_delete_project` · `cja_list_filters` · `cja_get_filter` · `cja_create_filter` · `cja_update_filter` · `cja_delete_filter` · `cja_list_calculated_metrics` · `cja_get_calculated_metric` · `cja_create_calculated_metric` · `cja_update_calculated_metric` · `cja_delete_calculated_metric` · `cja_list_annotations` · `cja_create_annotation` · `cja_delete_annotation` · `cja_run_report`
### 组织和多组织管理
`list_org_profiles` · `get_current_org` · `switch_org_profile` · `switch_sandbox`
---
## 服务器 2 — RT-CDP 代理 (`rtcdp_proxy.py`)
一个轻量级 stdio→HTTPS 代理，将工具调用转发到 **Adobe 的官方 Real-Time CDP MCP 服务器**，位于 `rtcdp-mcp.adobe.io`。身份验证头源自活动 `orgs.json` 配置文件，因此 `switch_org_profile` 跨两个服务器都有效。
通过 `.env` 配置上游 URL（可选 — 默认使用 Adobe 的端点）：
```
RTCDP_MCP_URL=https://rtcdp-mcp.adobe.io/mcp
```
> **注意：** Adobe 的 RT-CDP MCP 服务器是一个受限功能。请联系您的 Adobe 代表以获取访问权限。
### 工具（从 Adobe 代理）
`search_audiences` · `preview_audience_membership` · `inspect_audience_evaluation_jobs` · `inspect_audience_export_jobs` · `search_destination_connectors` · `search_destination_accounts` · `search_destination_flows` · `search_destination_input_connections` · `search_destination_output_connections` · `search_source_connectors` · `search_source_accounts` · `search_source_flows` · `search_source_input_connections` · `search_source_output_connections` · `search_identity_namespaces` · `search_merge_policies` · `search_organizations` · `inspect_flow_runs`
---
## 项目布局
```
server.py # AEP FastMCP 入口点
rtcdp_proxy.py # RT-CDP stdio→HTTPS 代理
auth.py # 身份验证 + HTTP 帮助程序；多组织配置文件状态
orgs.json # 您的凭证 — GITIGNORED，永远不要提交
orgs.example.json # orgs.json 模板
.env / .env.example # 备用单组织配置
requirements.txt
hooks/pre-commit # 阻止凭证提交
scripts/install_hooks.py # 安装预提交钩子
tools/
datasets.py # 目录服务：数据集、批次、沙盒
schemas.py # 架构注册表：XDM 架构、类、字段组
profiles.py # 实时客户资料、身份、合并策略
segments.py # 分段：定义、作业、流式传输
query.py # 查询服务：临时 SQL、模板、计划
ajo.py # AJO：旅程、活动、提供、资格规则
cja.py # 客户旅程分析
flows.py # 流服务：连接、数据流、流运行
observability.py # 可观察性洞察：指标、警报
data_hygiene.py # 数据卫生：数据集 TTL、记录删除
data_prep.py # 数据准备：映射集、表达式验证
computed_attributes.py # 计算属性
access_control.py # ABAC：角色、权限、有效策略
orgs.py # 多组织：列出/切换配置文件和沙盒
usage_logger.py # @track 装饰器 — 将工具调用记录到 CSV
```
---
## 添加新工具模块
1. 使用 `register(mcp)` 函数创建 `tools/new_domain.py`
2. 添加到 `server.py`：导入并调用 `new_domain.register(mcp)`
3. 在 `CLAUDE.md` 中遵循工具模式
## 身份验证流程
`auth.py` 首先检查 `orgs.json` 配置文件，然后回退到 `.env` 环境变量。
- **OAuth S2S**（推荐）：设置 `client_id` + `client_secret` — 令牌在过期前自动刷新
- **静态令牌**（开发/测试）：设置 `access_token` — 无自动刷新
## 要求
```
mcp[cli]>=1.3.0
httpx>=0.27.0
python-dotenv>=1.0.0
```
