MCP 自动化客户留存工作流，响应时间从天到分

在 Amazon Quick 中自动化客户留存工作流可以把五天的流失响应周期缩短到几分钟。上季度，一家中型 SaaS 公司因为留存团队需要花费五天来识别和联系不满意的客户，导致失去了 12% 的高风险账户。等到有人手动审查完 CSAT 电子表格和通话记录，这些客户已经流失了。Amazon Quick 将响应窗口从天级缩短到分钟级。

本文介绍如何在 Amazon Quick 中构建自动化留存流水线。流水线从结构化数据中检测不满意的客户，分析通话记录中的情感，按留存优先级对客户进行评分，然后生成针对各种情况定制的留存优惠。本文将教你：

配置 Quick Space，使其包含呼叫中心数据集和通话记录。

创建并注册自定义 MCP Action 用于客户评分。

构建 Chat Agent，结合定量的关键业绩指标 (KPI) 与定性的记录分析。

将分析转化为可复用的 Quick Flow。

使用 Amazon Quick Automate 协调完整流水线。

留存流水线按顺序连接四个 Amazon Quick 组件。Quick Dashboard 监控呼叫中心的 KPI，包括 CSAT（客户满意度评分，客户在每次通话后给出 1-5 的评分）、FCR（首次解决率，团队是否在单次通话中解决了问题）和 AHT（平均处理时间）。它识别 CSAT 分数为 2 或更低的客户。

Quick Chat Agent 使用自然语言查询结构化数据和非结构化的通话记录。它结合定量分数与定性的情感信号，显示客户为何处于风险中，而不仅仅是他们处于风险。

Amazon Quick Flows 将可重复的基于 Chat 的分析转化为计划的或按需的自动化，无需手动干预即可运行。其最后一步将结果格式化为下游自动化可以直接使用的结构化高风险客户列表。

Quick Automate 执行多步骤流水线。它获取 Flow 生成的结构化高风险客户列表，并通过自定义 MCP Action 对这些客户进行评分。MCP Action 是一个无服务器端点，用自定义业务逻辑扩展 Quick Automate。详见 Quick Automate MCP 开发者指南。流水线随后按优先级对客户排序、生成留存信函，并将其上传到 Amazon S3 以供分发。

你在 AWS 基础设施上运行工作流，可以访问审计日志、断点和基于角色的访问控制。

Automate 流水线架构

流水线执行五个顺序步骤。它从 Amazon S3 下载高风险客户列表和呼叫中心数据。评分步骤按留存优先级对这些客户进行排序（基于 CSAT 和每个客户最近一次问题发生的时间），并选出排名最靠前的两个高优先级案例。流水线随后起草参考每个客户具体问题的奖励补额信函，将其保存为 PDF，并将完成的信函上传到 Amazon S3 以供发送和存档。

下表总结了每个步骤：

此解决方案使用 Amazon S3、Amazon Quick（Dashboard、Chat、Flows、Automate）和自定义 MCP Action。

要完成本演练，你需要：

• 激活了 Amazon Quick 的 AWS 账户（Dashboard、Chat、Flows 和 Automate）

• CSV 格式的结构化呼叫中心数据集，命名为 contact_center_data.csv（列：customer_id、csat_score、call_date、call_reason、team_id）。本文附带示例数据集。

• 名为 Call_transcripts.docx 的通话记录文档（PDF 或 TXT 格式）。

• 用于流水线输入和输出的 Amazon S3 存储桶（例如：amzn-s3-demo-bucket）

以下部分从头到尾介绍构建过程，从准备 Quick Space 和注册 MCP Action 连接器，到组装 Chat Agent、Flow 和 Automate 工作流。请按顺序完成，因为每个步骤都建立在之前创建的组件基础上。

准备 Quick Space

Quick Space 是数据层，使 Chat Agent 能够访问结构化数据集（用于筛选和聚合）和非结构化文档（用于情感分析）。首先创建 Space 有助于验证每个下游组件都从相同的信息源中提取数据。

从 Quick 控制台选择 Spaces。

选择"Create space"。

对于 Space 名称，输入 ContactSpace。

在随后的屏幕上，选择 Datasets。

然后选择 Add datasets，搜索并勾选你的 contact_center_data 数据集，然后选择 Add。

选择 File Uploads，然后选择 Upload Files，上传你的 Call_transcripts.docx 记录文件，然后选择 Open。

等待索引完成。Status 列对数据集和文档都显示 Ready。

