Cloud bills spiral when engineering ignores utilization. A FinOps agent AWS cost rightsizing loop closes that gap by continuously matching instance types to actual workload demand. This tutorial builds one with Python, boto3, and an LLM for recommendation synthesis—no vendor console clicking required.
Prerequisites
- AWS credentials with
AmazonEC2ReadOnlyAccessandCloudWatchReadOnlyAccess(use a read-only IAM role; the agent only mutates instances in an explicit apply step). - Python 3.11+ and
pip install boto3 openai. - An API key for an OpenAI-compatible inference endpoint. We’ll route through n4n.ai, a single OpenAI-compatible gateway across 240+ models with automatic fallback if a provider is rate-limited, so the agent stays resilient without code changes.
- Environment variables:
AWS_DEFAULT_REGION,N4N_API_KEY.
export AWS_DEFAULT_REGION=us-east-1
export N4N_API_KEY=sk-...
Step 1: Collect EC2 utilization data
CloudWatch holds the signal. CPU is native; memory requires the CloudWatch agent, but we’ll keep the example to CPU and add a memory field if you have it.
import boto3
from datetime import datetime, timedelta
def get_avg_cpu(instance_id: str, days: int = 14) -> float | None:
cw = boto3.client("cloudwatch")
end = datetime.utcnow()
start = end - timedelta(days=days)
resp = cw.get_metric_statistics(
Namespace="AWS/EC2",
MetricName="CPUUtilization",
Dimensions=[{"Name": "InstanceId", "Value": instance_id}],
StartTime=start,
EndTime=end,
Period=86400,
Statistics=["Average"],
)
if not resp["Datapoints"]:
return None
vals = [dp["Average"] for dp in resp["Datapoints"]]
return sum(vals) / len(vals)
Run this for one instance to validate:
print(get_avg_cpu("i-0abc123")) # e.g. 7.32
Expected output: a float percentage or None if the instance launched recently.
Step 2: Pull current instance inventory
We need the running fleet and their shapes.
def list_running_instances():
ec2 = boto3.client("ec2")
resp = ec2.describe_instances(
Filters=[{"Name": "instance-state-name", "Values": ["running"]}]
)
out = []
for res in resp["Reservations"]:
for inst in res["Instances"]:
out.append({
"id": inst["InstanceId"],
"type": inst["InstanceType"],
"launch_time": inst["LaunchTime"].isoformat(),
"tags": {t["Key"]: t["Value"] for t in inst.get("Tags", [])},
})
return out
Step 3: Assemble the analysis payload
Merge inventory with metrics. Keep the payload small; the LLM doesn’t need raw time series.
def build_payload():
instances = list_running_instances()
payload = []
for inst in instances:
cpu = get_avg_cpu(inst["id"])
if cpu is None:
continue
payload.append({
"instance_id": inst["id"],
"current_type": inst["type"],
"avg_cpu_14d": round(cpu, 2),
"tags": inst["tags"],
})
return payload
payload = build_payload()
print(f"Collected {len(payload)} instances")
Expected output: Collected 12 instances (varies with your fleet).
Step 4: Query the FinOps agent for rightsizing recommendations
The LLM turns metrics into concrete instance-type swaps. We constrain output to JSON so it’s machine-parseable.
import os, json
from openai import OpenAI
client = OpenAI(base_url="https://api.n4n.ai/v1", api_key=os.environ["N4N_API_KEY"])
SYSTEM_PROMPT = """You are a senior FinOps engineer. Given EC2 instances with 14-day average CPU, recommend rightsizing.
Rules: if avg_cpu_14d < 10, downsize one family step; if > 80, upsize. Preserve x86/arm arch unless tags say otherwise.
Respond ONLY with JSON: [{"instance_id": str, "current_type": str, "recommended_type": str, "reason": str}]"""
def get_recommendations(payload):
resp = client.chat.completions.create(
model="openai/gpt-4o-mini",
messages=[
{"role": "system", "content": SYSTEM_PROMPT},
{"role": "user", "content": json.dumps(payload, indent=2)},
],
temperature=0.1,
)
return json.loads(resp.choices[0].message.content)
recs = get_recommendations(payload)
The FinOps agent AWS cost rightsizing core is now functional: it emits a list of swaps with rationales.
Step 5: Dry-run and apply
Never resize blindly. Print first, then gate behind a human or a tag policy.
def apply_dry_run(recs):
for r in recs:
print(f"[DRY-RUN] {r['instance_id']}: "
f"{r['current_type']} -> {r['recommended_type']} | {r['reason']}")
apply_dry_run(recs)
Sample output:
[DRY-RUN] i-0abc123: t3.large -> t3.medium | avg_cpu_14d 6.4%, safe downsize
[DRY-RUN] i-0def456: m5.xlarge -> m5.2xlarge | avg_cpu_14d 88.1%, sustained pressure
To actually resize, stop the instance, modify, and start. Keep it in the same step only after you’ve validated the dry-run against your SLA:
def resize_instance(ec2, instance_id, new_type):
ec2.stop_instances(InstanceIds=[instance_id])
waiter = ec2.get_waiter("instance_stopped")
waiter.wait(InstanceIds=[instance_id])
ec2.modify_instance_attribute(
InstanceId=instance_id,
Attribute="instanceType",
Value=new_type,
)
ec2.start_instances(InstanceIds=[instance_id])
Operationalizing the agent
Run this on a schedule. Package as a Lambda with a 300s timeout (CloudWatch calls are the slow part), or a cron job on a small box. Persist recs to S3 or DynamoDB so you can diff week-over-week and detect flapping. Add a Slack webhook that posts the dry-run list; only instances tagged auto-finops=allow get passed to resize_instance.
The FinOps agent AWS cost rightsizing loop pays for itself when it catches one forgotten t3.2xlarge serving a cron job that runs for nine minutes a day. Wire the metrics collection to your existing observability, and let the LLM handle the tedious mapping of utilization to current instance families—its training data knows the SKU matrix better than your on-call does at 3 a.m.