Build an AI Cost Allocation Tagger with Claude API

pip install anthropic boto3

import boto3
import anthropic
import json
 
client = anthropic.Anthropic()
ec2 = boto3.client("ec2")
rds = boto3.client("rds")
 
 
def get_untagged_resources() -> list[dict]:
    """Find resources missing required tags: team, environment, project."""
    required_tags = {"team", "environment", "project"}
    untagged = []
 
    for reservation in ec2.describe_instances()["Reservations"]:
        for instance in reservation["Instances"]:
            tags = {t["Key"].lower(): t["Value"] for t in instance.get("Tags", [])}
            missing = required_tags - set(tags.keys())
            if missing:
                untagged.append({
                    "resource_id": instance["InstanceId"],
                    "resource_type": "ec2_instance",
                    "existing_tags": tags,
                    "missing_tags": list(missing),
                    "name": tags.get("name", ""),
                    "vpc_id": instance.get("VpcId"),
                    "security_groups": [sg["GroupName"] for sg in instance.get("SecurityGroups", [])],
                })
 
    return untagged

def gather_context(resource: dict) -> dict:
    """Pull surrounding signals that hint at ownership — name patterns,
    which VPC/subnet it's in, what security groups it shares with tagged resources."""
    context = {"resource": resource}
 
    # Find tagged resources in the same VPC — they often belong to the same team
    if resource.get("vpc_id"):
        same_vpc = ec2.describe_instances(
            Filters=[{"Name": "vpc-id", "Values": [resource["vpc_id"]]}]
        )
        tagged_neighbors = []
        for reservation in same_vpc["Reservations"]:
            for instance in reservation["Instances"]:
                tags = {t["Key"].lower(): t["Value"] for t in instance.get("Tags", [])}
                if "team" in tags:
                    tagged_neighbors.append({"name": tags.get("name", ""), "team": tags["team"]})
        context["tagged_neighbors_in_vpc"] = tagged_neighbors[:5]
 
    return context

INFER_PROMPT = """Infer the missing tags for this AWS resource based on naming
patterns and surrounding context. Be conservative — only suggest a tag if
you have reasonable evidence, otherwise mark it "uncertain".
 
Resource: {resource_id} ({resource_type})
Existing tags: {existing_tags}
Missing tags: {missing_tags}
Resource name: {name}
Security groups: {security_groups}
 
Tagged resources in the same VPC (evidence of team ownership by proximity):
{tagged_neighbors}
 
Respond with ONLY valid JSON:
{{"inferred_tags": {{"team": "...", "environment": "...", "project": "..."}},
  "confidence": "high"|"medium"|"low",
  "reasoning": "one sentence explaining the inference"}}
 
For any tag you cannot infer with reasonable confidence, use "uncertain" as the value."""
 
 
def infer_tags(context: dict) -> dict:
    resource = context["resource"]
    response = client.messages.create(
        model="claude-sonnet-5",
        max_tokens=400,
        messages=[{
            "role": "user",
            "content": INFER_PROMPT.format(
                resource_id=resource["resource_id"],
                resource_type=resource["resource_type"],
                existing_tags=resource["existing_tags"],
                missing_tags=resource["missing_tags"],
                name=resource["name"],
                security_groups=resource["security_groups"],
                tagged_neighbors=context.get("tagged_neighbors_in_vpc", []),
            )
        }]
    )
    text = response.content[0].text.strip()
    if text.startswith("```"):
        text = text.split("```")[1].replace("json", "", 1).strip()
    return json.loads(text)

def generate_tagging_plan(untagged_resources: list[dict]) -> str:
    """Never apply tags directly — always output a reviewable diff."""
    plan_lines = ["# Proposed tag changes — review before applying\n"]
 
    for resource in untagged_resources:
        context = gather_context(resource)
        inference = infer_tags(context)
 
        if inference["confidence"] == "low":
            plan_lines.append(f"# SKIPPED (low confidence): {resource['resource_id']} — needs manual review")
            continue
 
        plan_lines.append(f"# {resource['resource_id']} ({resource.get('name', 'unnamed')})")
        plan_lines.append(f"# Confidence: {inference['confidence']} — {inference['reasoning']}")
        for key, value in inference["inferred_tags"].items():
            if value != "uncertain":
                plan_lines.append(f"aws ec2 create-tags --resources {resource['resource_id']} --tags Key={key},Value={value}")
        plan_lines.append("")
 
    return "\n".join(plan_lines)

python cost_tagger.py --output tagging-plan.txt
 
# tagging-plan.txt:
# # i-0abc123 (worker-node-payments-3)
# # Confidence: high — name pattern matches 4 other tagged "payments" resources in same VPC
# aws ec2 create-tags --resources i-0abc123 --tags Key=team,Value=payments
# aws ec2 create-tags --resources i-0abc123 --tags Key=environment,Value=production
#
# # SKIPPED (low confidence): i-0xyz789 — needs manual review