创建 MCP Action 连接器

MCP Action 连接器连接 Amazon Quick 和你的自定义业务逻辑。它定义连接器并使其作为 HTTP 端点可用。你编写包含评分算法的 AWS Lambda 函数（用 MCP 协议包装），通过 Amazon API Gateway 公开，然后在 Amazon Quick 中注册该端点。

Model Context Protocol (MCP) 是开放标准，允许 AI 智能体（如 Amazon Quick）以一致的请求-响应格式调用外部工具。如果你是这些服务的新用户，AWS Lambda 让你在无需配置服务器的情况下运行代码，Amazon API Gateway 将该代码公开为网址（HTTP 端点），Amazon Quick 可以调用。你首先创建 Lambda 函数，然后在其前面放置 API Gateway 端点，最后在 Amazon Quick 中注册该端点。

以下 Lambda 函数实现了最小 MCP 服务器。它处理 Amazon Quick 在连接期间调用的三个 JSON-RPC 方法：initialize（握手）、tools/list（它公告 score_customers 工具以便 Amazon Quick 可以发现它）和 tools/call（工具被调用时运行评分）。该工具的 inputSchema 遵循 JSON Schema Draft 7，这是 Amazon Quick 发布时所需的：

import json

# Tool catalog advertised to Amazon Quick during discovery (tools/list).
# inputSchema uses JSON Schema Draft 7 (required is an array at the root).
TOOLS = [
    {
        "name": "score_customers",
        "description": "Scores contact center customers by retention priority.",
        "inputSchema": {
            "type": "object",
            "properties": {
                "customers": {
                    "type": "array",
                    "description": "Customer records to score.",
                    "items": {
                        "type": "object",
                        "properties": {
                            "customer_id": {"type": "string"},
                            "csat_score": {"type": "number"},
                            "days_since_last_issue": {"type": "number"}
                        },
                        "required": ["customer_id"]
                    }
                }
            },
            "required": ["customers"]
        }
    }
]

def score_customers(customers):
    scored = []
    for c in customers:
        score = 0
        # Weight factors: low CSAT = high churn risk
        score += (5 - c.get("csat_score", 3)) * 20
        # Recency: issues in last 7 days score higher
        score += 30 if c.get("days_since_last_issue", 30) < 7 else 0
        scored.append({
            "customer_id": c["customer_id"],
            "retention_score": round(score, 1),
            "explanation": f"CSAT:{c.get('csat_score')}, days_since_last_issue:{c.get('days_since_last_issue')}"
        })
    scored.sort(key=lambda x: x["retention_score"], reverse=True)
    return scored

def lambda_handler(event, context):
    """MCP-compliant JSON-RPC handler for customer retention scoring."""
    req = json.loads(event.get("body", "{}"))
    method = req.get("method")
    req_id = req.get("id")

def rpc(result):
        return {"statusCode": 200,
                "headers": {"Content-Type": "application/json"},
                "body": json.dumps({"jsonrpc": "2.0", "id": req_id, "result": result})}

if method == "initialize":
        return rpc({
            "protocolVersion": "2024-11-05",
            "capabilities": {"tools": {}},
            "serverInfo": {"name": "customer-scoring", "version": "1.0.0"}
        })

if method == "tools/list":
        return rpc({"tools": TOOLS})

if method == "tools/call":
        params = req.get("params", {})
        if params.get("name") == "score_customers":
            args = params.get("arguments", {})
            scored = score_customers(args.get("customers", []))
            return rpc({"content": [{"type": "text", "text": json.dumps(scored)}]})

return {"statusCode": 200,
            "headers": {"Content-Type": "application/json"},
            "body": json.dumps({"jsonrpc": "2.0", "id": req_id,
                                "error": {"code": -32601, "message": "Method not found"}})}

部署函数到 AWS Lambda

打开 AWS Lambda 控制台并选择"Create function"。选择"Author from scratch"。

对于"Function name"，输入 customer-scoring。对于"Runtime"，选择最近支持的 Python 版本（例如，Python 3.13）。选择"Create function"。

在"Code"选项卡中，用之前显示的函数替换默认代码，然后选择"Deploy"。

函数部署后，使用以下 AWS 命令行界面 (AWS CLI) 命令在其前面创建一个 API Gateway 端点。如果您愿意，也可以在 API Gateway 控制台中执行相同的步骤。按顺序运行命令。每条命令都会产生一个值（如 API ID），供下一条命令使用。

