1. 从自然语言到 Redshift 结果集：Agentic Analytics 到底解决什么问题

电商和游戏行业的数据团队经常遇到一个尴尬场景：运营同学想看"上周充值事件按小时的分布，并预测未来两周趋势"，这句话本身不复杂，但要落地成 Redshift 查询，得先找表、确认字段类型、写 SQL、跑一遍看结果对不对，再整理成报告。整个过程里，真正花在"分析"上的时间可能不到三成，剩下七成都在做数据搬运和 SQL 调试。

Agentic Analytics 想解决的就是这段搬运成本。它的核心思路是：让一个大语言模型驱动的智能体，自己决定"先看有哪些集群、再看有哪些库、再看表结构、最后生成并执行 SQL"，把原本需要人肉完成的元数据探索和 SQL 迭代，交给 Agent 自动跑完。你只需要用自然语言描述业务问题，Agent 负责把它翻译成可执行、可验证的查询链路。

这套链路要跑通，需要三个角色配合。第一个是Amazon Redshift MCP Server，它把 Redshift 的 Data API 封装成一组标准工具（list_clusters、list_databases、list_schemas、list_tables、list_columns、execute_query），让 Agent 能像调用函数一样操作数据仓库，而不需要自己处理连接池、凭证和 SQL 注入防护。第二个是Strands Agents SDK，它负责把模型、工具和系统提示词组装成一个可推理的 Agent，内置了工具调用循环，模型决定调哪个工具、传什么参数，SDK 负责执行并把结果喂回模型。第三个是Amazon Bedrock AgentCore Runtime，它提供一个无服务器的托管运行环境，把 Agent 打包成容器、推到 ECR、部署成可调用的 Runtime 端点，你不需要自己维护 ECS 或 Lambda 的冷启动配置。

适合谁跟做？如果你已经有一个 Redshift 集群（或者 Serverless 工作组），手上有 AWS 账号，会用 Python 写几十行代码，那这篇的步骤可以直接复现。如果你只是想理解 Agentic Analytics 的架构长什么样，也可以只看第 2 节和第 3 节的配置片段，把 MCP Server 的连接参数和 Agent 的入口逻辑搞清楚。

我试过把这套链路跑在一个模拟的游戏充值事件表上，从"帮我总结 charge_events 的事件情况并预测未来两周趋势"这句话开始，Agent 自动完成了 list_clusters → list_databases → list_schemas → list_tables → list_columns → execute_query 六步工具调用，最后输出了一份带周度趋势、Top 5 高峰时段和业务建议的 Markdown 报告。下面把每一步的配置和踩坑点拆开讲。

2. TaoToken 统一 Key 与 Redshift MCP Server 的前置配置

在动手写 Agent 之前，先把两个前置条件准备好：一个是模型侧的调用凭证，一个是 Redshift 侧的访问权限。这两块如果没配好，后面 Agent 跑起来会在工具调用阶段直接报错，而且报错信息往往不直观，容易让人以为是代码问题。

2.1 用 TaoToken 统一 Key 管理模型调用

Strands Agents SDK 支持多种模型提供商，包括 Amazon Bedrock、Anthropic、Ollama 以及 OpenAI 兼容接口。如果你希望用一个统一的 Key 来管理模型调用，而不是在每个环境里分别配置 Bedrock 的 IAM 凭证，可以用 TaoToken 的 OpenAI 兼容接口。它的 Base URL 是https://taotoken.net/api，你需要在 TaoToken 控制台生成一个 API Key，然后在 Strands 的模型配置里指向这个端点。

具体来说，Strands 的Agent初始化时，model参数可以传一个模型 ID 字符串（走 Bedrock），也可以传一个自定义的模型客户端。如果你走 OpenAI 兼容路径，需要先安装strands-agents和对应的 provider 包，然后在代码里这样配置：

import os from strands import Agent from strands.models.openai import OpenAIModel os.environ["OPENAI_API_KEY"] = "你的 TaoToken API Key" os.environ["OPENAI_BASE_URL"] = "https://taotoken.net/api" model = OpenAIModel( model_id="claude-3-7-sonnet", client_args={ "api_key": os.environ["OPENAI_API_KEY"], "base_url": os.environ["OPENAI_BASE_URL"], } ) agent = Agent(model=model, system_prompt="你是 Redshift 数据分析助手")

