I always ask myself what I want to become in the future. The AI era came too fast and I felt both fear (am I going to be redundant?) and unlimited possibility (now I have all the knowledge, the skills, and the appropriate time to achieve whatever I want). I will start with the possibilities and I will finish with my fears. I always wanted to develop software that can reduce carbon footprint and cost on AWS. In one of my previous companies my focus was the cost optimization pillar. At first I thought it was not interesting, but from a stakeholder's perspective money always matters (well, this is not always true — some companies would never care about the costs). Years went by and now I think I am mature enough to develop something that could actually save costs. I focused on Strands SDK because it is a model-driven approach to agent architecture. It mimics how you do OOP, or how you model the solution for a difficult problem — you define the problem, you select the inputs, and you understand the connections between the problem, the inputs, and the final solution. With experience I understood that selecting and writing the right tools for a single agent could lead to better results than reaching for a swarm architecture right away. Contents The model: configuration as the single source of truth The session: one object, every analyzer reads from it Cost Explorer wrappers: the most expensive API in AWS Bootstrap: let the spend tell you which regions matter The cost layer: trends, anomalies, commitments Compute: where Strands' model-driven approach earns its keep Storage: the single biggest hidden cost Databases and network: same shape, different problem Logs, KMS, secrets: small line items that compound Security tooling and governance: paying for what you don't have The synthesis tool: one PDF, one call, at the end Wiring the agent: the most underrated part of Strands CLI: same code, your account or somebody else's PDF renderer Fears Below is the example. The model: configuration as the single source of truth The first decision was where the agent's knobs live. I have seen too many AWS scripts where the threshold for "idle" is hardcoded somewhere on line 437, and the lookback window is hardcoded somewhere else, and you have to read the entire file to figure out what the script considers waste. So I started with a single CONFIG dict, fed by environment variables, and I pinned the small set of constants that the analyzers reference (the EOL Lambda runtimes, the Cost Explorer filter that strips out credits and refunds). The same code now runs in Lambda, in CI, or on my laptop — only the env vars change. from future import annotations import argparse import os import sys import traceback from datetime import date , datetime , timedelta , timezone from pathlib import Path from typing import Any , Callable import boto3 from botocore.exceptions import ClientError from strands import Agent , tool from strands.models.bedrock import BedrockModel import pdf_renderer CONFIG = { " model " : os . getenv ( " FINOPS_BEDROCK_MODEL " , " us.anthropic.claude-sonnet-4-5-20250929-v1:0 " ), " region " : os . getenv ( " AWS_REGION " , " us-east-1 " ), " output_dir " : Path ( os . getenv ( " FINOPS_OUTPUT_DIR " , str ( Path . home () / " Documents/finops-reports " ))), " lookback_months " : int ( os . getenv ( " FINOPS_LOOKBACK_MONTHS " , " 6 " )), " delta_usd " : float ( os . getenv ( " FINOPS_DELTA_USD " , " 5 " )), " delta_pct " : float ( os . getenv ( " FINOPS_DELTA_PCT " , " 25 " )), " stale_days " : int ( os . getenv ( " FINOPS_STALE_DAYS " , " 90 " )), " idle_days " : int ( os . getenv ( " FINOPS_IDLE_DAYS " , " 14 " )), " max_regions " : int ( os . getenv ( " FINOPS_MAX_REGIONS " , " 8 " )), " sns_topic_arn " : os . getenv ( " FINOPS_SNS_TOPIC_ARN " ), } NO_CREDITS = { " Not " : { " Dimensions " : { " Key " : " RECORD_TYPE " , " Values " : [ " Credit " , " Refund " ]}}} EOL_LAMBDA_RUNTIMES = { # Python " python2.7 " , " python3.6 " , " python3.7 " , " python3.8 " , " python3.9 " , # Node.js " nodejs4.3 " , " nodejs6.10 " , " nodejs8.10 " , " nodejs10.x " , " nodejs12.x " , " nodejs14.x " , " nodejs16.x " , " nodejs18.x " , # Ruby " ruby2.5 " , " ruby2.6 " , " ruby2.7 " , # Go (now use provided.al2/al2023) " go1.x " , # Java " java8 " , " java8.al2 " , " java11 " , # .NET " dotnetcore1.0 " , " dotnetcore2.0 " , " dotnetcore2.1 " , " dotnetcore3.1 " , " dotnet5.0 " , " dotnet6 " , " dotnet7 " , # Custom " provided " , } The session: one object, every analyzer reads from it If I follow the OOP analogy, the Session is my object model. It holds the boto3 session, the account identity, the active regions, and a lazy client cache so I do not pay the cost of constructing a new boto3 client for every region in every tool. The two globals ( SESSION and FINDINGS ) are deliberate — every @tool reads from s() and writes through record() , and that is the only state the agent shares. I tried passing the session through tool arguments first, but Strands tools are called by the model, not by my code, and I do not want the model to invent session arguments. The helpers below — try_aws , paginate , per_region , cw_stats — are the boring infrastructure that lets every analyzer survive a region with broken IAM and keep going instead of crashing the whole run. class Session : """ Bundles credentials, account identity, region list, and a client cache. """ def init ( self , boto : boto3 . Session , account_id : str , alias : str | None , primary_region : str , regions : list [ str ]): self . boto = boto self . account_id = account_id self . alias = alias self . primary_region = primary_region self . regions = regions self . _clients : dict [ tuple [ str , str ], Any ] = {} def client ( self , service : str , region : str | None = None ): key = ( service , region or self . primary_region ) if key not in self . _clients : self . _clients [ key ] = self . boto . client ( service , region_name = key [ 1 ]) return self . _clients [ key ] SESSION : Session | None = None FINDINGS : dict [ str , dict ] = {} def s () -> Session : if SESSION is None : raise RuntimeError ( " Session not initialized — call build_session() first " ) return SESSION def record ( section : str , data : dict ) -> dict : """ Store findings under section AND return them. Each tool ends with this. """ FINDINGS [ section ] = data return data def today () -> date : return datetime . now ( timezone . utc ). date () def money ( x ) -> float : try : return round ( float ( x or 0 ), 4 ) except ( TypeError , ValueError ): return 0.0 def try_aws ( fn : Callable , default ): """ Run fn(); on any AWS/runtime error log to stderr and return default.
Lets tools keep going even if some regions/services lack permission. """ try : return fn () except Exception as e : print ( f " [finops-agent] { type ( e ). name } : { e } " , file = sys . stderr ) return default def paginate ( client , op : str , key : str , ** kw ) -> list : items : list = [] for page in client . get_paginator ( op ). paginate ( ** kw ): items . extend ( page . get ( key , [])) return items def per_region ( service : str , fn : Callable [[ str , Any ], dict ]) -> dict [ str , dict ]: """ Call fn(region, client_for_service) for each active region, swallow errors,
return {region: result}. Tools that scan multi-region use this. """ out = {} for region in s (). regions : client = s (). client ( service , region ) out [ region ] = try_aws ( lambda c = client , r = region : fn ( r , c ), {}) return out def cw_stats ( client , namespace : str , metric : str , dims : list [ dict ], days : int , stat : str = " Average " , period : int = 86400 ) -> list [ float ]: end = datetime . now ( timezone . utc ) start = end - timedelta ( days = days ) r = try_aws ( lambda : client . get_metric_statistics ( Namespace = namespace , MetricName = metric , Dimensions = dims , StartTime = start , EndTime = end , Period = period , Statistics = [ stat ], ), { " Datapoints " : []}, ) return [ d [ stat ] for d in r . get ( " Datapoints " , [])] Cost Explorer wrappers: the most expensive API in AWS Cost Explorer is the most opinionated API in AWS — and one of the few where every call costs you money (0 spend in the last 30 days. Falls back to [primary]. """ end = date . today () + timedelta ( days = 1 ) start = end - timedelta ( days = 31 ) r = try_aws ( lambda : sess . client ( " ce " , region_name = " us-east-1 " ). get_cost_and_usage ( TimePeriod = { " Start " : start . isoformat (), " End " : end . isoformat ()}, Granularity = " MONTHLY " , Metrics = [ " UnblendedCost " ], GroupBy = [{ " Type " : " DIMENSION " , " Key " : " REGION " }], Filter = NO_CREDITS , ), { " ResultsByTime " : []}, ) pairs : list [ tuple [ str , float ]] = [] for tp in r . get ( " ResultsByTime " , []): for g in tp . get ( " Groups " , []): rg , cost = g [ " Keys " ][ 0 ], money ( g [ " Metrics " ][ " UnblendedCost " ][ " Amount " ]) if cost > 0 and rg and rg not in ( " NoRegion " , " global " ): pairs . append (( rg , cost )) pairs . sort ( key = lambda x : - x [ 1 ]) out = [ r for r , _ in pairs ] if primary not in out : out . insert ( 0 , primary ) return out or [ primary ] The cost layer: trends, anomalies, commitments These are the first four tools the agent will call. discover_account always runs first — it anchors the account identity in the conversation and tells the agent "here is what you are looking at." Then analyze_cost_trends builds the spend profile and the 7-day-vs-prior-7-day deltas, which is where most "why is the bill suddenly higher?" questions are answered. analyze_anomalies_and_budgets checks whether the account even has Cost Anomaly Detection or Budgets configured — most accounts I have looked at do not, and that itself is a finding. analyze_commitments looks at Reserved Instance and Savings Plans coverage, but it includes a guardrail recommendation: skip RI/SP if compute spend is below $50/month. Telling a hobby account to buy a 1-year commitment is the kind of advice that makes people stop trusting the agent. # TOOLS @tool def discover_account () -> dict : """ First tool to call. Returns account identity, alias, organization status,
and the regions that have non-zero spend. No arguments. """ sess = s () org = try_aws ( lambda : sess . client ( " organizations " ). describe_organization (). get ( " Organization " , {}), {}, ) return record ( " account " , { " account_id " : sess . account_id , " account_alias " : sess . alias , " primary_region " : sess . primary_region , " active_regions " : sess . regions , " is_payer_or_member " : bool ( org . get ( " Id " )), " master_account_id " : org . get ( " MasterAccountId " ), }) @tool def analyze_cost_trends () -> dict : """ Last N months by service + 7d-vs-prior-7d service deltas + current-month forecast. """ months = _ce_monthly_with_top ( CONFIG [ " lookback_months " ]) end = today () + timedelta ( days = 1 ) cur = _ce_groups ( today () - timedelta ( days = 7 ), end , " SERVICE " ) prior = _ce_groups ( today () - timedelta ( days = 14 ), today () - timedelta ( days = 7 ), " SERVICE " ) deltas = [] for svc in set ( cur ) | set ( prior ): c , p = cur . get ( svc , 0.0 ), prior . get ( svc , 0.0 ) delta = c - p pct = ( delta / p * 100 ) if p > 0 else ( 100.0 if c > 0 else 0.0 ) flagged = abs ( delta ) > CONFIG [ " delta_usd " ] or abs ( pct ) > CONFIG [ " delta_pct " ] if flagged or abs ( delta ) >= 0.5 : deltas . append ({ " service " : svc , " current_7d " : round ( c , 2 ), " prior_7d " : round ( p , 2 ), " delta_usd " : round ( delta , 2 ), " delta_pct " : round ( pct , 1 ), " flagged " : flagged , }) deltas . sort ( key = lambda x : - abs ( x [ " delta_usd " ])) forecast_resp = try_aws ( lambda : s (). client ( " ce " ). get_cost_forecast ( TimePeriod = { " Start " : today (). isoformat (), " End " : ( today (). replace ( day = 1 ) + timedelta ( days = 32 )). replace ( day = 1 ). isoformat (), }, Granularity = " MONTHLY " , Metric = " UNBLENDED_COST " , ), None , ) forecast = ( round ( money ( forecast_resp [ " Total " ][ " Amount " ]), 2 ) if forecast_resp and " Total " in forecast_resp else None ) return record ( " costs " , { " monthly " : months , " deltas_7d " : deltas [: 15 ], " forecast_current_month_usd " : forecast , " last_month_total_usd " : months [ - 1 ][ " total_usd " ] if months else 0 , " trailing_months " : CONFIG [ " lookback_months " ], }) @tool def analyze_anomalies_and_budgets () -> dict : """ Cost Anomaly Detection (last 90d) + all AWS Budgets state. """ sess = s () anom_resp = try_aws ( lambda : sess . client ( " ce " ). get_anomalies ( DateInterval = { " StartDate " : ( today () - timedelta ( days = 90 )). isoformat (), " EndDate " : today (). isoformat (), }, ), { " Anomalies " : []}, ) anomalies = [ { " id " : a . get ( " AnomalyId " ), " score " : a . get ( " AnomalyScore " , {}). get ( " CurrentScore " ), " impact_usd " : round ( money ( a . get ( " Impact " , {}). get ( " TotalImpact " , 0 )), 2 ), " service " : ( a . get ( " RootCauses " ) or [{}])[ 0 ]. get ( " Service " ), " start " : a . get ( " AnomalyStartDate " ), " end " : a . get ( " AnomalyEndDate " ), } for a in anom_resp . get ( " Anomalies " , []) ] monitor_count = len ( try_aws ( lambda : sess . client ( " ce " ). get_anomaly_monitors (). get ( " AnomalyMonitors " , []), [])) raw_budgets = try_aws ( lambda : sess . client ( " budgets " ). describe_budgets ( AccountId = sess . account_id ). get ( " Budgets " , []), [], ) budgets = [] for b in raw_budgets : actual = money ( b . get ( " CalculatedSpend " , {}). get ( " ActualSpend " , {}). get ( " Amount " , 0 )) forecast = money ( b . get ( " CalculatedSpend " , {}). get ( " ForecastedSpend " , {}). get ( " Amount " , 0 )) limit = money ( b [ " BudgetLimit " ][ " Amount " ]) budgets . append ({ " name " : b [ " BudgetName " ], " limit_usd " : limit , " actual_usd " : round ( actual , 2 ), " forecast_usd " : round ( forecast , 2 ), " exceeded_actual " : bool ( limit ) and actual > limit , " exceeded_forecast " : bool ( limit ) and forecast > limit , " time_unit " : b . get ( " TimeUnit " ), }) return record ( " anomalies_budgets " , { " anomalies " : anomalies , " anomaly_monitor_count " : monitor_count , " anomaly_detection_configured " : monitor_count > 0 , " budgets " : budgets , " budget_count " : len ( budgets ), }) @tool def analyze_commitments () -> dict : """ RI + Savings Plans coverage and OnDemand exposure for the last 30 days. """ sess = s () start , end = today () - timedelta ( days = 30 ), today () + timedelta ( days = 1 ) sp_cov = try_aws ( lambda : sess . client ( " ce " ). get_savings_plans_coverage ( TimePeriod = { " Start " : start . isoformat (), " End " : end . isoformat ()}, Granularity = " MONTHLY " , ), { " SavingsPlansCoverages " : []}, ) sp_summary = [ { " covered_usd " : money ( c . get ( " Coverage " , {}). get ( " SpendCoveredBySavingsPlans " , 0 )), " ondemand_usd " : money ( c . get ( " Coverage " , {}). get ( " OnDemandCost " , 0 )), " coverage_pct " : money ( c . get ( " Coverage " , {}). get ( " CoveragePercentage " , 0 )), } for c in sp_cov . get ( " SavingsPlansCoverages " , []) ] ri_cov = try_aws ( lambda : sess . client ( " ce " ). get_reservation_coverage ( TimePeriod = { " Start " : start . isoformat (), " End " : end . isoformat ()}, Granularity = " MONTHLY " , ), { " CoveragesByTime " : []}, ) ri_summary = [ { " metric " : k , " value " : v } for tp in ri_cov . get ( " CoveragesByTime " , []) for k , v in tp . get ( " Total " , {}). get ( " CoverageHours " , {}). items () ] ondemand = sum ( x [ " ondemand_usd " ] for x in sp_summary ) rec = ( " skip RI/SP —
Smarter Cloud Spending: FinOps Agent Powered by Strands SDK and Amazon Bedrock
Martin Nanchev