将占位符值 (<API_ID>、<ROOT_RESOURCE_ID>、<REGION>、<ACCOUNT_ID>) 替换为您自己的值：

# 1. 创建 REST API
aws apigateway create-rest-api \
--name customer-scoring-api \
--endpoint-configuration types=REGIONAL

# 记下响应中的 "id" 值 -> <API_ID>

# 2. 获取根资源 ID
aws apigateway get-resources --rest-api-id <API_ID>

# 记下 "id" 值 -> <ROOT_RESOURCE_ID>

# 3. 创建 POST 方法
aws apigateway put-method \
--rest-api-id <API_ID> \
--resource-id <ROOT_RESOURCE_ID> \
--http-method POST \
--authorization-type NONE

# 4. 与 Lambda 集成
aws apigateway put-integration \
--rest-api-id <API_ID> \
--resource-id <ROOT_RESOURCE_ID> \
--http-method POST \
--type AWS_PROXY \
--integration-http-method POST \
--uri "arn:aws:apigateway:<REGION>:lambda:path/2015-03-31/functions/arn:aws:lambda:<REGION>:<ACCOUNT_ID>:function:<LAMBDA_FUNCTION_NAME>/invocations"

# 5. 允许 API Gateway 调用 Lambda
aws lambda add-permission \
--function-name <LAMBDA_FUNCTION_NAME> \
--statement-id apigateway-invoke \
--action lambda:InvokeFunction \
--principal apigateway.amazonaws.com \
--source-arn "arn:aws:execute-api:<REGION>:<ACCOUNT_ID>:<API_ID>/*/POST/"

# 6. 部署
aws apigateway create-deployment \
--rest-api-id <API_ID> \
--stage-name prod

# 您的端点 URL：
# https://<API_ID>.execute-api.<REGION>.amazonaws.com/prod/mcp

create-deployment 命令返回部署 ID 和时间戳。使用您的 API ID 和区域按以下格式自己构造端点 URL：https://<API_ID>.execute-api.<REGION>.amazonaws.com/prod/mcp。记下此 URL。在 Amazon Quick 中注册连接器时需要使用它。

生产考虑事项： 为简单起见，本演练直接在 Lambda 前面使用 API Gateway。对于生产工作负载，建议考虑在 Lambda 前面使用 Agentcore 网关，它提供开箱即用的代理级身份验证、限流和可观测性。

注册 MCP 连接器

在 Amazon Quick 首页，选择"More"，然后选择"Connectors"。

在右侧部分，为您的团队选择"Create"。

选择"Model Context Protocol"。

在接下来的屏幕上，选择"No, create new"。

输入以下值并选择"Next"：

• 名称： mcp-customer-score
• 描述： 根据聚合的关键绩效指标查找客户得分
• MCP 服务器端点： 粘贴端点 URL https://<API_ID>.execute-api.<REGION>.amazonaws.com/prod/mcp
• 连接类型： Public network

在"Authenticate"步骤中，选择"Service authentication"，然后打开"Auth configuration"列表并选择"None"。此端点故意处于开放状态用于演示目的。在生产环境中，您通常会添加身份验证层，最常见的是双腿 OAuth (2LO)，以保护端点。由于您部署的评分端点使用 –authorization-type NONE，它接受未经身份验证的请求，因此不需要客户端 ID 或密钥。（仅当 MCP 服务器受 OAuth 保护时才使用 OAuth 选项。）

选择"Create and continue"。

等待几分钟，然后在"Available"部分中选择新连接器。确认状态显示为"Ready"。

为联系中心分析创建自定义 Chat Agent

自定义 Chat Agent 连接到您的 ContactSpace，并可访问您注册的 MCP Action 连接器。系统提示是关键组件。它告诉代理如何解释数据、对记录提出哪些问题，以及何时调用评分工具而不是直接从数据集回答。

从 Quick 首页，选择"Chat agents"，然后选择"Blank"，再选择"Skip"从头创建自定义聊天代理。

对于"Agent name"，输入 Contact Center Strategy Analyst。

在"Knowledge Sources"下，链接 ContactSpace。

在"Actions"下，添加 mcp-customer-score 连接器。

对于系统提示，输入类似以下的文本：

You are a Contact Center Strategy Analyst. You have access to:
1. contact_center_data.csv - structured CSAT, call, and team data
2. Call_transcripts.docx - unstructured call transcript documents
3. mcp-customer-score tool - calculates retention priority scores
When asked about at-risk customers:
- Query the structured dataset for low CSAT scores
- Cross-reference with transcripts to identify sentiment and root causes
- When asked to score customers, call the mcp-customer-score tool
- Always explain WHY a customer is at risk, not just that they are
- Format results as tables when showing multiple customers