这里的关键点是base_url必须指向https://taotoken.net/api，不要带多余的路径后缀。API Key 建议放在环境变量里，不要硬编码进strands_agent.py，因为后面部署到 AgentCore Runtime 时，容器环境变量是更安全的注入方式。

如果你更习惯用 Bedrock 原生路径，也可以继续用model_id="us.anthropic.claude-3-7-sonnet-20250219-v1:0"这种写法，前提是 AgentCore Runtime 的执行角色有bedrock:InvokeModel权限。两种方式不冲突，你可以根据团队现有的凭证管理体系来选。

2.2 Redshift MCP Server 的连接参数

Redshift MCP Server 通过uvx启动，命令是uvx awslabs.redshift-mcp-server@latest。它依赖 AWS 凭证来调用 Redshift Data API，所以你需要确保运行环境里有可用的 AWS 凭证（环境变量、IAM 角色或~/.aws/credentials都行）。在 Strands Agent 里，MCP Client 的配置长这样：

from strands.tools.mcp import MCPClient from mcp import stdio_client, StdioServerParameters redshift_mcp_client = MCPClient( lambda: stdio_client( StdioServerParameters( command="uvx", args=["awslabs.redshift-mcp-server@latest"], env={ "AWS_DEFAULT_REGION": "us-west-2", "AWS_REGION": "us-west-2", } ) ) )

注意AWS_DEFAULT_REGION和AWS_REGION要和你 Redshift 集群所在区域一致。如果你的集群是 Serverless 工作组，list_clusters工具也能识别，但cluster_identifier要填工作组名称，而不是集群 ID。

2.3 表权限初始化：别跳过这一步

Redshift Data API 执行查询时，用的是调用者的 IAM 身份，但表级别的 SELECT 权限仍然需要在数据库里显式授予。如果你直接跑execute_query，很可能会遇到permission denied for relation xxx这类错误。所以在 Agent 启动时，最好先跑一段权限初始化逻辑，把需要访问的表授权给当前用户或 PUBLIC。

async def initialize_table_permissions(mcp_client, cluster_id, database_name, tables): for table in tables: grant_sql = f"GRANT SELECT ON TABLE {table} TO PUBLIC;" try: await mcp_client.call_tool_async( "execute_query", { "cluster_identifier": cluster_id, "database_name": database_name, "sql": grant_sql } ) except Exception as e: print(f"授权 {table} 失败: {e}") continue

这段代码在@app.entrypoint里调用一次即可。生产环境里不建议用TO PUBLIC，应该授权给具体的 IAM 映射用户或角色，但演示阶段用 PUBLIC 能快速跑通。

2.4 AgentCore Runtime 的执行角色权限

AgentCore Runtime 在部署时会自动创建一个执行角色（如果你设置auto_create_execution_role=True）。这个角色默认只有基本的日志权限，你需要手动给它加上 Redshift Data API 的权限，否则 MCP Server 调用list_clusters时会报AccessDenied。需要附加的策略至少包括：

{ "Version": "2012-10-17", "Statement": [ { "Effect": "Allow", "Action": [ "redshift:DescribeClusters", "redshift-data:ListDatabases", "redshift-data:ListSchemas", "redshift-data:ListTables", "redshift-data:DescribeTable", "redshift-data:ExecuteStatement", "redshift-data:DescribeStatement", "redshift-data:GetStatementResult" ], "Resource": "*" } ] }

如果你用的是 Redshift Serverless，还需要加上redshift-serverless:GetWorkgroup和redshift-serverless:ListWorkgroups。这些权限加在 AgentCore Runtime 自动创建的角色上，而不是你本地的 IAM 用户上，因为实际执行查询的是 Runtime 容器里的角色。

3. 可复制配置：strands_agent.py 与 deploy.py 的完整片段

这一节给出可以直接复制运行的代码。整个项目只有四个文件：strands_agent.py（Agent 主逻辑）、deploy.py（部署脚本）、test_client.py（测试客户端）、requirements.txt（依赖）。先看依赖：

strands-agents strands-agents-tools bedrock-agentcore bedrock-agentcore-starter-toolkit aws-opentelemetry-distro>=0.10.0 mcp

3.1 strands_agent.py：入口点与工具加载

