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_idclaude-3-7-sonnet, client_args{ api_key: os.environ[OPENAI_API_KEY], base_url: os.environ[OPENAI_BASE_URL], } ) agent Agent(modelmodel, system_prompt你是 Redshift 数据分析助手)这里的关键点是base_url必须指向https://taotoken.net/api不要带多余的路径后缀。API Key 建议放在环境变量里不要硬编码进strands_agent.py因为后面部署到 AgentCore Runtime 时容器环境变量是更安全的注入方式。如果你更习惯用 Bedrock 原生路径也可以继续用model_idus.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-serverlatest。它依赖 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( commanduvx, args[awslabs.redshift-mcp-serverlatest], 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 fGRANT 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_roleTrue。这个角色默认只有基本的日志权限你需要手动给它加上 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.pyAgent 主逻辑、deploy.py部署脚本、test_client.py测试客户端、requirements.txt依赖。先看依赖strands-agents strands-agents-tools bedrock-agentcore bedrock-agentcore-starter-toolkit aws-opentelemetry-distro0.10.0 mcp3.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-serverlatest] app.entrypoint async def strands_agent_bedrock(payload, context): try: redshift_mcp_client MCPClient( lambda: stdio_client( StdioServerParameters( commandMCP_COMMAND, argsMCP_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( modelMODEL_ID, system_promptSYSTEM_PROMPT, toolsredshift_tools, ) user_input payload.get(prompt, No prompt found) response agent(user_input) return response except Exception as e: return fAgent 执行错误: {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( entrypointstrands_agent.py, auto_create_execution_roleTrue, auto_create_ecrTrue, requirements_filerequirements.txt, regionregion, agent_nameredshift-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(fAgent 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_nameus-west-2, configboto3.session.Config(read_timeout300, connect_timeout60) ) PROMPT 帮我总结 testdb 中 charge_events 的事件情况并根据历史趋势分析未来两周用户可能的事件趋势 response client.invoke_agent_runtime( agentRuntimeArnagent_runtime_arn, qualifierDEFAULT, runtimeUserId123, runtimeSessionIdsession_id, payloadjson.dumps({prompt: PROMPT}) ) all_data if text/event-stream in response.get(contentType, ): for line in response[response].iter_lines(chunk_size1024): if line: all_data line.decode(utf-8, errorsignore) \n else: for event in response.get(response, []): all_data event.decode(utf-8, errorsignore) \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_clustersAgent 扫描当前账号下所有可用的 Redshift 集群和 Serverless 工作组确认test-workgroup存在且状态为available。这一步的输出里包含集群的 endpoint、数据库名称和 IAM 角色信息。第二步是list_databases连接到test-workgroup查询系统视图发现testdb数据库可访问。第三步是list_schemas在testdb里列出所有 schema确认publicschema 存在。第四步是list_tables和list_columnsAgent 在publicschema 下找到charge_events表并查看它的字段结构确认有event_time、user_id、amount这些关键字段。第五步是execute_queryAgent 根据表结构生成 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 failedMCP Server 启动失败local proxy failed或MCPClient connection error通常意味着uvx命令在容器里跑不起来。可能的原因有三个容器镜像里没有安装uv需要在 Dockerfile 里加pip install uv、awslabs.redshift-mcp-server包下载超时可以换成固定版本号比如awslabs.redshift-mcp-server0.1.0、或者环境变量AWS_REGION没传进 MCP Server 的子进程。排查方法是先在本地跑uvx awslabs.redshift-mcp-serverlatest确认能启动再检查 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 URLMCP Server 的启动命令和参数、KeyAWS 凭证或 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的次数能明显缩短响应时间。