选择"Launch chat agent"。

使用 Chat Agent 分析高风险客户

Chat Agent 访问您的 Space 中的结构化数据集和非结构化记录。使用它来识别高风险客户并了解其满意度低的原因。

从 Amazon Quick 首页，与新的自定义聊天代理聊天。首先，从列表中选择"Contact Center Strategy Analyst"代理。

输入以下提示：

Show me customers that rated low CSAT (at or below 2) using contact center data

代理返回包含客户 ID、通话日期、CSAT 得分和相关指标的表格。这表明谁处于风险中，但不说明原因。

输入后续提示：

Using the call transcripts, summarize the main reasons these customers were dissatisfied

代理读取通话记录并总结其发现。例如，它可能会发现客户 4471 在三通通话中反复表达对计费错误的沮丧。客户 8832 就服务中断进行了一次激烈的互动。这一区别很重要。重复的小挫折需要与一次重大事件不同的留存方法。

为了生成下游自动化可使用的结构化结果，输入最后一个提示：

For the top 20 most at-risk customers, return only a JSON array where
each object has customer_id, csat_score, and days_since_last_issue
(calculated from the most recent call date). Do not include any other text.

代理返回高风险客户的结构化 JSON 列表。从聊天响应中复制此 JSON 输出，将其保存为名为 at_risk_customers.json 的文件，然后上传到您的 Amazon S3 存储桶（例如，上传到 inputs/ 前缀）。自动化在后面的步骤中从 Amazon S3 下载此文件。

将分析转换为 Quick Flow

除了一次性分析外，Quick Flows 允许您通过一个步骤将 Chat 对话转换为可重用的自动化，以便可以按需或按计划重新运行相同的分析。以下步骤演示如何将您刚刚进行的对话转换为 Flow。

在 Chat 界面中，选择工具栏上的 +，然后选择"Flow"。

选择"Generate new flow"，保留建议的提示并选择"Generate"。

查看流程，然后选择"Publish"。

选择"Run mode"，如果需要输入任何输入，然后启动流程。

Flow 现在是一个可重用的自动化。与您的团队分享或计划它每周运行一次，以便在需要时重新生成分析。

创建自动化工作流

所需连接器： 这些连接器应该是我们创建自动化的自动化组的一部分。

• Amazon S3（两个连接：一个用于源数据，一个用于输出上传）
• Custom Agent（使用在 MCP 操作连接器步骤中创建的 MCP 评分代理）

在 Amazon Quick 首页，选择"More"，然后选择"Automations"。

在接下来的页面中，选择"Groups"选项卡。

然后选择"Create Group"。

选择所需的操作，MCP 操作和 Amazon S3 操作。然后选择"Next"。

在接下来的页面中，选择"Done"。

上述组将用于创建自动化项目。按照说明创建自动化项目。

在 Amazon Quick 首页，选择"More"，然后选择"Automations"。

在接下来的屏幕上，选择"Create Project"，然后选择"Create Project"。

为自动化输入名称，选择"Automation group"，然后选择"Create"。

在接下来的屏幕上，选择"Start building"。

选择"Skip"。下面的屏幕出现。

对于后续步骤，请按照下面显示的说明完成自动化项目。

步骤 1：从 Amazon S3 加载通话数据

选择 New step，然后在左侧显示“What do you want to automate”的聊天助手中输入以下内容。此初始步骤会从 Amazon S3 中的原始联络中心 CSV 文件加载通话数据：

Log an info message: "Starting download of customer contact data from S3"
Download the file "contact_center_data_complete.csv" from the S3 bucket contactcenterbucketdemo".
Read the downloaded file as a CSV with headers included and comma delimiter. Store the result as call_data.
Log an info message: "Successfully loaded customer contact data from S3 CSV file"
Have all the above under one step

选择右上角的关闭（x）符号。

该步骤将创建为如下所示的形式。

步骤 2：筛选高风险客户

选择该步骤下方的添加（+）符号，然后选择 Process step。

将模式切换回规划模式。在聊天窗口中输入以下内容，然后选择对话符号。

Log an info message: "Starting to filter call data for known high-risk customers with negative sentiment"
Download the file "input/NegativeSentiment.docx" from the S3 bucket "contactcenterbucketdemo".
Have all the above under one step