AgentCore Runtime 通过@app.entrypoint装饰器识别请求入口。容器启动后，Runtime 会把用户的 payload 传给这个函数，函数返回的内容就是 Agent 的响应。核心逻辑是：在 entrypoint 里创建 MCP Client、加载 Redshift 工具、初始化表权限、构造 Agent、执行用户输入。

#!/usr/bin/env python3 """Strands Agent with Redshift MCP Tools for AgentCore Runtime""" from strands import Agent from strands.tools.mcp import MCPClient from mcp import stdio_client, StdioServerParameters from bedrock_agentcore.runtime import BedrockAgentCoreApp app = BedrockAgentCoreApp() AWS_REGION = "us-west-2" DATABASE_NAME = "testdb" CLUSTER_ID = "test-workgroup" TABLES = [ "public.activity_events", "public.charge_events", "public.fight_events", "public.login_logout_events" ] MODEL_ID = "us.anthropic.claude-3-7-sonnet-20250219-v1:0" MCP_COMMAND = "uvx" MCP_ARGS = ["awslabs.redshift-mcp-server@latest"] @app.entrypoint async def strands_agent_bedrock(payload, context): try: redshift_mcp_client = MCPClient( lambda: stdio_client( StdioServerParameters( command=MCP_COMMAND, args=MCP_ARGS, env={ "AWS_DEFAULT_REGION": AWS_REGION, "AWS_REGION": AWS_REGION } ) ) ) with redshift_mcp_client: redshift_tools = redshift_mcp_client.list_tools_sync() agent = Agent( model=MODEL_ID, system_prompt=SYSTEM_PROMPT, tools=redshift_tools, ) user_input = payload.get("prompt", "No prompt found") response = agent(user_input) return response except Exception as e: return f"Agent 执行错误: {str(e)}" if __name__ == "__main__": app.run()

SYSTEM_PROMPT是控制 Agent 行为的关键。它需要明确几件事：只执行 SELECT、每个查询必须带 LIMIT、查询失败先 ROLLBACK 再重试、输出用中文 Markdown。下面这段可以直接用：

SYSTEM_PROMPT = """你是一位专业的 AWS Redshift 数据分析师助手。 ## SQL 执行安全规范 - 仅执行 SELECT 查询，严禁 INSERT、UPDATE、DELETE、CREATE、DROP 等写操作 - 每个查询必须包含 LIMIT 子句，避免返回过大结果集 - 查询前必须验证表名和字段名的存在性 - 如果查询失败，必须先执行 ROLLBACK 或 COMMIT 结束当前事务，再重新开始新查询 - 避免在字符串字段上使用日期函数，需要先进行类型转换 ## 输出要求 - 全程使用中文回复 - 以 Markdown 格式组织内容，包含清晰的标题层级 - 内容结构：数据概览与质量评估、详细分析过程和思维逻辑、关键发现和数据洞察、业务建议和行动建议 """

3.2 deploy.py：一键部署到 AgentCore Runtime

部署脚本用bedrock_agentcore_starter_toolkit的Runtime类。configure阶段会解析 entrypoint、生成.bedrock_agentcore.yaml、Dockerfile 和.dockerignore，并在云端创建 ECR 仓库和 CodeBuild 项目。launch阶段会构建镜像、推送到 ECR、部署 Runtime。

#!/usr/bin/env python3 from bedrock_agentcore_starter_toolkit import Runtime import time def deploy(): region = "us-west-2" agentcore_runtime = Runtime() agentcore_runtime.configure( entrypoint="strands_agent.py", auto_create_execution_role=True, auto_create_ecr=True, requirements_file="requirements.txt", region=region, agent_name="redshift-analytics-agent" ) launch_result = agentcore_runtime.launch() status_response = agentcore_runtime.status() status = status_response.endpoint["status"] end_status = ["READY", "CREATE_FAILED", "DELETE_FAILED", "UPDATE_FAILED"] while status not in end_status: print(f"状态: {status} - 等待中...") time.sleep(10) status_response = agentcore_runtime.status() status = status_response.endpoint["status"] if status == "READY": return { "region": region, "agent_arn": launch_result.agent_arn, } return None if __name__ == "__main__": result = deploy() if result: print(f"Agent ARN: {result['agent_arn']}")

agent_name要替换成你自己的名称，不能和已有 Runtime 重名。部署完成后，agent_arn是后面测试客户端要用的关键参数。

