AEP MCP 服务器

两个 MCP 服务器，让 Claude（以及任何兼容 MCP 的客户端）能够以自然语言访问 Adobe Experience Platform、Adobe Journey Optimizer 和 Real-Time CDP。

快速开始

1. 运行设置脚本

./setup.sh

这将创建一个 .venv，安装所有依赖项，安装 pre-commit 钩子，并打印出要粘贴的确切 MCP 配置 JSON——包括您机器上正确的绝对 venv Python 路径。

Windows： 手动运行这些步骤——python3 -m venv .venv，.venv\Scripts\activate，pip install -r requirements.txt。

2. 配置凭据

从 Adobe Developer Console → 项目 → OAuth 服务器到服务器 填写您的凭据：

# orgs.json was created by setup.sh — edit it now

或者使用环境变量（单组织回退）：

cp .env.example .env
# edit .env

3. 注册到 Claude Code

将 setup.sh 打印的 JSON 粘贴到 .claude/mcp_servers.json 中，例如：

{
  "aep": {
    "command": "/path/to/aep-mcp-server/.venv/bin/python",
    "args": ["/path/to/aep-mcp-server/server.py"]
  }
}

始终使用 venv Python 路径——MCP 客户端在任何活动 shell 之外启动服务器，因此裸的 python3 将找不到已安装的包。

4. 运行

python3 server.py                        # AEP server, default org profile
AEP_PROFILE=MyClient python3 server.py  # start with a specific profile

Related MCP server: Fourth Brain Demo

多组织配置

orgs.json 支持多个命名配置文件——在跨不同 Adobe 组织或沙盒工作时非常有用：

{
  "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 构建。所有工具都接受可选的 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 entrypoint
rtcdp_proxy.py             # RT-CDP stdio→HTTPS proxy
auth.py                    # Auth + HTTP helpers; multi-org profile state
orgs.json                  # Your credentials — GITIGNORED, never commit
orgs.example.json          # Template for orgs.json
.env / .env.example        # Fallback single-org config
requirements.txt
hooks/pre-commit           # Blocks credential commits
scripts/install_hooks.py   # Installs the pre-commit hook
tools/
  datasets.py              # Catalog Service: datasets, batches, sandboxes
  schemas.py               # Schema Registry: XDM schemas, classes, field groups
  profiles.py              # Real-Time Customer Profile, identity, merge policies
  segments.py              # Segmentation: definitions, jobs, streaming
  query.py                 # Query Service: ad-hoc SQL, templates, scheduled
  ajo.py                   # AJO: journeys, campaigns, offers, eligibility rules
  cja.py                   # Customer Journey Analytics
  flows.py                 # Flow Service: connections, dataflows, flow runs
  observability.py         # Observability Insights: metrics, alerts
  data_hygiene.py          # Data Hygiene: dataset TTL, record deletes
  data_prep.py             # Data Prep: mapping sets, expression validation
  computed_attributes.py   # Computed Attributes
  access_control.py        # ABAC: roles, permissions, effective policies
  orgs.py                  # Multi-org: list/switch profiles and sandboxes
  usage_logger.py          # @track decorator — logs tool calls to CSV

添加新工具模块

1. 创建 tools/new_domain.py，包含一个 register(mcp) 函数

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