步骤 3：聚合客户指标

按照上述操作创建步骤 3（Aggregate Customer Metrics），并使用以下指令：

Loop through each item in the aggregated metrics result table.
For each item, log an info message with this exact format:
"Customer: {item['customer_id']}, Total Calls: {item['total_calls']}, Avg CSAT: {item['avg_csat_score']}, FCR Rate: {item['first_call_resolution_rate']}%, Resolved Rate: {item['resolved_rate']}%, Avg Transfers: {item['avg_transfer_count']}, Avg Hold Time: {item['avg_hold_time_seconds']}s, Complaint Ratio: {item['complaint_ratio']}"
Have all the above under one step

这两条指令结合起来，将重新创建完整的步骤 3，包括其中用于日志记录的嵌套子步骤。主步骤负责所有计算逻辑，子步骤负责按客户输出日志。

步骤 4：识别排名最高的客户并生成信函

按照上述操作创建步骤 4（Identify Top customers and generate letter），并使用以下指令：

Log an info message: "Starting customer bonus letter generation process"
Collect all rows from customer_aggregated_metrics into a list called customer_records by looping through the table and appending each row.
Use an inline agent in "pro" mode with this instruction:
"Analyze the provided customer aggregated metrics data: {customer_records}. This data contains real customer records with actual customer IDs and performance scores. Identify and return the top 2 performing customers based on their aggregated scores from this actual data. Do not generate fictional customer IDs - use only the real customer IDs present in the provided customer records data. Return the actual customer information including the real customer IDs and their corresponding scores from the data."
Set the output schema to an object with a required property "top_customers" which is an array (minContains 2, maxContains 2) of objects. Each object has "customer_id" (string, required) and "score" (number).
Give the agent these tools: python_repl (coding tool) and the configured MCP tools for customer lookup.
Log the result: "top_customers are:" followed by the top_customers value.
Store the top_customers list. Create a helper function called generate_timestamp that returns the current datetime formatted as "%Y%m%d_%H%M%S". Call it and store as timestamp.
Create an empty list called bonus_letters.
For each customer in top_customers:
1. Get the customer_id from the customer object.
2. Create a PDF using create_pdf with the following HTML content:
   - Style: body with font-family Arial, line-height 1.6, margin 40px, color #333
   - A "header" div centered with h2 "Customer Retention Department" and italic paragraph "Valued Customer Appreciation Program"
   - A "content" div containing:
   - Bold greeting: "Dear Valued Customer (ID: {customer_id}),"
   - Paragraph expressing appreciation for continued partnership
   - Paragraph acknowledging past service issues and offering heartfelt apologies, mentioning their feedback has been invaluable
   - A "highlight" div (background-color #f0f8ff, padding 15px, border-left 4px solid #007acc, margin 20px 0) containing bold text: "As a gesture of our commitment to your satisfaction and our appreciation for your loyalty, we are pleased to offer you a $100 bonus credit."
   - Paragraph about the bonus reflecting genuine desire to make things right and mutual success
   - Paragraph about enhanced service protocols and dedicated customer success team
   - Paragraph hoping for opportunity to demonstrate renewed commitment
   - A "signature" div with margin-top 40px containing:
   - "With sincere regards and appreciation,"
   - A line break
   - Bold "Customer Retention Team", then "Customer Success Department", then italic "Dedicated to Your Success"
3. Upload the PDF to S3 bucket "contactcenterbucketdemo" with Key = "output/" + customer_id + timestamp + ".pdf"
4. Append the PDF to the bonus_letters list.
After the loop completes, log: "Successfully generated bonus letters for top 2 customers"
Have all the above under one step

测试并部署自动化流程

对自动化流程进行端到端验证，然后提交并部署，使其能够按计划运行。

在画布中选择 Debug（或 Test）以运行自动化流程。它会自行运行，通过 Download file 步骤从 Amazon S3 下载高风险客户文件。在日志面板中检查每个步骤的输出。

确认评分步骤返回排名最高的两位客户、信函引用了正确的客户问题，并且 S3 上传成功完成。

运行结果正确后，选择 Commit 创建一个版本。

选择 Deploy，选中已提交的版本，并确认所需的凭证和连接（您的 Amazon S3 连接以及 mcp-customer-score 连接器）。

要按计划运行自动化流程，请在 Deployment 页面上选择 Create Trigger，然后设置运行频率（或 cron 表达式）、开始时间和时区。

有关更多详细信息，请参阅