Prefer a visual version? Open the responsive HTML README.
Results preview. These are the actual frontend UI screenshots from the analytics session.
PDF preview. GitHub README files do not auto-embed PDFs inline, so this image opens the full downloaded report: edp-report-q7.pdf.
This repository is part of the Enterprise Data Platform. For the full project overview, architecture diagram, and build order, start there.
Previous: platform-orchestration-mwaa-airflow: the Airflow DAG on MWAA orchestrates the pipeline that produces the Gold tables this agent queries.
This is the Natural Language (NL) Analytics Agent for the Enterprise Data Platform. It's the final layer of the platform: everything before this (DMS (Database Migration Service), Glue, dbt, MWAA (Amazon Managed Workflows for Apache Airflow)) exists to produce a clean, curated Gold data layer. This agent makes that data accessible to anyone who can ask a question in business terms, without needing to know SQL, table names, or partition structures.
The Gold layer holds carefully curated, business-ready aggregations. Getting value from it still requires an analyst who can write Athena SQL, knows the exact table and column names, and understands the partition structure well enough not to run expensive full-table scans. Most people at a company can't do all three. This agent removes that barrier.
A user asks: "Show me monthly transaction volume for Berlin over the last 12 months."
The agent:
- Classifies the question (analytical query, conversational follow-up, or chart retype)
- Identifies the correct Gold table from the Glue Catalog and dbt schema catalog
- Generates an Athena SQL query in a single Claude call with the full schema already in context
- Validates the SQL against hard guardrails (SELECT only, Gold DB only, LIMIT required, no DDL)
- Checks that the SQL actually answers the question before running it: reads the SQL blind, infers its intent, and regenerates once with correction feedback if there's a mismatch
- Runs the query via Athena and validates the result for obvious anomalies
- Produces a chart matched to the data shape (line, bar, scatter, pie, or multi-line)
- Returns a business-language insight alongside the SQL it ran, every assumption it made, the scan cost, and an intent verdict
If it interpreted "transactions" as completed orders only, it says so explicitly before returning the result, so the user can catch that interpretation and correct it.
Amazon Athena is a serverless SQL (Structured Query Language) query engine that runs directly over S3 (Simple Storage Service) data. The cost model is pay-per-byte-scanned, not per compute hour. A query that scans the whole table because a partition filter is missing doesn't just run slowly; it costs real money and could easily hit the WorkGroup scan limit.
This is different from Databricks, BigQuery (Google's managed data warehouse), or Snowflake, which are managed warehouses with internal storage. Those platforms already have built-in NL (Natural Language) query features. Athena doesn't. Nobody ships an NL-to-SQL product that reasons about S3 partition structures and Glue Catalog metadata for cost optimisation. That's what this agent does.
The schema context is also richer here than in most text-to-SQL systems. The agent reads from two sources simultaneously:
- Glue Catalog (live): column names, data types, partition keys. Always current.
- dbt catalog.json (from S3): column descriptions, model documentation, accepted values, lineage. Written after every successful pipeline run by the MWAA DAG's
upload_dbt_artifactstask.
Most NL-to-SQL tools only see column names. This agent sees the business meaning behind every column.
flowchart TD
subgraph Startup ["Startup - once per ECS task"]
LoadSchemas["load_all_schemas()\nGlue Catalog + dbt catalog.json\nAll 7 Gold schemas into system prompt"]
end
subgraph Input ["Input"]
User([User NL Question\ne.g. via CLI or HTTP])
Streamlit[Streamlit UI\nBrowser interface]
User --> Streamlit
end
subgraph AgentLoop ["Agent Reasoning Loop - ECS Fargate"]
direction TB
Classify["classify_question()\nanalytical / conversational / retype"]
GenerateSQL["Call 1: Generate SQL + Assumptions\nClaude Platform on AWS\nSchema already in system prompt"]
ValidateSQL["validate_sql()\nsqlparse guardrails"]
IntentCheck["Call 2: Infer SQL intent\nClaude Platform on AWS - SQL only\nquestion withheld except language hint"]
VerdictCheck["Call 3: Compare original question\nvs inferred SQL intent\nClaude Platform on AWS verdict"]
Verdict{"Verdict says\nmismatch?"}
Execute["execute_query()\nAthena SDK"]
TrackCost["cost.py\nDataScannedInBytes to USD"]
ValidateResults["validate_results()\nsanity checks"]
RenderChart["render_chart()\nmatplotlib + Plotly"]
Summarise["Call 4: Generate Insight\nClaude Platform on AWS"]
end
subgraph AWS ["AWS Data Platform"]
GlueCatalog[(Glue Catalog\nLive Schema)]
DbtArtifacts[(dbt catalog.json\nS3 Metadata)]
Athena[(Athena\nedp dev gold)]
S3Gold[(S3 Gold Layer\nParquet)]
AuditLog[(S3 Audit Log\nJSON per query)]
EngineerLog[(S3 Engineer Log\nCSV per request)]
end
subgraph OutputBlock ["Output Package"]
SQL[SQL Executed]
Assumptions[Assumptions Flagged]
ResultTable[Result Table]
Chart[Chart PNG or HTML]
Summary[Plain-English Insight]
CostLine[Scan Cost in USD]
VerdictOut[Intent Verdict + Detail]
end
LoadSchemas --> GlueCatalog
LoadSchemas --> DbtArtifacts
LoadSchemas -->|pre-loads into system prompt| GenerateSQL
Streamlit -->|POST /ask| Classify
Classify -->|analytical| GenerateSQL
Classify -->|conversational or retype| Summary
GenerateSQL --> ValidateSQL
ValidateSQL -->|pass| IntentCheck
ValidateSQL -->|fail: reason sent back| GenerateSQL
IntentCheck --> VerdictCheck
VerdictCheck --> Verdict
Verdict -->|mismatch: regenerate with feedback| GenerateSQL
Verdict -->|matches| Execute
Execute --> Athena
Athena -->|reads| S3Gold
Execute --> TrackCost
Execute --> ValidateResults
ValidateResults --> RenderChart
RenderChart --> Summarise
Summarise --> OutputBlock
Summarise --> AuditLog
Summarise --> EngineerLog
classDef box fill:#f0f4f8,stroke:#333,stroke-width:1px;
class Startup,Input,AgentLoop,AWS,OutputBlock box;
The FastAPI backend is the analytical engine. Non-technical stakeholders don't interact with it directly. The user-facing layer is a Streamlit browser app that wraps the backend.
A stakeholder opens a URL in their browser, types a business-language question in a text box, and clicks submit. Streamlit POSTs the question to the FastAPI /ask endpoint. When the response arrives, Streamlit renders everything inline: the business-language insight (streamed token-by-token as it generates), an interactive Plotly chart with a raw data table tab, the SQL in a "Query details" expander, the assumptions the agent made, and the scan cost in USD.
No command-line access needed. No SQL knowledge needed. No understanding of Athena, Glue, or partition structures required. That's the whole point.
The Streamlit app runs in the same ECS (Elastic Container Service) Fargate container as the FastAPI backend. A startup script (entrypoint.sh) launches both together when the container starts: FastAPI on port 8080, Streamlit on port 8501. The internet-facing ALB (Application Load Balancer) has a separate listener for each port, so stakeholders open http://<alb-dns>:8501 and FastAPI stays reachable on port 80 for direct API access. For local development, the same startup script runs both with a single docker-compose up.
The UI has several features beyond a simple text box:
- Multilingual interface. UI labels, section headings, placeholder text, and button labels are translated into eight scripts: Chinese (zh), Japanese (ja), Korean (ko), Arabic (ar), Russian (ru), Greek (el), Hebrew (he), and Thai (th). For Latin-script languages (Italian, French, Spanish, English, and others), Claude detects the language from the question text and replies in that language. The translation dictionaries cover 20+ string keys so the entire page feels native to the user's language, not just the answer.
- Numbered Q&A cards. Each question-answer pair is rendered as a numbered card with a border, not a chat bubble. The card shows the question, the streaming insight, the chart (or table), and a collapsible "Query details" section with SQL, assumptions, scan cost, and a query intent check.
- Query intent check. Before Athena runs, Claude first reads only the SQL and infers what business question the SQL is answering. The original question is withheld from that call except for a small language hint. A separate Claude verdict call then compares the original question with the inferred SQL intent and returns a
Yes/Nomismatch verdict. If a genuine mismatch is detected, FastAPI regenerates the SQL once with the discrepancy as feedback before any Athena query runs, so no scan cost is spent on SQL that is already known to be wrong. The final verdict is shown as a badge in the UI and recorded in the engineer log. - Chart/Table tabs. Every result has two tabs: an interactive Plotly chart and a raw data table. The user can switch between them without re-running the query.
- PDF report download. A "Download PDF" button generates a 2-page stakeholder report. Page 1 has the question, KPI tiles, the business-language summary, and the chart. Page 2 has the business-language methodology (assumptions rewritten without SQL, table names, or technical identifiers), plus a query intent cross-check. The PDF uses the army olive brand palette (
#4B5320), includes a geometric E logo mark, a generation timestamp, and "Page X of Y" page numbers in the footer. CJK (Chinese, Japanese, Korean) scripts use the Noto CJK font installed in the Docker image so characters render correctly. - Backend PDF report endpoint. Non-browser clients such as the Slack MCP gateway can call
POST /report/pdfwith a completed/askresponse and receive the branded PDF as base64. This keeps Slack report delivery on the same backend contract instead of rebuilding report logic in the gateway. - Conversation export and import. A sidebar button downloads the full conversation as a JSON file. A file uploader lets the user restore a previous conversation in a new session, so multi-day analysis workflows don't start from scratch.
- Session sidebar. Shows the session start time and a running count of questions asked in the current session.
What Streamlit is
Streamlit is a Python library. I write Python code and Streamlit turns it into a web page with a text box, buttons, and charts. There's no HTML, CSS, or JavaScript involved. The entire UI is Python, which fits naturally into a project where everything else is already Python.
What "running on a server" means
When a browser opens a web address, it connects to a program that is listening for incoming connections. That program is called a server. In this project, Streamlit is that server. When a stakeholder opens http://alb-url:8501, their browser connects to the Streamlit program running inside the ECS container in AWS. Streamlit sends the web page back to the browser. This is no different from visiting any website: there's always a server somewhere that sends the page.
Why the code calls localhost
Both Streamlit and FastAPI run inside the same ECS container. A container is a small private computer with its own isolated network. When the Streamlit code calls localhost:8080, it means "call the program listening on port 8080 inside this container", which is FastAPI. The stakeholder's browser never sees or uses localhost. They only ever type the ALB's DNS address. Localhost is purely internal, invisible to the outside world.
sequenceDiagram
autonumber
actor S as Stakeholder
participant ALB as AWS Load Balancer
participant UI as Streamlit UI (web page)
participant API as Analytics Backend (FastAPI)
participant CL as Claude Platform on AWS
participant ATH as Amazon Athena
participant S3 as S3 Gold and Audit
note over UI,API: These two live inside the same server (ECS container)
S->>ALB: Opens the app in their browser
ALB->>UI: Forwards the request to the UI
UI-->>S: Browser shows a text box and Submit button
S->>ALB: Types a question and clicks Submit
ALB->>UI: Forwards the question
UI->>API: Sends the question internally (no network hop: same server)
API->>CL: Call 1: Generate SQL
CL-->>API: SQL query
API->>API: Validate SQL guardrails
API->>CL: Call 2: Infer SQL intent (SQL only, question withheld except language hint)
CL-->>API: Inferred SQL intent
API->>CL: Call 3: Compare original question vs inferred intent
CL-->>API: Yes/No mismatch verdict + discrepancy detail
loop Retry once on intent mismatch
API->>CL: Regenerate SQL with correction detail
CL-->>API: Corrected SQL query
API->>API: Validate corrected SQL guardrails
API->>CL: Re-infer corrected SQL intent for audit
CL-->>API: Corrected inferred SQL intent
end
API->>ATH: Run query against Gold tables
ATH-->>API: Results: rows and columns
API->>CL: Call 4: Generate insight
CL-->>API: 2-3 sentence insight and chart title
API->>S3: Write audit log
API-->>UI: Returns insight, chart, cost, and verdict
UI-->>S: Renders the answer, chart, and cost in the browser
The stakeholder never runs a command. They open a URL, type a question, and read the answer. Everything else happens inside AWS.
Slack uses the same analytics backend as the browser UI, but it enters through a different front door. The browser path is:
Stakeholder -> AWS Load Balancer -> Streamlit UI -> FastAPI
The Slack path is:
Stakeholder -> Slack -> Slack MCP Gateway -> FastAPI
Both paths reach the same FastAPI /ask endpoint, so both get the same SQL
guardrails, intent check, Athena execution, chart logic, PDF report generation,
and S3 audit logging.
sequenceDiagram
autonumber
actor S as Stakeholder
participant Slack as Slack Workspace
participant Gateway as Slack MCP Gateway (ECS Fargate)
participant API as Analytics Backend (FastAPI)
participant CL as Claude Platform on AWS
participant ATH as Amazon Athena
participant S3 as S3 Gold and Audit
S->>Slack: Types a question in a channel or DM
Slack->>Gateway: Sends app mention / message event (Socket Mode)
Gateway->>Gateway: Extracts question, user, channel, and thread context
Gateway->>API: POST /ask with question and session_id
API->>CL: Call 1: Generate SQL
CL-->>API: SQL query + assumptions
API->>API: Validate SQL guardrails
API->>CL: Call 2: Infer SQL intent (SQL only, question withheld except language hint)
CL-->>API: Inferred SQL intent
API->>CL: Call 3: Compare original question vs inferred intent
CL-->>API: Yes/No mismatch verdict + discrepancy detail
loop Retry once on intent mismatch
API->>CL: Regenerate SQL with correction detail
CL-->>API: Corrected SQL query
API->>API: Validate corrected SQL guardrails
API->>CL: Re-infer corrected SQL intent for audit
CL-->>API: Corrected inferred SQL intent
end
API->>ATH: Run query against Gold tables
ATH->>S3: Reads curated Gold data
ATH-->>API: Results: rows and columns
API->>CL: Call 4: Generate insight
CL-->>API: 2-3 sentence insight and chart title
API->>S3: Write audit log and chart/report artifacts
API-->>Gateway: Returns insight, chart, cost, SQL, and verdict
opt PDF report requested
Gateway->>API: POST /report/pdf with completed /ask response
API-->>Gateway: Branded PDF as base64
end
Gateway->>Slack: Posts concise answer in channel
Gateway->>Slack: Adds SQL, assumptions, cost, and request ID in thread
Slack-->>S: Shows the answer, chart/report, and verification details
What each part achieves:
- Stakeholder in Slack keeps the experience where business users already work. They ask a question in business terms instead of opening AWS or writing SQL.
- Slack Workspace receives the message and sends it to the Slack app. Slack is the chat system, not the analytics engine.
- Slack MCP Gateway translates between Slack events and the analytics API. It handles app mentions, direct messages, Slack formatting, file uploads, and thread replies.
- Analytics Backend (FastAPI) does the actual reasoning work. It generates SQL, validates it, checks intent, runs Athena, builds charts, and writes logs.
- Claude Platform on AWS helps with language tasks: SQL generation, SQL intent cross-checking, and the final written insight.
- Amazon Athena executes the validated SELECT query against the Gold data.
- S3 Gold stores the curated business tables produced by dbt. This is the only data layer the agent queries.
- S3 Audit records what happened so an engineer can later inspect the question, SQL, assumptions, scan cost, request ID, and report output.
Slack is therefore an access layer, not a second analytics implementation. If a guardrail improves in FastAPI, both Streamlit and Slack benefit.
AWS Load Balancer (ALB)
The ALB (Application Load Balancer) is the front door to the application. Every request from a browser arrives here first. The stakeholder never connects directly to the application code; the ALB sits in front of it and decides where to send the traffic. It checks which port the request is for (port 8501 for the Streamlit web page, port 80 for the raw API) and forwards it to the right place inside the container. This is the only part of the system that has a public internet address.
Streamlit UI
Streamlit is a Python library that turns Python code into a web page. I wrote the entire interface in Python: no HTML, no JavaScript. Streamlit runs as a small web server inside the container. When a stakeholder opens the URL in their browser, Streamlit sends them the page with the text box and Submit button. When they submit a question, Streamlit forwards it to FastAPI (on the same server, so there's no network round trip) and waits for the response. When it arrives, Streamlit renders the insight, chart, KPI tiles, and SQL in the browser. Streamlit's job is display only: it holds no data and does no analysis.
Analytics Backend (FastAPI)
FastAPI is a Python web framework, meaning a program that listens for incoming requests and responds to them. This is the reasoning engine of the application. When a question arrives from Streamlit, FastAPI orchestrates three Claude calls, validates the SQL at each step, runs the Athena query, builds the chart, writes the audit log, and packages the response. All the business logic (SQL guardrails, intent checking, retry on mismatch, chart type detection) lives here. FastAPI runs on port 8080 inside the same container as Streamlit.
Claude Platform on AWS
Claude is Anthropic's AI model, accessed here through Claude Platform on AWS using the ECS task IAM role rather than a long-lived API key. I call it three times for each analytical question. The first call generates the SQL query and a list of assumptions (Claude reads the question against the full schema embedded in the system prompt). The second call reads only the SQL (the original question is withheld) and infers what business question the SQL is answering. This cross-check catches cases where the SQL technically runs but answers the wrong question. The third call reads the data returned by Athena and writes the business-language insight. Claude is stateless: each call is independent, and the relevant context is passed explicitly each time.
Amazon Athena
Athena is AWS's serverless SQL query engine. It doesn't have its own database. Instead, it reads Parquet files directly from S3 and runs SQL against them using the Glue Catalog as its schema reference. FastAPI sends Athena a SELECT query, Athena scans the relevant Gold Parquet files, and returns rows and columns. The cost is pay-per-byte-scanned: a well-partitioned query that touches one month's data costs a fraction of a cent. A query that accidentally scans the full table costs proportionally more. This is why SQL guardrails and partition filters matter here in a way they don't in a fixed-cost warehouse.
S3 Gold and Audit
S3 (Simple Storage Service) appears in the diagram for two different purposes. The Gold bucket holds the seven pre-computed Parquet mart tables (monthly_revenue_trend, revenue_by_country, and five others) that Athena queries. These are produced by dbt and refreshed every time the pipeline runs. The Audit path is a separate prefix in the Bronze bucket where FastAPI writes a JSON record after every question: the question, the SQL, the number of rows returned, the Athena scan cost in USD, the intent verdict, response time, and retry count. This log is what the engineer session download in the sidebar reads from.
The agent starts each ECS task by loading all Gold schemas once and embedding them in the Claude system prompt. Claude knows every table and column before it sees the first question. This eliminates the multi-turn list_tables / get_schema tool-call round trips that text-to-SQL systems typically need.
SchemaResolver.load_all_schemas() runs once at startup. It reads all Gold tables from Glue Catalog (column names, data types, partition keys) and overlays dbt catalog.json from S3 (column descriptions, model documentation) for every table. The result (all 7 Gold schemas, roughly 2,500 tokens) is embedded directly in the system prompt.
If catalog.json isn't present yet (the pipeline hasn't run), it falls back to Glue-only schema and logs a warning. No crash, no partial startup.
Before generating any SQL, the agent makes a lightweight Claude call to classify the question into one of three types:
- Analytical: needs a new Athena query. Takes the full pipeline path.
- Conversational: can be answered from prior conversation context alone (e.g. "what did you say last?", "translate that to French"). Goes straight to Claude with the session summary, no Athena.
- Retype: the user wants the same data rendered as a different chart type (e.g. "show that as a bar chart"). Re-runs the previous SQL with a forced chart type override.
Multi-turn context is also resolved here. If the user has asked previous questions in the same session, a summary of those turns is appended to the system prompt so Claude can resolve references like "what about Q4?" without needing the full history in every API call.
Claude reads the question against the schema already in the system prompt and returns a SELECT query plus a list of assumptions (for example, "'transactions' interpreted as completed orders only"). No tool calls needed in the common case.
SQLValidator parses the query with sqlparse and enforces hard rules: SELECT only, Gold database only, no DDL keywords anywhere, LIMIT present. If validation fails, the error reason is sent back to Claude with a correction request. Up to 3 attempts before raising SQLValidationError to the user. The Athena WorkGroup bytes_scanned_cutoff_per_query setting in Terraform is the hard cost backstop.
Before running any Athena query, the agent checks that the generated SQL actually answers the original question. It makes a second Claude call passing only the SQL and asks Claude to infer what business question the SQL is answering. The original question is structurally withheld from this intent-inference call, except for a small language hint so the response uses the user's language. Then the agent makes a separate Claude verdict call with both the original question and the inferred SQL intent, and Claude returns a Yes/No mismatch verdict plus a one-sentence discrepancy detail.
If a genuine mismatch is detected (wrong metric, wrong table, wrong filter), the SQL is regenerated once with the discrepancy detail as feedback. This retry happens before touching Athena, so no query cost or scan time is spent on SQL that's already known to be wrong. The loop runs at most once per question.
AthenaExecutor starts the query, polls until complete, and reads the result CSV from the athena-results S3 bucket. cost.py converts DataScannedInBytes from the Athena execution metadata to USD. No pre-execution cost estimation needed. Gold tables are small pre-aggregations, and the WorkGroup hard stop handles any outliers.
ResultValidator checks the DataFrame for obvious anomalies: negative values in revenue columns, unexpected nulls on key columns. Zero rows is a valid result for Gold tables: an aggregation with no matching data is a legitimate answer, not a bug. Flags are surfaced in the output, never block execution.
ChartGenerator selects chart type from the shape of the result and keywords in the question. Five types are supported: line (time + 1 metric), multi-line (time + 2+ metrics), bar (categorical + metric), scatter (2 numeric + categorical, correlation question), and pie/donut (proportion question, ≤ 8 rows). The chart is uploaded to S3 and returned as a presigned URL (Uniform Resource Locator).
InsightGenerator makes a Claude call with the original question, SQL, result sample, and assumptions, and returns a 2-3 sentence business-language insight.
Two records are written after every analytical query:
Audit log: a structured JSON record at s3://{bronze_bucket}/metadata/agent-audit/ containing the original question, SQL, assumptions, row count, bytes scanned, cost in USD, validation flags, and insight. The audit log is itself queryable via Athena.
Engineer log: a single-row CSV at s3://{bronze_bucket}/metadata/engineer-log/date={date}/session={id}/{request_id}.csv with 17 columns: session ID, request ID, timestamp, question, SQL executed, Claude's inferred interpretation, verdict, discrepancy detail, bytes scanned, Athena cost, response time, Athena execution ID, SQL retry count, row count returned, chart type rendered, language detected, and prompt version. One file per request so individual rows can be read without scanning the full log. The engineer log download button in the sidebar fetches and concatenates all rows for the current session.
A rate limiter enforces a maximum of 10 requests per 60-second window per session and returns HTTP 429 on breach. This protects both the Claude Platform on AWS budget and Athena scan costs.
These are non-negotiable and enforced before any query reaches Athena.
| Guardrail | How it's enforced |
|---|---|
| SELECT only | sqlparse rejects anything that isn't a single SELECT statement |
| Gold DB only | Target database validated against edp_{env}_gold whitelist |
| LIMIT required | Injected if the model omits it, default 1000 rows |
| No DDL in any form | DROP, DELETE, INSERT, UPDATE, CREATE, ALTER, TRUNCATE rejected in statement or subquery |
| Partition filter required | Queries against large tables must include at least one partition key filter |
| Retry safety | Retry only on transient errors (throttling, timeout). Never on semantic failures (table not found, permission denied) |
| Cost hard stop | Athena WorkGroup bytes_scanned_cutoff_per_query in Terraform processing module |
| Read-only IAM | The agent's ECS task role has zero write permissions on Bronze, Silver, or Gold data |
The MWAA DAG includes a final task upload_dbt_artifacts that runs after every successful dbt test. It copies target/manifest.json and target/catalog.json from the dbt workspace to s3://{bronze_bucket}/metadata/dbt/.
The agent reads this path at query time, never from a local cache. When a dbt model is renamed, a column description is updated, or a new Gold table is added, the agent sees the change automatically at the next query after the next pipeline run. Schema drift from dbt refactors is impossible because the agent never holds a stale copy.
The agent queries the Gold layer. There are 7 mart tables, each designed to answer a specific category of business question. The questions below are grounded in the exact columns those tables expose.
Understanding the boundaries matters as much as knowing what works. Some questions sound reasonable but can't be answered with the current data, either because the mart doesn't have the right dimension, or because that field was never captured anywhere in the pipeline. Both sets are listed below.
monthly_revenue_trend: one row per year-month. Columns: order_year, order_month, total_orders, unique_customers, total_revenue, cancelled_orders.
- What were total sales last month?
- Show me revenue by month for this year.
- How has total order volume changed over the past 12 months?
- Which month had the highest number of cancellations?
- How many unique customers placed an order in each month this year?
- Is revenue trending up or down compared to the same month last year?
revenue_by_country: one row per country, all-time totals for completed orders. Columns: country, total_orders, total_customers, total_revenue, avg_order_value.
- Which country generates the most revenue?
- What is the average order value for customers in Germany?
- How many distinct customers have placed completed orders in France?
- Rank all countries by total revenue.
- Which country has the highest average order value?
payment_method_performance: one row per payment method. Columns: payment_method, total_transactions, successful, failed, refunded, success_rate_pct, total_processed, revenue_captured.
- Which payment method has the highest failure rate?
- How much revenue was lost to refunds across all payment methods?
- What percentage of bank transfer payments complete successfully?
- Which payment method processes the most total volume?
- How many transactions were refunded via credit card?
top_selling_products: one row per product. Columns: product_id, product_name, category, brand, total_orders, total_units_sold, total_revenue, avg_revenue_per_unit, revenue_rank.
- What are the top 10 best-selling products by total revenue?
- Which product has sold the most units?
- What is the average revenue per unit for the top 5 products?
- Which brand appears most often in the top 20 products by revenue rank?
- Which product generates the most revenue per unit sold?
product_category_performance: one row per category-brand combination. Columns: category, brand, total_orders, products_in_category, total_units_sold, total_revenue, avg_revenue_per_unit.
- Which product category generates the most revenue?
- Which brand has the highest revenue across all categories?
- How many distinct products has each category sold?
- Which category sells the most units?
- Compare average revenue per unit across Electronics, Clothing, and Sports.
carrier_delivery_performance: one row per carrier. Columns: carrier, total_shipments, delivered, failed, delivery_success_rate_pct, avg_delivery_days, fastest_delivery_days, slowest_delivery_days.
- Which carrier has the highest delivery success rate?
- What is the average delivery time for DHL?
- Which carrier has the most failed shipments?
- How do FedEx and UPS compare on average delivery days?
- Which carrier achieves the fastest single delivery on record?
- What is the spread between fastest and slowest delivery times for each carrier?
customer_segments: one row per customer. Columns: customer_id, first_name, last_name, email, country, signup_date, total_orders, lifetime_value, first_order_date, last_order_date, segment (VIP / Regular / Low Value / Never Ordered), order_frequency_band (Loyal / Occasional / One-Time / No Orders).
- How many VIP customers do we have?
- Which country has the most Loyal customers?
- Who are the top 10 customers by lifetime value?
- How many customers have never placed an order?
- What share of customers are one-time buyers?
- Which customers signed up in the last 90 days and are already VIP?
- How is lifetime value distributed across segments in the UK?
The agent responds in the same language as the question. Any language Claude supports works. Here are three examples grounded in the actual Gold tables.
German
-
Welche Zahlungsmethode hat die höchste Ausfallrate, und wie viel Umsatz ging dadurch verloren? (Which payment method has the highest failure rate, and how much revenue was lost as a result?)
-
Wie hat sich der monatliche Gesamtumsatz in diesem Jahr im Vergleich zum Vorjahreszeitraum entwickelt? (How has total monthly revenue developed this year compared to the same period last year?)
-
Welche fünf Kunden haben den höchsten Lifetime Value, und aus welchem Land kommen sie? (Which five customers have the highest lifetime value, and which country are they from?)
Chinese (Simplified)
-
哪个运输承运商的平均配送天数最少，成功交付率最高？ (Which shipping carrier has the lowest average delivery days and the highest delivery success rate?)
-
按国家划分，哪个市场的总收入最高，平均订单价值是多少？ (By country, which market has the highest total revenue, and what is the average order value?)
-
在所有产品类别中，哪个品牌的单位平均收入最高？ (Across all product categories, which brand has the highest average revenue per unit?)
Italian
-
Quali sono i 10 prodotti più venduti per fatturato totale e quante unità ha venduto ciascuno? (What are the top 10 products by total revenue, and how many units has each sold?)
-
Quanti clienti VIP abbiamo e qual è il loro valore medio nel corso della vita? (How many VIP customers do we have, and what is their average lifetime value?)
-
Qual è il tasso di successo delle transazioni per ogni metodo di pagamento? (What is the transaction success rate for each payment method?)
These questions sound reasonable but cannot be answered with the current Gold layer. The reason for each is specific: either the mart lacks a dimension, or the field was never captured anywhere in the pipeline.
No time dimension in revenue_by_country
"What was Germany's revenue last month?"
revenue_by_country is a lifetime aggregation with no order_year or order_month column. monthly_revenue_trend has time but no country. There's no mart that combines both. The agent will say it can't answer this precisely and explain why.
No country dimension in monthly_revenue_trend
"Show me monthly revenue broken down by country."
Same problem in reverse. The two finance marts cover different cuts of the same underlying data but there's no mart that crosses both dimensions. Adding one would require a new dbt model.
Stock levels not in Gold
"Which products are running low on inventory?"
stock_qty exists in the Silver products table and flows through stg_products, but no Gold mart exposes it. It was excluded because it's a point-in-time operational field, not a business aggregation. The agent has no visibility into it.
No cost-of-goods data
"What is the gross margin on Electronics?"
Revenue is captured throughout the pipeline (line_total, payment_amount), but cost of goods sold is not. There is no unit_cost, cogs, or margin column anywhere from the OLTP source through to Gold. Margin questions of any kind cannot be answered.
No shipping cost data
"Which carrier is the most cost-effective?"
carrier_delivery_performance tracks delivery days and success rates but not the fee charged per shipment. Shipping cost is not in the source OLTP schema and therefore not anywhere in the pipeline.
No city or region geography
"Which cities in France drive the most revenue?"
Customer geography is captured at country level only. The CDC simulator generates a country field but no city, region, state, or postal code. These fields don't exist in Silver or Gold.
No real-time data
"What orders came in the last hour?"
Gold is updated by the daily batch pipeline. The latest data reflects the most recent successful edp_pipeline run. The agent can't see anything more recent than that.
No marketing or acquisition data
"Which marketing channel brings the most VIP customers?"
The OLTP system records what customers ordered, not how they were acquired. There is no campaign, referral, UTM parameter, or acquisition source anywhere in the data model.
No cart abandonment or browse data
"Which products do customers view but not buy?"
The platform captures CDC events from a PostgreSQL OLTP database. Only database writes are tracked, which means only orders that were created. Browse events, add-to-cart actions, and session data don't exist.
No refund reason or return details
"What is the most common reason customers return Electronics?"
payment_method_performance tracks a refund count per payment method, but there is no refund reason, return category, or customer comment field anywhere. The OLTP schema has a status field on payments but no reason code.
No product co-purchase patterns
"Customers who buy X also buy Y: what are the top product pairs?"
There's no basket analysis mart. top_selling_products and product_category_performance aggregate each product independently. Building this would require a self-join on order items at the intermediate layer, which hasn't been modelled.
This section covers how to test the agent end-to-end. There are two ways to ask it a question.
Run these in order. Each one takes about a minute. If any step fails, the numbered troubleshooting items below explain exactly what to do.
[ ] 1. Create a Claude Platform on AWS workspace (AWS console → Claude Platform on AWS → Workspaces)
[ ] 2. Apply infra: cd terraform-platform-infra-live && make apply dev
[ ] 3. Trigger the Airflow DAG (edp_pipeline) in the MWAA console and wait for it to finish
[ ] 4. Test locally: cd platform-analytics-agent && source .venv/bin/activate
export $(grep -v '^#' .env | xargs)
python -m agent.main "Which country has the highest total revenue?"
[ ] 5. If local works, test ECS: see Track 2 below
[ ] 6. Destroy when done: cd terraform-platform-infra-live && make destroy dev
Step 4 runs the agent locally against real AWS services and prints the answer in the terminal. Steps 1-4 are all that's needed for a quick functional test.
Track 1 (recommended, start here): The agent code runs locally but connects to the real AWS dev environment (real Athena, real Glue, real S3). No ECS needed. This is the fastest way to iterate and confirms all AWS integrations are working.
Track 2: The agent is deployed to ECS Fargate. I shell into the running container using the AWS CLI and make HTTP requests to the FastAPI server from inside the container. This confirms the deployed Docker image and ECS service are healthy.
Start with Track 1. Only move to Track 2 once Track 1 is producing correct answers.
1. Store the Claude workspace ID in SSM Parameter Store.
The agent authenticates to Claude via the ECS task IAM role (Claude Platform on AWS). No API key is needed. The only prerequisite is the Claude workspace ID, stored in SSM Parameter Store as a plain string.
In the AWS console, navigate to Systems Manager > Parameter Store in eu-central-1. Create a parameter:
| Field | Value |
|---|---|
| Name | /edp/dev/claude/workspace_id |
| Type | String |
| Value | Workspace ID from AWS console → Claude Platform on AWS → Workspaces |
This only needs to be done once per environment. The parameter survives terraform destroy.
From the terminal:
aws ssm put-parameter \
--name "/edp/dev/claude/workspace_id" \
--type "String" \
--value "<workspace-id>" \
--overwrite \
--profile dev-admin \
--region eu-central-1Then Terraform applies with:
TF_VAR_claude_provider=aws_claude_platformThe ECS task role grants aws-external-anthropic:CreateInference, aws-external-anthropic:CountTokens, sts:GetWebIdentityToken, and sts:TagGetWebIdentityToken on the workspace ARN. No static API key is stored anywhere.
2. Confirm the MWAA pipeline has run and Gold data exists.
The agent queries the Gold Athena tables. If the pipeline hasn't run yet, every query returns zero rows (or a table-not-found error if the Glue Catalog is empty). Log into the Airflow UI for the MWAA environment (edp-dev-mwaa) and confirm edp_pipeline has at least one successful DAG run. If it hasn't, trigger it manually and wait for it to complete (about 6-8 minutes).
You can also confirm Gold data exists by running a quick Athena query in the AWS console:
SELECT COUNT(*) FROM "edp_dev_gold"."fct_orders" LIMIT 1;If this returns a count greater than zero, the Gold layer is ready.
3. Confirm the deploy workflow completed.
When the latest code changes are pushed to main, GitHub Actions runs the CI workflow first (lint, type check, unit tests, Docker build). Once CI passes, the Deploy workflow triggers automatically and builds the Docker image, pushes it to ECR (Elastic Container Registry), and updates the ECS task definition.
In the GitHub repository for platform-analytics-agent, click Actions and confirm both the CI and Deploy workflows have a green tick for the latest push. If Deploy is still running, wait for it to finish before testing Track 2.
This runs the agent Python code directly against the real AWS dev environment. It uses the local dev-admin AWS profile for credentials, connects to real Athena, real Glue Catalog, and real SSM. The output is identical to what ECS produces: the only difference is that the compute runs locally instead of on Fargate.
Step 1: Create the .env file.
Copy .env.example to .env:
cd platform-analytics-agent
cp .env.example .envOpen .env and fill in the values. The only things to change are the bucket names and the AWS account ID. Find the account ID by running:
aws sts get-caller-identity --profile dev-admin --query Account --output textUpdate .env with the account ID substituted in:
AWS_REGION=eu-central-1
AWS_PROFILE=dev-admin
ENVIRONMENT=dev
BRONZE_BUCKET=edp-dev-YOUR_ACCOUNT_ID-bronze
GOLD_BUCKET=edp-dev-YOUR_ACCOUNT_ID-gold
ATHENA_RESULTS_BUCKET=edp-dev-YOUR_ACCOUNT_ID-athena-results
ATHENA_WORKGROUP=edp-dev-workgroup
GLUE_GOLD_DATABASE=edp_dev_gold
CLAUDE_PROVIDER=aws_claude_platform
ANTHROPIC_AWS_WORKSPACE_ID=<workspace-id-from-ssm>
ANTHROPIC_AWS_INFERENCE_GEO=us
COST_THRESHOLD_USD=0.10
MAX_ROWS=1000
Step 2: Activate the virtual environment.
source .venv/bin/activateIf make setup hasn't been run yet:
make setup
source .venv/bin/activateStep 3: Load the .env file.
The agent reads these values from environment variables, not from the .env file directly. Load them into the shell session:
export $(grep -v '^#' .env | xargs)Step 4: Ask a first question.
python -m agent.main "Which country has the highest total revenue?"The agent takes 12-20 seconds to respond. On the first run there's an additional few seconds for schema loading (Glue Catalog + dbt catalog.json from S3). What the agent prints:
- The SQL it generated
- Every assumption it made ("'revenue' interpreted as the
total_pricecolumn on completed orders") - The result table
- Bytes scanned and cost in USD
- A 2-3 sentence business-language insight
- A presigned S3 URL (Uniform Resource Locator) for the chart PNG
The agent picks a chart type automatically from the shape of the query result and, where relevant, keywords in the question.
| Chart type | Question keywords that trigger it | Data shape required | Example question |
|---|---|---|---|
| Line | (data-driven, no keyword needed) | Time column + 1 numeric metric | "Show monthly revenue over the last year" |
| Multi-line | (data-driven, no keyword needed) | Time column + 2+ numeric metrics | "Show me both revenue and order volume trends over the last 12 months" |
| Bar | (data-driven, no keyword needed) | Categorical + 1 numeric metric, no time | "Which country generates the most revenue?" |
| Scatter | correlation, relationship, vs, versus, against, affect, predict, scatter |
2+ numeric columns + 1 categorical label column | "Is there a correlation between payment volume and revenue lost by payment method?" |
| Pie / Donut | share, proportion, distribution, breakdown, percentage, split, mix, composition, percent |
1 categorical + 1 numeric, ≤ 8 rows, no time | "What is the revenue breakdown by payment method?" |
| Table | (fallback) | No numeric columns, or no matching shape | "List all payment methods and their status" |
Detection priority when rules overlap: scatter > multiline > line > pie > bar > table.
Step 5: Try these test questions.
Run each one with python -m agent.main "question". They're ordered to cover different Gold tables, different chart types, and different kinds of reasoning.
# Bar chart: top categories, single aggregation
python -m agent.main "Show me total revenue by country"
# Line chart: time-series, tests date reasoning
python -m agent.main "What does monthly order volume look like over the last 12 months?"
# Multi-line chart: two metrics on the same time axis
python -m agent.main "Show me both revenue and order volume trends over the last 12 months"
# Scatter chart: correlation between two numeric variables
python -m agent.main "Is there a correlation between payment volume and revenue lost by payment method?"
# Pie / Donut chart: proportion question, ≤8 categories
python -m agent.main "What is the revenue breakdown by payment method?"
# Filtering + aggregation: tests WHERE clause generation
python -m agent.main "Which product categories are most popular in Germany?"
# Count + group by: tests COUNT vs SUM disambiguation
python -m agent.main "How many unique customers placed orders in each country?"
# Trend question: tests pattern-recognition in insight generation
python -m agent.main "Is revenue growing or declining? Show me the trend."What each response should show:
- The SQL should have a
WHEREclause with a partition filter (e.g.,dt >= ...): this confirms partition-aware query generation is working - The assumptions list should explain any interpretation decisions Claude made
- The insight should be specific to the actual numbers in the result, not generic
- The presigned URL should be a real S3 URL: open it in a browser to see the chart
- Bytes scanned should be small (under 10 MB for Gold tables): confirms the partition filters are working
- Cost should be under $0.001 per query: confirms the Gold layer's efficiency
Step 6: Test multi-turn follow-up.
For multi-turn follow-up (where the agent remembers the previous question), the HTTP endpoint is required. That's covered in Track 2. The CLI mode (python -m agent.main) is single-turn only: each invocation is independent.
Once Track 1 passes, this confirms the Docker image running on ECS Fargate is working. The ALB (Application Load Balancer) in front of the service is internal-only: it can't be reached directly from a browser. Instead, I use aws ecs execute-command to open a shell inside the running container and make HTTP requests from there.
Step 1: Confirm the ECS service has a running task.
aws ecs describe-services \
--cluster edp-dev-analytics-agent \
--services edp-dev-analytics-agent \
--profile dev-admin \
--region eu-central-1 \
--query "services[0].{running:runningCount,pending:pendingCount,taskDef:taskDefinition}" \
--output jsonThe expected result is running: 1. If it shows running: 0, wait a minute and run it again. If it stays at zero, check the CloudWatch logs (Step 3).
Step 2: Get the running task ID.
aws ecs list-tasks \
--cluster edp-dev-analytics-agent \
--desired-status RUNNING \
--profile dev-admin \
--region eu-central-1 \
--query "taskArns[0]" \
--output textThis returns a long ARN. The task ID is the last segment after the final /. For example, from:
arn:aws:ecs:eu-central-1:123456789012:task/edp-dev-analytics-agent/3e4ea78b0be54d61
The task ID is 3e4ea78b0be54d61. You'll use this in the next steps.
Step 3: Check the health endpoint.
Replace TASK_ID with the value from Step 2.
aws ecs execute-command \
--cluster edp-dev-analytics-agent \
--task TASK_ID \
--container agent \
--interactive \
--command "python -c \"import urllib.request; print(urllib.request.urlopen('http://localhost:8080/health').read().decode())\"" \
--profile dev-admin \
--region eu-central-1Expected output: {"status":"ok"}
If SessionManagerPlugin is not found, install it first:
brew install --cask session-manager-pluginStep 4: Ask a question to the ECS-deployed agent.
Replace TASK_ID with the task ID from Step 2 and replace the question text as needed.
aws ecs execute-command \
--cluster edp-dev-analytics-agent \
--task TASK_ID \
--container agent \
--interactive \
--command "python -c \"
import urllib.request, json
body = json.dumps({'question': 'Which country has the highest total revenue?'}).encode()
req = urllib.request.Request(
'http://localhost:8080/ask',
data=body,
headers={'Content-Type': 'application/json'},
method='POST'
)
resp = urllib.request.urlopen(req, timeout=120)
data = json.loads(resp.read().decode())
print()
print('INSIGHT:', data['insight'])
print()
print('SQL EXECUTED:')
for a in data.get('assumptions', []):
print(' -', a)
print()
print('BYTES SCANNED:', data['bytes_scanned'])
print('COST USD:', data['cost_usd'])
print('CHART TYPE:', data['chart_type'])
print('CHART URL:', data.get('presigned_url'))
print('SESSION ID:', data['session_id'])
\"" \
--profile dev-admin \
--region eu-central-1The command takes 15-30 seconds. The agent is loading schemas from Glue, generating SQL with Claude, running the Athena query, and producing the insight.
Output:
INSIGHT: a 2-3 sentence business-language answer based on the actual data- The assumptions list: what the agent interpreted (for example, "'revenue' means total price on completed orders only")
BYTES SCANNED: how much Athena data was read (should be small for Gold tables, under 10 MB)COST USD: the Athena query cost (should be under $0.001)CHART URL: an S3 presigned URL. Open it in a browser to see the chart image.SESSION ID: copy this for follow-up questions
Step 5: Ask a follow-up question (multi-turn).
Copy the SESSION ID from Step 4 and paste it in as YOUR_SESSION_ID. The agent remembers the previous question and answer.
aws ecs execute-command \
--cluster edp-dev-analytics-agent \
--task TASK_ID \
--container agent \
--interactive \
--command "python -c \"
import urllib.request, json
body = json.dumps({
'question': 'Now break that down by city',
'session_id': 'YOUR_SESSION_ID'
}).encode()
req = urllib.request.Request(
'http://localhost:8080/ask',
data=body,
headers={'Content-Type': 'application/json'},
method='POST'
)
resp = urllib.request.urlopen(req, timeout=120)
data = json.loads(resp.read().decode())
print('INSIGHT:', data['insight'])
print('SESSION ID:', data['session_id'])
\"" \
--profile dev-admin \
--region eu-central-1Step 6: Read the CloudWatch logs if something goes wrong.
All errors and startup messages are written to CloudWatch. This is the first place to look when a request fails.
aws logs tail /ecs/edp-dev-analytics-agent \
--follow \
--profile dev-admin \
--region eu-central-1Press Ctrl-C to stop tailing. The most common startup errors are documented in the troubleshooting section below.
These six questions cover every Gold table, every chart type, and several multi-turn follow-up patterns. Run them in sequence during a demo session.
| # | Question | Chart type | What it tests |
|---|---|---|---|
| 1 | "Which country has the highest total revenue?" | Bar | Top-N aggregation, rank ordering |
| 2 | "Show me monthly revenue as a trend over the last year" | Line | Time-series, date reasoning |
| 3 | "Show me both revenue and order volume trends over the last 12 months" | Multi-line | Two metrics on one time axis |
| 4 | "Is there a correlation between payment volume and revenue lost by payment method?" | Scatter | Correlation question, trend line |
| 5 | "What is the revenue breakdown by payment method?" | Pie / Donut | Proportion question, ≤8 categories |
| 6 | "Compare average order value across all countries" | Bar | Horizontal bar with many categories |
| 7 | "How many unique customers have placed orders in each country?" | Bar | COUNT DISTINCT, SUM vs COUNT disambiguation |
| 8 | "Is there a seasonal pattern in order volume?" | Line | Pattern-recognition insight |
For multi-turn follow-up, after question 2 ask: "Which month had the lowest volume and why do you think that is?": the agent answers referencing the same SQL execution without re-running the query.
"Missing required environment variables": The export $(grep -v '^#' .env | xargs) step was skipped, or a variable name is misspelled in .env.
"SchemaResolutionError: Glue Catalog unreachable": The infra isn't up, or the dev-admin profile doesn't have permission to call glue:GetTables. Confirm terraform apply completed successfully and the edp_dev_gold Glue database exists in the AWS console.
"No tables found in Gold database": The MWAA pipeline hasn't run yet. Trigger edp_pipeline in the Airflow UI and wait for it to complete.
"ParameterNotFound: /edp/dev/claude/workspace_id": The SSM prerequisite step was skipped. Run the aws ssm put-parameter command from the Prerequisites section.
"AccessDenied on SSM GetParameter": The ECS task role doesn't have permission to read this SSM path. Confirm terraform apply ran successfully and the analytics-agent module is included.
"Athena query failed: TABLE_NOT_FOUND": Either the Gold Glue Catalog tables don't exist (run the MWAA pipeline) or the GLUE_GOLD_DATABASE value in .env is wrong (should be edp_dev_gold with underscores, not hyphens).
Deploy workflow triggered but ECS task keeps stopping: Check CloudWatch Logs at /ecs/edp-dev-analytics-agent for the startup error. The most common cause is a missing SSM parameter or a container startup crash.
"Athena query failed: Insufficient permissions on glue:GetDatabase for database/silver": dbt-athena embeds the dbt source name (silver) as the Glue database in Gold view definitions, not the full name (edp_dev_silver). The task role needs Glue read permissions on both edp_dev_silver and the literal name silver. This is already fixed in the Terraform module.
"Athena query failed: PERMISSION_DENIED s3:ListBucket on edp-dev-...-silver": Gold views read from the underlying Silver Parquet files. The task role needs s3:GetObject and s3:ListBucket on the Silver S3 bucket. This is already fixed in the Terraform module.
"Athena: Unable to verify/create output bucket": The task role needs s3:GetBucketLocation on the Athena results bucket (not just s3:GetObject). This is already fixed in the Terraform module.
The agent runs as an ECS (Elastic Container Service) Fargate service. It starts automatically when the ECS service is created by Terraform. The deploy.yml GitHub Actions workflow builds and deploys a new image on every merge to main.
ECS cluster: edp-dev-analytics-agent
ECS service: edp-dev-analytics-agent
ECR repository: edp-dev-analytics-agent
CloudWatch log group: /ecs/edp-dev-analytics-agent
HTTP endpoints (reachable from inside the container via ECS Exec):
POST http://localhost:8080/ask
GET http://localhost:8080/health
ALB DNS (internal, not reachable from the public internet):
internal-edp-dev-agent-alb-753909442.eu-central-1.elb.amazonaws.com
The ALB is intentionally internal-only. Reaching it from outside the VPC requires a bastion host or VPN tunnel. For dev testing, ECS Exec as described in Track 2 is simpler and requires no additional infrastructure.
The ECS task role is defined in terraform-platform-infra-live/modules/analytics-agent/main.tf and scoped to exactly what the agent needs. Nothing more.
Athena:
- athena:StartQueryExecution
- athena:GetQueryExecution
- athena:GetQueryResults
- athena:StopQueryExecution
S3 (read):
- s3:GetObject on {bronze_bucket}/metadata/dbt/*
- s3:GetObject on {gold_bucket}/*
- s3:GetObject, s3:PutObject on {athena_results_bucket}/*
S3 (write: agent outputs only):
- s3:PutObject on {bronze_bucket}/metadata/agent-audit/*
- s3:PutObject on {bronze_bucket}/metadata/engineer-log/*
- s3:PutObject on {gold_bucket}/charts/*
Glue:
- glue:GetTable
- glue:GetDatabase
- glue:GetPartitions
on edp_{env}_gold database only
SSM:
- aws-external-anthropic:CreateInference and CountTokens on the Claude workspace ARN
- sts:GetWebIdentityToken and sts:TagGetWebIdentityToken (OIDC credential exchange)
Each phase has a clear deliverable. No phase starts until the previous one passes make lint, make typecheck, and make test.
Project skeleton with CI from the first commit. No business logic yet.
pyproject.toml,.python-version,requirements.txt,requirements-dev.txtMakefile: setup, lint, typecheck, test, run targetsDockerfile(two-stage build, non-root user) +docker-compose.yml.env.example,.gitignoreagent/exceptions.py: named exception hierarchy (AgentError,SchemaResolutionError,SQLValidationError,CostLimitError,ExecutionError,ResultValidationError)agent/config.py: frozen dataclasses driven by environment variables, fail fast at startup if any required variable is missingagent/logging.py: structured JSON logger used by every module from day one.github/workflows/ci.yml: ruff + mypy + pytest on every pushtests/conftest.py: shared fixtures for mocked AWS clients and mocked Claude Platform on AWS responses
Deliverable: make lint, make typecheck, make test all pass. Docker image builds cleanly. CI is green.
Written before any AWS code so the executor is coded to the permission boundary, not retrofitted after.
All AWS infrastructure lives in terraform-platform-infra-live/modules/analytics-agent/: the same repo and state file as the rest of the platform. This is intentional: the agent's IAM role references bucket names, KMS key ARN, and Glue database name that are outputs of sibling modules. Keeping everything in one state file means no manual tfvars to maintain, and the teardown workflow covers the agent automatically.
Resources: ECR repository (scan on push, lifecycle policy keeps last 10 images), ECS cluster (FARGATE, Container Insights enabled), CloudWatch log group (30-day retention, KMS-encrypted), task execution role (ECR pull + CloudWatch write only), task IAM role (scoped exactly), security group (egress port 443 only), ECS task definition (512 CPU / 1024 MB, lifecycle.ignore_changes so CI updates the image without Terraform re-deploying).
Task IAM role grants: Gold S3 read-only, Athena results bucket read/write, Bronze metadata/dbt/* read, Bronze metadata/agent-audit/* write, Glue Gold catalog read-only, Athena query execution on the platform workgroup only, Claude Platform on AWS CreateInference and CountTokens on the workspace ARN, STS GetWebIdentityToken and TagGetWebIdentityToken for the OIDC credential exchange, KMS decrypt on platform key only. No wildcard resources on AWS data services. The workspace ID is read from /edp/{env}/claude/workspace_id at Terraform apply time.
Deliverable: terraform plan in terraform-platform-infra-live/environments/dev produces the correct IAM role. All application AWS code is written inside this permission boundary.
The Gold layer has 7 small, pre-aggregated tables with 5-10 columns each. All schemas are loaded eagerly at startup and embedded in the system prompt: Claude knows every table and column before it sees the first question. This eliminates the list_tables / get_schema tool call round trips from the common case and is the single biggest latency saving in the design.
agent/schema.py:SchemaResolverclass:load_all_schemas(): called once at startup. Readscatalog.jsonfroms3://{bronze_bucket}/metadata/dbt/and fetches all Gold tables fromglue_client.get_tables(). Merges physical schema (column names, types, partition keys) with business context (column descriptions, accepted values, model docs) for every table. Returns a single dict covering all 7 Gold tables (~2,500 tokens total). This dict is embedded directly in the system prompt so Claude starts every query with full schema awareness.get_schema(table_name): available as a tool for edge cases where Claude needs to re-examine one table during reasoning, but won't be called in normal operation.- Graceful fallback if
catalog.jsondoesn't exist yet (pipeline hasn't run): falls back to Glue-only schema with a warning logged.
tests/test_schema.py: parametrized tests with full mock fixtures for both Glue and S3 responses, including the fallback path.
Deliverable: SchemaResolver.load_all_schemas() returns the complete merged schema for all Gold tables in one call. Tested with and without catalog.json present.
Guardrails are built before the SQL generator so no generated SQL can ever bypass them.
agent/validator.py:SQLValidator:- Parses with sqlparse
- Rejects anything that isn't a single SELECT statement
- Rejects any DDL keyword anywhere in the statement or any subquery (
DROP,DELETE,INSERT,UPDATE,CREATE,ALTER,TRUNCATE) - Rejects any database reference outside
edp_{env}_gold - Injects
LIMIT 1000if missing - Checks that at least one partition key filter is present for large tables
- Returns validated SQL or raises
SQLValidationErrorwith a reason string Claude can act on
tests/test_validator.py: parametrized, one test case per guardrail, both passing and failing inputs
Deliverable: SQLValidator enforces all guardrails. No SQL can reach Athena without passing through it.
The agentic loop is a first-class module, not wired ad-hoc inside main.py.
agent/prompts.py: all prompts in one place, reviewed and tuned independently of code:- System prompt: includes the full pre-loaded Gold schema dict from Phase 3, guardrail rules, and output format expectations. Because all schemas are embedded here, Claude can answer most questions in a single non-tool-call response.
- SQL generation prompt: question + schema context → SELECT query + assumptions list
- Insight prompt: question + SQL + result sample → 2-3 sentence business-language insight
- Tool definitions:
get_schema(for edge cases where Claude needs to re-examine one table)
agent/claude_client.py:ClaudeClient:- For the common case (question maps clearly to one Gold table): single Claude call, no tool use needed. Claude reads the schema from the system prompt and returns SQL + assumptions directly.
- For edge cases (ambiguous question, needs to re-examine a specific table): handles
tool_usecontent blocks, dispatchesget_schema, sendstool_resultback, repeats until text response. - Retries on transient errors (throttling, timeout) with exponential backoff.
- Hard fails immediately on semantic errors (table not found, permission denied) with no retry.
tests/test_claude_client.py: mocked Anthropic SDK responses covering single-turn (common case), tool-use fallback, and retry scenarios.
Deliverable: ClaudeClient handles both the single-call common path and the tool-use fallback path correctly. Retry behaviour tested.
Gold queries are simple: SELECT from one pre-aggregated table with optional WHERE filters. A second review pass designed for complex JOINs adds latency and tokens with no benefit here. Single-pass generation with validation feedback is the right design.
agent/generator.py:SQLGenerator:- Calls
ClaudeClientwith the question. Claude reads the schema from the system prompt and returns a SELECT query and a list of assumptions. - Runs the result through
SQLValidator. - If validation fails, sends the error reason back to Claude and asks for a corrected query. Up to 3 attempts before raising
SQLValidationErrorto the user. - No second review pass: Gold SQL is simple enough that sqlparse guardrail validation is sufficient.
- Returns validated SQL and flagged assumptions.
- Calls
tests/test_generator.py: mockedClaudeClient, tests the validation feedback loop, tests assumption extraction.
Deliverable: SQLGenerator handles validation failures gracefully and recovers automatically. Single Claude call in the common case.
Gold tables are small pre-aggregated tables. The worst-case scan cost for any Gold query in a dev environment is a fraction of a cent: complex pre-execution cost estimation via Glue partition enumeration is unnecessary overhead. The Athena WorkGroup bytes_scanned_cutoff_per_query setting (configured in the Terraform processing module) is the hard cost backstop. Actual cost is tracked post-execution from the Athena result metadata and recorded in the audit log.
agent/executor.py:AthenaExecutor:execute(sql): starts the Athena query, polls until complete, reads the result CSV from the S3 athena-results bucket.- Reads
Statistics.DataScannedInBytesfrom the completed query execution and converts to USD (~$5 per TB scanned). - Returns a pandas DataFrame, actual bytes scanned, and actual cost in USD.
- Retries on transient Athena errors (throttling, internal service error). Fails immediately on query errors (syntax, permission) with no retry.
agent/cost.py: lightweight utility: one function that convertsDataScannedInBytesto USD. No Glue calls, no S3 enumeration.tests/test_executor.py: mocked Athena start/poll/result cycle, failure handling, cost calculation tested.
Deliverable: Full Athena execution path works correctly. Actual cost tracked per query from execution metadata.
agent/result_validator.py:ResultValidator: checks numeric values within plausible bounds (negative revenue is flagged), checks for unexpected nulls on key columns. Zero rows is a valid result for Gold tables: an aggregation with no matching data is a legitimate answer, not a bug. Returns a list of flags, never blocks execution, always surfaces flags in output.agent/insight.py:InsightGenerator: final Claude call that takes the original question, SQL, result DataFrame, and assumptions, and returns a 2-3 sentence business-language insight. Uses the insight prompt fromprompts.py. Structured output so malformed responses raiseAgentError, not crash.agent/audit.py:AuditLogger: writes a structured JSON record tos3://{bronze_bucket}/metadata/agent-audit/after every query. Fields: question, SQL, assumptions, row count, bytes scanned, cost in USD, validation flags, insight, timestamp. The audit log is itself queryable via Athena.tests/test_result_validator.py,tests/test_insight.py
agent/main.py: orchestrates the complete reasoning chain: load schemas → generate SQL → validate → execute → validate results → generate insight → audit log → return output. Handles errors at each stage with clear user-facing messages. CLI entry point:python -m agent.main "question".tests/test_integration.py: marked@pytest.mark.integration, runs against the real AWS dev environment, not mocks. Run manually before deploy, not in CI.
Deliverable: python -m agent.main "Show total orders by country" returns SQL, result table, flagged assumptions, and a 2-sentence insight against live Athena data in under 25 seconds.
agent/charts.py:ChartGenerator:- Detects chart type from data shape and question keywords (five types supported)
- Uploads PNG to
s3://{gold_bucket}/charts/, returns presigned URL (valid 1 hour) - Plotly interactive HTML version returned in the HTTP endpoint response
- See the chart type reference below for which question phrasing triggers which chart
- FastAPI route added to
agent/main.py: POST/askaccepts{"question": "...", "session_id": "..."}, returns full JSON response: insight, assumptions, validation flags, execution ID, bytes scanned, cost in USD, session ID, chart type, presigned chart URL, interactive HTML chart agent/session.py:SessionStorewith TTL eviction (1 hour default).Conversation.context_summary()returns the last 5 turns formatted for Claude, enabling multi-turn follow-ups- GET
/healthreturns{"status": "ok"}: used by the ALB target group health check
terraform-platform-infra-live/modules/analytics-agent/main.tfextended with: internal ALB in private subnets, ALB security group (port 80), ECS security group (port 8080 from ALB only), target group with/healthcheck, ECS service with rolling deploy andlifecycle.ignore_changes.github/workflows/ci.yml: quality gate (ruff, mypy) + unit tests + Docker build check.github/workflows/deploy.yml: OIDC (OpenID Connect) authentication, ECR push, ECS task definition update, rolling deploy with stability wait
A Streamlit browser app (ui/app.py) that wraps the FastAPI backend so non-technical stakeholders can query the agent without touching a command line.
What was built:
ui/app.py: numbered Q&A cards (not chat bubbles), session state for multi-turn follow-ups, bordered card layout- Streaming insight display. The business-language insight streams token-by-token as Claude generates it, so stakeholders see progress immediately instead of waiting for the full response.
- Status badges. A spinner and coloured status badge show each pipeline step in real time (generating SQL, running query, generating insight).
- Chart/Table tabs. Every result renders both an interactive Plotly chart and a raw data table via
st.tabs. The user switches between them without re-running the query. - Query intent check. Before Athena runs, Claude first reads only the SQL and infers its business intent. The original question is withheld from that call except for a language hint. A separate Claude verdict call compares the original question with the inferred SQL intent. If a mismatch is detected, FastAPI regenerates the SQL once with correction feedback before any query runs. The final verdict appears as a badge in the Details expander so stakeholders can see whether a correction was applied.
- Multilingual interface. Eight script-based translation dictionaries (Chinese, Japanese, Korean, Arabic, Russian, Greek, Hebrew, Thai) plus Claude-based language detection for Latin-script languages (Italian, French, Spanish, English, and others). All UI labels, headings, placeholders, and buttons translate automatically.
- Conversation export and import. A sidebar button downloads the conversation as JSON. A file uploader restores it in a later session.
- Session sidebar. Displays session start time and running question count.
entrypoint.shstarts uvicorn on port 8080 (background) then Streamlit on port 8501 (foreground). Both run in the same ECS Fargate container.- The internet-facing ALB has separate listeners: port 80 routes to FastAPI, port 8501 routes to Streamlit.
- Deployed automatically via the existing CI/deploy pipeline on every push to
main.
Reliability, auditability, and a professional PDF output for non-technical stakeholders.
What was built:
- Stakeholder PDF. A 2-page report generated by fpdf2. Page 1 has KPI tiles (key numbers pulled from the result), the business-language insight, and the chart. Page 2 has the methodology in business terms (no SQL, no table names, no technical identifiers). Army olive brand palette (
#4B5320), geometric E logo mark, olive accent bars, CJK font support. The PDF is safe to hand to a non-technical executive. - Engineer audit log. A 17-column CSV written to
s3://{bronze_bucket}/metadata/engineer-log/date={date}/session={session_id}/{request_id}.csvfor every request. Columns:session_id,request_id,timestamp_utc,question_asked,sql_executed,claude_interpretation,discrepancy_detail,verdict,bytes_scanned,athena_cost_usd,response_time_seconds,athena_query_execution_id,sql_retry_count,row_count_returned,chart_type_rendered,language,prompt_version. Theprompt_versioncolumn (e.g.v1) lets me filter the log by prompt version to compare verdict rates before and after a prompt change. A "Prepare Session Log" button in the sidebar fetches and merges all session rows from S3 into a single downloadable CSV. - Verdict computation. The pre-Athena intent check writes a
verdictfield (Yes/No) and adiscrepancy_detailsentence to the response, audit log, and engineer log. If verdict isYes, the SQL is regenerated once before Athena runs. The UI shows the final verdict as a badge in the Details expander. - Circuit breaker. A 30-second timeout on every Claude Platform on AWS call. Up to 3 attempts with 2s/5s/10s exponential backoff on transient errors (throttling, timeout). Semantic errors (table not found, permission denied) fail immediately with no retry.
- Rate limiter. 10 requests per 60-second sliding window per
session_id. Excess requests return HTTP 429 with aretry_after_secondsfield. Implemented withcollections.dequein-process: no Redis needed. - Request UUID tracing. Every request gets a
request_id(UUID v4) at the top of the handler. It flows through the audit log, engineer log, and JSON export.
The Gold layer is pre-aggregated. Each table directly answers a specific business question with 5-10 columns and tens to low hundreds of rows. All 7 Gold schemas (~2,500 tokens total) are loaded at startup and embedded in the system prompt, so Claude knows every table before it sees the first question. This eliminates the multi-turn schema resolution loop and is the single biggest design decision affecting latency and cost.
| Step | Time |
|---|---|
| Claude call 1: schema already in prompt, generate SQL + assumptions | 6-10s |
| SQL validation (local, sqlparse) | <0.1s |
| Claude call 2: infer business intent from SQL only | 1-2s |
| Claude call 3: compare original question vs inferred SQL intent | 1-2s |
| Athena execution on small Gold table | <2s |
| Result validation (local, pandas) | <0.1s |
| Claude call 4: insight generation | 3-5s |
| Chart generation + S3 upload | 1-2s |
| Audit log + engineer log write | 0.3s |
| Typical total | 15-23 seconds |
| Component | Input tokens | Output tokens |
|---|---|---|
| System prompt with all Gold schemas | ~2,500 | : |
| User question | ~30 | : |
| SQL + assumptions | : | ~200 |
| Insight prompt + question + result sample | ~700 | ~150 |
| Verdict (infer question from SQL, compare) | ~250 | ~60 |
| Total per question | ~3,480 | ~410 |
Claude-sonnet-4-6 pricing: $3.00 per million input tokens, $15.00 per million output tokens.
| Component | Cost |
|---|---|
| Claude Platform on AWS (~3,480 input + ~410 output tokens) | ~$0.017 |
| Athena scan (Gold table, <5 MB) | <$0.001 |
| S3 operations (audit log, chart upload) | <$0.001 |
| Total per question | ~$0.018 |
| Component | Per session cost |
|---|---|
| ECS Fargate (0.5 vCPU, 1 GB, 3 hours) | ~$0.08 |
| Claude Platform on AWS (50 questions × ~$0.018) | ~$0.90 |
| Athena (50 queries, <5 MB each) | ~$0.001 |
| S3 (audit logs, chart PNGs) | ~$0.01 |
| ALB (3 hours) | ~$0.05 |
| Total per session | ~$1.04 |
Claude Platform on AWS cost dominates. Athena cost on Gold tables is negligible. The pre-aggregated Gold layer cuts both response time and Claude token usage roughly in half compared to querying Silver directly.
Question: "Show me monthly transaction volume for Berlin over the last 12 months"
Agent output:
Interpretation (please confirm before I proceed):
Table: gold.monthly_revenue_trend (via fct_orders)
Filter: city = 'Berlin', order_date between 2024-04-01 and 2025-03-31
Metric: COUNT(DISTINCT order_id) grouped by year_month
Note: 'transactions' interpreted as placed orders with status = 'completed'
SQL executed:
SELECT
date_trunc('month', order_date) AS month,
COUNT(DISTINCT order_id) AS transaction_volume
FROM edp_dev_gold.fct_orders
WHERE city = 'Berlin'
AND order_date >= DATE '2024-04-01'
AND order_date < DATE '2025-04-01'
AND status = 'completed'
GROUP BY 1
ORDER BY 1
LIMIT 1000
Result: 12 rows returned
Bytes scanned: 4.3 MB | Cost: $0.000022
Insight:
Berlin completed orders peaked in November 2024 at 1,847 transactions,
driven by seasonal demand. Volume has been broadly stable through Q1 2025
at around 1,400 to 1,500 monthly transactions, roughly 12% above the same
period in 2024.
Chart: [interactive Plotly line chart rendered inline]
| Tool | What it does |
|---|---|
| Python 3.11.8 | Agent runtime |
| Claude Platform on AWS (claude-sonnet-4-6) | Question interpretation, SQL generation, insight summarisation |
| boto3 | AWS SDK: Athena, Glue Catalog, S3, SSM |
| sqlparse | SQL parsing and validation |
| FastAPI | HTTP endpoint for the agent API |
| Streamlit | Python library that turns Python code into a browser-accessible web page: the stakeholder UI |
| matplotlib | Static chart PNG generation |
| Plotly | Interactive chart HTML generation |
| fpdf2 | PDF report generation (2-page layout, CJK font support) |
| ECS Fargate | Serverless container runtime |
| Amazon Athena | Executes generated SQL against Gold S3 data |
| AWS Glue Data Catalog | Live physical schema: column names, types, partition keys |
| dbt catalog.json | Business schema context: descriptions, accepted values, documentation |
| pytest | Unit and integration testing |
| ruff | Python linting |
| mypy | Static type checking |
| Docker | Local development and CI builds |
platform-analytics-agent/
├── agent/ ← Python agent source code
│ ├── main.py ← CLI entry point and FastAPI app
│ ├── config.py ← frozen dataclasses, env var validation, fail fast on missing vars
│ ├── exceptions.py ← named exception hierarchy
│ ├── logging.py ← structured JSON logger used by every module
│ ├── prompts.py ← all Claude prompts in one place: system prompt (with schemas), insight
│ ├── claude_client.py ← Claude client: Claude Platform on AWS path, tool-use fallback, retry
│ ├── schema.py ← schema resolver: load_all_schemas() at startup, Glue + dbt catalog.json
│ ├── validator.py ← SQL validator: sqlparse guardrail rules, SELECT-only, Gold DB only
│ ├── generator.py ← SQL generator: single-pass, validation feedback loop (3 attempts)
│ ├── cost.py ← lightweight utility: converts DataScannedInBytes to USD
│ ├── executor.py ← Athena SDK: execute, poll, read results from S3, record actual cost
│ ├── result_validator.py ← result sanity checks: numeric bounds, null rates (zero rows is valid)
│ ├── insight.py ← insight generator: final Claude call, structured output
│ ├── charts.py ← matplotlib PNG and Plotly HTML chart generation
│ ├── session.py ← SessionStore + Conversation: multi-turn context management
│ └── audit.py ← structured JSON audit log writer to S3
│ (no infra/ directory: all AWS infrastructure lives in terraform-platform-infra-live)
├── tests/ ← pytest unit and integration tests
│ ├── conftest.py ← shared fixtures: mocked boto3 clients, mocked Claude responses
│ ├── test_config.py
│ ├── test_exceptions.py
│ ├── test_schema.py
│ ├── test_validator.py
│ ├── test_claude_client.py
│ ├── test_generator.py
│ ├── test_executor.py ← includes cost conversion tests
│ ├── test_result_validator.py
│ ├── test_insight.py
│ ├── test_session.py
│ ├── test_charts.py
│ └── test_integration.py ← marked @pytest.mark.integration, runs against real AWS dev
├── ui/ ← Streamlit browser UI (Phase 13)
│ └── app.py ← the web page: question input, insight display, chart rendering
├── .python-version ← 3.11.8 (pyenv)
├── pyproject.toml ← ruff, mypy, pytest config
├── requirements.txt ← runtime dependencies
├── requirements-dev.txt ← dev tools: ruff, mypy, pytest
├── entrypoint.sh ← starts FastAPI (port 8080) and Streamlit (port 8501) together
├── Dockerfile ← two-stage build, non-root user
├── docker-compose.yml ← local dev
├── Makefile ← setup, lint, typecheck, test, run
└── README.md ← this file
The agent calls real AWS services (S3, Athena, SSM, Glue) and Claude Platform on AWS. If the tests made those real calls on every push, they would cost money, be slow, and require valid credentials in CI. Instead, the tests intercept those calls and return fake responses from memory. No real network traffic happens at all.
Two tools make this possible.
moto pretends to be AWS. When the application code calls boto3.client("s3").put_object(...), moto catches that call and stores the data in a Python dictionary in RAM. The code never knows it isn't talking to real S3. When the test finishes, everything is discarded.
unittest.mock replaces any Python function with a fake version. When the code calls Claude Platform on AWS to generate SQL, mock substitutes that function with one that instantly returns a hardcoded string like SELECT * FROM revenue_by_country LIMIT 10. No real inference request is sent.
There are no CSV test data files because the agent doesn't read CSV files: it reads AWS API responses and Claude Platform on AWS responses. Mocking those responses directly is more accurate than representing them as CSV, and the mocks stay in sync with the code automatically.
flowchart TD
A[Push to GitHub] -->|triggers CI| B[Four jobs run in parallel\nall must pass before deploy]
B --> C[Lint\nruff checks code style]
B --> D[Type check\nmypy checks type annotations]
B --> E[Unit tests\nmoto replaces AWS API calls\nunittest.mock replaces Claude Platform on AWS\nno real network traffic or cost]
B --> F[Docker build\nverifies image builds cleanly\nno push yet]
C --> G{All four pass?}
D --> G
E --> G
F --> G
G -->|No: fix the failure| H[Pipeline stops here\nDeploy workflow is blocked]
G -->|Yes: safe to deploy| I[Deploy workflow triggers\nBuilds and pushes Docker image to ECR\nUpdates ECS task definition\nECS rolling deploy: old tasks replaced\nwith new ones one at a time]
Integration tests also exist but are not part of the standard CI run. They are marked @pytest.mark.integration and only run when explicitly triggered with real AWS credentials against the deployed dev environment. They validate the full pipeline end-to-end: real Glue schema loading, real Athena query, real Claude Platform on AWS call.
All 14 phases complete. The full agent is deployed to ECS Fargate on AWS dev and end-to-end tested. Non-technical stakeholders open a browser, type a business-language question in any supported language, and see a streaming business-language insight, an interactive Plotly chart, numbered Q&A cards, assumptions, a query intent cross-check with verdict badge, and scan cost. PDF reports can be downloaded directly from the browser (2-page stakeholder layout, olive brand, KPI tiles, no SQL or technical identifiers, CJK font support). Engineers can download a session audit CSV from the sidebar.
| Phase | Status |
|---|---|
| 1: Foundation (skeleton, CI, Docker, exceptions, config, logging) | Complete |
| 2: IAM and infra (ECR, ECS cluster, task role, task definition) | Complete |
| 3: Schema resolver (Glue + dbt catalog.json → system prompt) | Complete |
| 4: SQL validator (sqlparse guardrails, SELECT-only, Gold DB only) | Complete |
| 5: Prompts and Claude client (single-call path, tool-use fallback, retry) | Complete |
| 6: SQL generator with feedback loop (3-attempt validation retry) | Complete |
| 7: Athena executor and cost tracking (poll, read results, DataScannedInBytes) | Complete |
| 8: Result validator, insight generator, audit log | Complete |
| 9: CLI entry point and end-to-end integration | Complete |
| 10: Charts (matplotlib PNG, Plotly HTML, S3 presigned URL) | Complete |
| 11: FastAPI HTTP endpoint and session state (multi-turn follow-ups) | Complete |
| 12: Deploy pipeline (OIDC, ECR push, ECS rolling deploy, ALB, ECS service) | Complete |
| 13: Streamlit UI (browser interface, same-container deploy, ALB port 8501) | Complete |
| 14: Enterprise hardening (engineer log, verdict, rate limiter, circuit breaker, stakeholder PDF) | Complete |
This is the last component of the platform. The full pipeline is: PostgreSQL → DMS CDC → Bronze S3 → Glue PySpark → Silver S3 → dbt/Athena → Gold S3 → Redshift Serverless → this agent.
Full platform overview: platform-docs has the complete build guide, architecture diagrams, design decisions, and step-by-step instructions for deploying the entire platform from scratch.
Two overlapping safeguards protect the platform from Claude Platform on AWS instability and runaway usage.
E3: Circuit breaker. Every Claude Platform on AWS call in agent/claude_client.py has a hard 30-second timeout. If the call times out or hits a transient error (rate limit, connection reset), the client retries up to three times with 2s, 5s, and 10s backoff. Permanent errors (authentication failures, malformed requests) propagate immediately without retry. If all attempts fail, the endpoint returns a clean error message to the user. The ECS container never hangs waiting for an unresponsive inference service.
E4: Rate limiting. agent/main.py enforces 10 requests per 60-second sliding window per session_id. Requests over the limit return HTTP 429 (Too Many Requests) with a message explaining when the next request will be accepted. The limiter uses an in-memory dictionary keyed by session ID, which is sufficient for a single-container deployment. At approximately $0.016 per question (Claude Platform on AWS plus Athena scan), uncontrolled request volume is the only runtime cost that can grow unexpectedly.
O3: Request tracing. Every /ask request generates a UUID (Universally Unique Identifier) at the top of the handler in agent/main.py. That request_id appears in:
- The structured engineer log CSV written to
s3://{bronze}/metadata/engineer-log/date=.../session=.../{request_id}.csv - The audit log entry written to
s3://{bronze}/metadata/agent-audit/ - All print statements inside the request handler
When a question returns a wrong answer, I can search for the request ID across FastAPI logs, the S3 audit trail, and Athena query history to see exactly what happened at each step without guessing.
- SQL guardrails: generated SQL is checked before execution. SELECT only, Gold database only, LIMIT always injected if absent, DDL keywords (
DROP,CREATE,INSERT,UPDATE,DELETE,ALTER) rejected. A SQL injection attempt cannot escape these constraints. - IAM read-only: the ECS task role has
athena:StartQueryExecutionands3:GetObjecton Gold only, plus write access to the audit log and Athena results paths. It cannot read Silver or Bronze, and cannot write anywhere except those two paths. - No credentials in the image: AWS credentials come from the ECS task IAM role via the instance metadata service. The Docker image contains no keys, tokens, or secrets.
Not yet implemented: Request queuing (S2). The /ask endpoint currently handles requests synchronously. A question holds a FastAPI worker thread open for the 5-15 seconds it takes Claude and Athena to respond. Under concurrent load, worker threads fill up and new requests receive connection errors rather than queuing. An SQS (Simple Queue Service) queue in front of /ask would accept the request immediately, return a job ID, and let the Streamlit UI poll for the result. Traffic spikes would produce a short queue backlog rather than dropped requests.