3.3 test_client.py：端到端调用

测试客户端用boto3的bedrock-agentcore客户端调用 Runtime。注意read_timeout要设大一点（比如 300 秒），因为 Agent 要跑多轮工具调用，响应时间可能超过默认的 60 秒。

#!/usr/bin/env python3 import boto3 import json import uuid def test_strands_agent(): agent_runtime_arn = "你的 Agent ARN" session_id = str(uuid.uuid4()) client = boto3.client( "bedrock-agentcore", region_name="us-west-2", config=boto3.session.Config(read_timeout=300, connect_timeout=60) ) PROMPT = "帮我总结 testdb 中 charge_events 的事件情况，并根据历史趋势分析未来两周用户可能的事件趋势" response = client.invoke_agent_runtime( agentRuntimeArn=agent_runtime_arn, qualifier="DEFAULT", runtimeUserId="123", runtimeSessionId=session_id, payload=json.dumps({"prompt": PROMPT}) ) all_data = "" if "text/event-stream" in response.get("contentType", ""): for line in response["response"].iter_lines(chunk_size=1024): if line: all_data += line.decode("utf-8", errors="ignore") + "\n" else: for event in response.get("response", []): all_data += event.decode("utf-8", errors="ignore") + "\n" print(all_data) if __name__ == "__main__": test_strands_agent()

runtimeSessionId用 UUID 生成，同一个 session 内的多轮对话会共享上下文。如果你要做多轮追问，复用同一个session_id即可。

4. 验证请求：一次端到端查询的成功结果长什么样

配置写完之后，跑一次完整链路，看看 Agent 到底做了什么。这一步很重要，因为 Agentic Analytics 的价值不在于"能跑通"，而在于"跑得对、跑得可解释"。

4.1 执行部署与调用

先跑python deploy.py，等待状态变成READY，拿到agent_arn。然后把agent_arn填进test_client.py，执行python test_client.py。如果一切正常，你会看到 Agent 的输出流式返回，内容是一份 Markdown 格式的分析报告。

4.2 Agent 的工具调用链路

从 AgentCore Runtime 的日志里，可以清楚看到 Agent 依次调用了这些工具：

第一步是list_clusters，Agent 扫描当前账号下所有可用的 Redshift 集群和 Serverless 工作组，确认test-workgroup存在且状态为available。这一步的输出里包含集群的 endpoint、数据库名称和 IAM 角色信息。

第二步是list_databases，连接到test-workgroup，查询系统视图，发现testdb数据库可访问。

第三步是list_schemas，在testdb里列出所有 schema，确认publicschema 存在。

第四步是list_tables和list_columns，Agent 在publicschema 下找到charge_events表，并查看它的字段结构，确认有event_time、user_id、amount这些关键字段。

第五步是execute_query，Agent 根据表结构生成 SQL，比如按周聚合充值事件数和总金额：

SELECT DATE_TRUNC('week', event_time) AS week_start, COUNT(DISTINCT user_id) AS user_count, COUNT(*) AS event_count, SUM(amount) AS total_amount FROM public.charge_events WHERE event_time >= DATEADD(week, -4, CURRENT_DATE) GROUP BY 1 ORDER BY 1 LIMIT 100;

执行后返回四行结果，对应四周的充值数据。Agent 拿到结果后，又生成了一条按小时聚合的查询，找出充值高峰时段，最后综合两份结果输出分析报告。

4.3 成功结果的判断标准

一次成功的端到端查询，应该满足三个条件。第一，Agent 输出的报告里包含具体的数字，比如"总周数 4 周""总充值金额 2787 万"，而不是泛泛而谈。第二，报告里能看到 SQL 执行痕迹，比如 Agent 会说明"我查询了 charge_events 表，按周聚合后发现……"。第三，如果某一步查询失败，Agent 应该能自己调整 SQL 重试，而不是直接报错退出。

我实测下来，从发出 prompt 到收到完整报告，耗时大约 40 到 60 秒，其中大部分时间花在模型推理和多轮工具调用上。如果你觉得太慢，可以精简 system prompt，或者把list_columns的结果缓存起来，减少重复的元数据查询。

5. 本篇常见错误排查：401、local proxy failed 与 reading choices

即使配置看起来没问题，实际跑的时候还是会遇到各种报错。这一节把最常见的几类错误和排查路径列出来，方便你对照日志定位。

5.1 401 Unauthorized：模型调用凭证问题

如果你在 Agent 日志里看到401 Unauthorized或AuthenticationError，大概率是模型侧的 Key 没配对。分两种情况：走 Bedrock 原生路径时，检查 AgentCore Runtime 的执行角色是否有bedrock:InvokeModel权限，以及模型 ID 是否在目标区域可用（比如us.anthropic.claude-3-7-sonnet-20250219-v1:0需要跨区域推理配置）。走 TaoToken 兼容路径时，检查OPENAI_API_KEY和OPENAI_BASE_URL是否都设置正确，Base URL 必须是https://taotoken.net/api，不要多加/v1或斜杠。

5.2 local proxy failed：MCP Server 启动失败

local proxy failed或MCPClient connection error通常意味着uvx命令在容器里跑不起来。可能的原因有三个：容器镜像里没有安装uv（需要在 Dockerfile 里加pip install uv）、awslabs.redshift-mcp-server包下载超时（可以换成固定版本号，比如awslabs.redshift-mcp-server@0.1.0）、或者环境变量AWS_REGION没传进 MCP Server 的子进程。排查方法是先在本地跑uvx awslabs.redshift-mcp-server@latest，确认能启动，再检查 Dockerfile 的依赖安装步骤。

5.3 reading choices：响应解析失败

reading choices或Error reading choices from response一般出现在模型返回格式不符合预期时。Strands SDK 期望模型返回标准的 chat completion 格式，如果你用的兼容接口返回了非标准结构，就会解析失败。解决办法是确认 TaoToken 的接口版本和模型 ID 匹配，比如claude-3-7-sonnet对应的模型 ID 要写对，不要用gpt-4这种不存在的组合。另外，如果响应里包含大量工具调用结果，可能会超出模型的 context window，导致返回被截断，这时候需要精简 system prompt 或减少一次性加载的工具数量。

5.4 OAuth 与权限相关报错

如果你看到OAuth或AccessDenied相关的错误，先确认 AgentCore Runtime 的执行角色是否附加了 Redshift Data API 权限。另一个常见坑是 Redshift 集群的 IAM 角色和 Data API 调用者不是同一个身份，导致list_clusters能返回集群信息，但execute_query时报permission denied。这时候需要在 Redshift 里执行GRANT SELECT ON TABLE xxx TO IAM_ROLE 'arn:aws:iam::xxx:role/xxx'，把权限授予 Runtime 的执行角色。

5.5 工具调用三件套检查清单

如果你用的是 Cline MCP、CC Switch 或 Codex 这类客户端来调试 MCP Server，记得检查三件套是否齐全：Base URL（MCP Server 的启动命令和参数）、Key（AWS 凭证或 TaoToken API Key）、Model ID（模型标识）。缺任何一个，工具调用都会失败。特别是 Model ID，在 Strands 里是model_id参数，在 MCP 客户端配置里可能是model字段，名称不统一，容易漏配。

6. 把 Agentic Analytics 接入你的日常工作流

跑通一次端到端查询之后，下一步是把它变成可复用的能力。最直接的方式是把 AgentCore Runtime 的 ARN 封装成一个内部 API，让运营或产品同学通过一个简单的 Web 界面提交自然语言问题，后端调用 Runtime 并返回报告。这样他们不需要懂 SQL，也不需要接触 Redshift 控制台。

如果你希望长期跑编码类或 Agent 类任务，可以关注 TaoToken 的 Coding Plan，它提供了更适合持续调用的额度方案。如果你只是想先验证模型对话效果，可以直接在模型对话页面测试 prompt。接入文档里有 MCP Server 和 AgentCore Runtime 的详细参数说明，遇到配置问题时可以对照排查。API Key 的管理在控制台的 API Keys 页面，建议为每个环境生成独立的 Key，方便审计和轮换。

一个实用的技巧是：在 system prompt 里加一句"如果查询结果少于 10 行，直接展示原始数据；如果超过 10 行，先做聚合再展示"。这样 Agent 在面对大表时不会返回一堆原始行，而是自动做 summary，报告的可读性会好很多。另一个技巧是把常用的表结构写进 system prompt 的注释里，减少 Agent 调用list_columns的次数，能明显缩短响应时间。