DEV Community: Martin Nanchev

A Layman's Guide to Amazon Bedrock AgentCore: Identity, Gateway, Memory, and Runtime in Terraform

Martin Nanchev — Fri, 31 Jul 2026 20:35:14 +0000

AgentCore Identity: who can do what
AgentCore Gateway: the right tool, no wasted tokens
AgentCore Memory: what the agent remembers
AgentCore Runtime: Firecracker, MicroVMs, and containers
Bonus: cross-account and cross-region without network magic
- Example 1: Runtime target — the network isn't there, because you don't need it
- Example 2: MCP target — semantic search, but you build the network yourself
- Which one, when
What it costs
In lieu of a conclusion

The AgentCore platform builds on more than 20 years of AWS experience solving the problems of Amazon's store and its end customers. The service reshapes many of the ideas we've already built up about compute, network, storage, and databases, and enables connectivity between accounts and between regions without requiring you to know software-defined networking in depth. It also gives you a way to hook legacy systems up to the new agents. The platform consists of several capabilities, which we'll go through one by one: Identity, Gateway, Memory, and Runtime.

AgentCore Identity: who can do what

AgentCore Identity manages the agent's identity and its access to external services and tools through an abstraction called workload identity. The keys for external services are kept in Secrets Manager under the prefix bedrock-agentcore-identity!. In practice, the service manages the identification flow — who wants to do what, and whether it's been approved. That way you guarantee no rogue behavior: the agent has exactly as much access as you've given it.

Two forms of authentication and authorization are on offer.

2LO (2-legged OAuth) — the agent presents itself to the external system with a pre-issued token: something like a pass issued on your behalf that defines exactly what it's allowed to do. The secrets are kept in Secrets Manager. In the iGaming sector, 2LO is something of a taboo to be avoided at all costs — the word PAT stirs up strong emotions in the infosec community, no matter whether the token is encrypted with a KMS key under a strict policy. A bit like "dracarys" around the dragons in Game of Thrones.

3LO (3-legged OAuth) — Here you have three parties: the agent, the user, and a third-party system. The agent by itself has no rights whatsoever in Jira, Confluence, or GitHub. The first time it asks to act on your behalf, the system shows you a consent screen — just like when an app asks whether it may access your camera. The screen states exactly what the agent will be able to do (for example, only read tickets), and when you hit Allow, the system issues a token with only those rights, which AgentCore Identity stores for you. From then on the agent works with that token and can perform only the approved actions — nothing more. For each of the systems (Jira, Confluence, GitHub) the consent is separate, with its own permissions.

A 2LO example:

resource "aws_bedrockagentcore_api_key_credential_provider" "github" {
  name               = "github-pat-token"
  api_key_wo         = local.envs["GITHUB_PAT"]
  api_key_wo_version = var.token_version
}

Identity and keys are only half the story, though — the agent also needs tools it can reach with them. Enter Gateway.

AgentCore Gateway: the right tool, no wasted tokens

AgentCore Gateway lets you connect to Lambda functions, Smithy models, OpenAPI-compatible APIs, and external MCP servers. It also solves a very important problem: when you hand an agent a lot of tools, it often gets it wrong and picks the wrong one. That costs tokens, and the customer pays for them. That's why Gateway uses semantic search to find the right tool.

The Terraform code defines the gateway and its target. The OpenAPI schema can be passed inline in JSON format; if it's in YAML, you convert it with Terraform functions — jsonencode(yamldecode(...)). One quirk: if the target is an AgentCore Runtime agent, the gateway must have no protocol type set — Runtime targets can't be added to a gateway with protocol_type = "MCP".

data "aws_iam_policy_document" "gtw_plcy" {
  statement {
    sid     = "GetResourceOauth2Token"
    effect  = "Allow"
    actions = ["bedrock-agentcore:GetResourceApiKey"]
    resources = ["arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:token-vault/default/apikeycredentialprovider/${aws_bedrockagentcore_api_key_credential_provider.github.name}",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/${aws_bedrockagentcore_gateway.gtw.gateway_id}",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:token-vault/default"
    ]
  }
  statement {
    sid     = "GetWorkloadAccessToken"
    effect  = "Allow"
    actions = ["bedrock-agentcore:GetWorkloadAccessToken"]
    resources = [
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/${aws_bedrockagentcore_gateway.gtw.gateway_id}"
    ]
  }
  statement {
    effect  = "Allow"
    actions = ["secretsmanager:GetSecretValue"]
    resources = [
      aws_bedrockagentcore_api_key_credential_provider.github.api_key_secret_arn[0].secret_arn
    ]
  }
}
data "aws_iam_policy_document" "assume_role" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["bedrock-agentcore.amazonaws.com"]
    }
    condition {
      test     = "StringEquals"
      values   = [data.aws_caller_identity.current.account_id]
      variable = "aws:SourceAccount"
    }
  }
}
resource "aws_iam_role" "gtw_role" {
  name               = "bedrock-agentcore-gateway-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_policy" "gtw_plcy" {
  policy = data.aws_iam_policy_document.gtw_plcy.json
  name   = "bedrock-agentcore-gateway-plcy"
}

resource "aws_iam_role_policy_attachment" "gtw_role_plcy_attach" {
  policy_arn = aws_iam_policy.gtw_plcy.arn
  role       = aws_iam_role.gtw_role.name
}

resource "aws_bedrockagentcore_gateway" "gtw" {
  name            = "gateway"
  role_arn        = aws_iam_role.gtw_role.arn
  authorizer_type = "AWS_IAM"
  protocol_type   = "MCP"
  exception_level = "DEBUG"
  protocol_configuration {
    mcp {
      search_type = "SEMANTIC"
    }
  }
}

resource "aws_bedrockagentcore_gateway_target" "github_target" {
  name               = "api-target"
  gateway_identifier = aws_bedrockagentcore_gateway.gtw.gateway_id
  description        = "External API target with API key authentication"

  credential_provider_configuration {
    api_key {
      provider_arn              = aws_bedrockagentcore_api_key_credential_provider.github.credential_provider_arn
      credential_location       = "HEADER"
      credential_parameter_name = "Authorization"
      credential_prefix         = "Bearer"
    }
  }

  target_configuration {
    mcp {
      open_api_schema {
        inline_payload {
          payload = jsonencode(yamldecode(file("${path.module}/github-tools.yaml")))
        }
      }
    }
  }
}

Sometimes, beyond access to external services, we want the agent to remember information about the end user, so it can offer more valuable solutions based on their context. That's done with AgentCore Memory.

AgentCore Memory: what the agent remembers

Memory is divided by strategy into: user preference (things the user likes), semantic (facts from the context), summarization, and custom. Extraction of the long-term records is asynchronous and takes time. There's also a quirk at creation time: each strategy blocks the creation of the next one, i.e. if you're setting up more than one, you have to wait between them — empirically around 190 seconds; in the code below I've set 300 to be safe. In Terraform it looks like this:

resource "aws_iam_role" "bedrock_memory_role" {
  name               = "bedrock-agentcore-memory-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_role_policy_attachment" "bedrock_plcy_attach" {
  role       = aws_iam_role.bedrock_memory_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonBedrockAgentCoreMemoryBedrockModelInferenceExecutionRolePolicy"
}

resource "aws_bedrockagentcore_memory" "memory" {
  name                      = "agent_memory"
  event_expiry_duration     = 30
  memory_execution_role_arn = aws_iam_role.bedrock_memory_role.arn
}

resource "aws_bedrockagentcore_memory_strategy" "user_preference" {
  name        = "userpreference_strategy"
  memory_id   = aws_bedrockagentcore_memory.memory.id
  type        = "USER_PREFERENCE"
  description = "user preference strategy"
  namespaces  = ["/users/{actorId}/preferences/"]
}

resource "time_sleep" "wait_after_usr_pref_strategy" {
  depends_on       = [aws_bedrockagentcore_memory_strategy.user_preference]
  create_duration  = "300s"
  destroy_duration = "300s"
  triggers = {
    strategy_id = aws_bedrockagentcore_memory_strategy.user_preference.memory_strategy_id
  }
}

resource "aws_bedrockagentcore_memory_strategy" "semantic" {
  name        = "semantic_strategy"
  memory_id   = aws_bedrockagentcore_memory.memory.id
  type        = "SEMANTIC"
  description = "semantic strategy"
  namespaces  = ["/users/{actorId}/facts/"]
  depends_on  = [time_sleep.wait_after_usr_pref_strategy]
}

resource "time_sleep" "wait_after_semantic_strategy" {
  depends_on       = [aws_bedrockagentcore_memory_strategy.semantic]
  create_duration  = "300s"
  destroy_duration = "300s"
  triggers = {
    strategy_id = aws_bedrockagentcore_memory_strategy.semantic.memory_strategy_id
  }
}

resource "aws_bedrockagentcore_memory_strategy" "summarization" {
  name        = "summarization_strategy"
  memory_id   = aws_bedrockagentcore_memory.memory.id
  type        = "SUMMARIZATION"
  description = "summarization strategy"
  namespaces  = ["/summaries/{actorId}/{sessionId}/"]
  depends_on  = [time_sleep.wait_after_semantic_strategy]
}

The records live in hierarchical namespaces, where {actorId} and {sessionId} are substituted with the real values at write time.

Here's how everything so far comes together in the agent itself — reading from long-term memory, talking to the tools through Gateway (MCP with SigV4 signing), and recording the turn as an event:

"""Repo Ops agent — AgentCore Runtime entrypoint.

Flow of every request:
  1. recall()  -> reads long-term memory (LTM) for this user
  2. Agent()   -> LLM conversation with the tools from Gateway (MCP over SigV4)
  3. create_event() -> records the turn as an event (STM); the strategies
                       extract LTM asynchronously after ~1 minute
"""
import os

import boto3
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from bedrock_agentcore.memory import MemoryClient
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient

REGION = os.environ.get("AWS_REGION", "eu-central-1")
GATEWAY_URL = os.environ["GATEWAY_URL"]   # .../mcp
MEMORY_ID = os.environ["MEMORY_ID"]
MODEL_ID = os.environ["MODEL_ID"]

app = BedrockAgentCoreApp()               # starts an HTTP server on :8080
memory = MemoryClient(region_name=REGION)


class SigV4HTTPXAuth(httpx.Auth):
    """Signs every MCP request with SigV4.

    The gateway has authorizer_type = "AWS_IAM", i.e. it expects a
    signature exactly like any other AWS API call. The Runtime role
    must have bedrock-agentcore:InvokeGateway on the gateway's ARN.
    """

    requires_request_body = True

    def __init__(self, service: str, region: str):
        self._creds = boto3.Session().get_credentials()
        self._service = service
        self._region = region

    def auth_flow(self, request: httpx.Request):
        aws_req = AWSRequest(
            method=request.method,
            url=str(request.url),
            data=request.content,
            headers={"host": request.url.host},
        )
        SigV4Auth(self._creds, self._service, self._region).add_auth(aws_req)
        request.headers.update(dict(aws_req.headers))
        yield request


def recall(actor_id: str, query: str) -> str:
    """Semantic search over long-term memory.

    The namespaces must match the ones in Terraform EXACTLY
    (including the trailing slash).
    """
    namespaces = [
        f"/users/{actor_id}/preferences/",
        f"/users/{actor_id}/facts/",
    ]
    chunks = []
    for ns in namespaces:
        try:
            records = memory.retrieve_memories(
                memory_id=MEMORY_ID,
                namespace=ns,
                query=query,
                top_k=3,
            )
            for rec in records:
                text = rec.get("content", {}).get("text", "")
                if text:
                    chunks.append(f"- {text}")
        except Exception as exc:
            # Memory must never take the agent down
            print(f"[memory] retrieve failed for {ns}: {exc}")
    return "\n".join(chunks)


def remember(actor_id: str, session_id: str, prompt: str, answer: str) -> None:
    """Records the turn as an event in short-term memory."""
    try:
        memory.create_event(
            memory_id=MEMORY_ID,
            actor_id=actor_id,
            session_id=session_id,
            messages=[(prompt, "USER"), (answer, "ASSISTANT")],
        )
    except Exception as exc:
        print(f"[memory] create_event failed: {exc}")


@app.entrypoint
def invoke(payload, context):
    prompt = payload.get("prompt", "")
    actor_id = payload.get("actor_id", "default")

    # session_id comes from --runtime-session-id at invocation time
    session_id = getattr(context, "session_id", None) or "local-dev-session-0000000000000000"

    memories = recall(actor_id, prompt)

    system_prompt = (
        "You are Repo Ops — a concise assistant for GitHub operations. "
        "Use the available tools for anything related to repositories, "
        "pull requests, and issues. Never make up data — if a tool "
        "returns an error, tell the user about it. Ask for confirmation "
        "before creating an issue."
    )
    if memories:
        system_prompt += f"\n\nWhat you remember about this user:\n{memories}"

    gateway = MCPClient(
        lambda: streamablehttp_client(
            GATEWAY_URL,
            auth=SigV4HTTPXAuth("bedrock-agentcore", REGION),
        )
    )

    with gateway:
        tools = gateway.list_tools_sync()
        print(f"[gateway] tools: {[t.tool_name for t in tools]}")
        agent = Agent(
            model=BedrockModel(model_id=MODEL_ID, region_name=REGION),
            tools=tools,
            system_prompt=system_prompt,
        )
        answer = str(agent(prompt))

    remember(actor_id, session_id, prompt, answer)
    return {"result": answer}


if __name__ == "__main__":
    app.run()

This code has to live somewhere, though. Time for the part I always freeze on in interviews.

AgentCore Runtime: Firecracker, MicroVMs, and containers

I've always wondered how to explain what AgentCore Runtime is. Questions about MicroVMs, virtual machines, containers, and processes are always hard to talk through. Maybe I don't fully understand it myself; maybe the rabbit hole is so deep that it's hard to explain simply. So I decided to start from a distance — perhaps also because these are exactly the questions I freeze on in interviews.

To bust the myth, I'll start with the MicroVM.

Two of the options for deploying an application (including an AI application) are a virtual machine and a container. The difference: containers package all of the application's dependencies and libraries and share a common kernel and operating system with the host, while each virtual machine has its own kernel, its own operating system, and a dedicated slice of the resources. The kernel manages the hardware through drivers, i.e. it has unrestricted access to CPU time and memory; applications, on the other hand, do not.

It's commonly assumed that a virtual machine is more secure than containers. Why? Perhaps because containers share a kernel with the host.

Whether that makes containers insecure compared to a virtual machine is debatable. With a virtual machine you get patching, driver management, and slower startup. There's also the distroless container option — no shell and no package managers, only what the end application needs.

The MicroVM tries to solve the slow-startup challenge while also offering the stronger isolation characteristic of virtual machines: each machine comes with its own virtualized kernel. Firecracker — the MicroVM technology behind AWS Lambda — lets you launch up to 150 MicroVMs per second, with each one booting in about 125 ms.

So what do virtual machines and containers have to do with AgentCore Runtime? Firecracker sits behind AgentCore Runtime: you deploy a container, and it runs on top of a Firecracker virtual machine. That gives you maximum isolation — one Firecracker machine can't use another's resources, that's forbidden at the hypervisor level — and at the same time the easiest possible deployment of any AI application in the form of a container with all its dependencies. Alongside the Firecracker process itself, the host also runs a program called the jailer — a second layer of defense (a sandbox) that wraps the virtual machine in case it gets compromised.

The technology guarantees security and isolation of every customer's data, memory, and processes. AgentCore Runtime stands on Amazon's experience with Lambda, containers, and instances.

Runtime lets you deploy any AI agent or MCP server, regardless of SDK or framework, as long as it exposes HTTP on port 8080 (for MCP servers the port is 8000). AWS says the idea is for the service to be framework agnostic — the future secure home for your agent, guarding a company's most valuable asset: its data. Together with memory, access management, connectivity to other services, and agent evaluation, this shapes a complete AI platform.

Preparing the application takes three steps.

Creating a container image registry where the application will live:

resource "aws_ecr_repository" "agent_repo" {
  name                 = "agent-repo"
  image_tag_mutability = "IMMUTABLE"
  image_scanning_configuration {
    scan_on_push = false
  }
  force_delete = true
}

A Dockerfile with all of the application's dependencies:

# AgentCore Runtime runs ONLY on Graviton (linux/arm64).
# That's why the build must go through buildx with --platform linux/arm64.
FROM public.ecr.aws/docker/library/python:3.13-slim

WORKDIR /app

# Dependencies first — Docker caches the layer, so subsequent builds are fast
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

# The Runtime contract: an HTTP server listening on 8080
EXPOSE 8080

CMD ["python", "app.py"]

Building and pushing the image:

resource "random_id" "agent_rand_tag" {
  keepers = {
    agent_img_tag = local.agent_code_hash
  }
  byte_length = 2
}

resource "null_resource" "build_and_push_agent" {
  triggers = {
    dockerfile_hash = local.agent_code_hash
  }

  provisioner "local-exec" {
    command = <<-EOT
      aws ecr get-login-password --region ${data.aws_region.current.region} | docker login --username AWS --password-stdin ${aws_ecr_repository.agent_repo.repository_url}
      docker build -t ${aws_ecr_repository.agent_repo.repository_url}:${random_id.agent_rand_tag.hex} ${path.module}/
      docker push ${aws_ecr_repository.agent_repo.repository_url}:${random_id.agent_rand_tag.hex}
    EOT
  }
}

The Terraform code assembles from the building blocks: to bring up AgentCore Runtime, you pass in everything else and create it last — like Voltron, if you've seen it, this is the head — with the required role:

data "aws_iam_policy_document" "runtime_permissions" {
  statement {
    actions   = ["ecr:GetAuthorizationToken"]
    effect    = "Allow"
    resources = ["*"]
  }

  statement {
    actions = [
      "ecr:BatchGetImage",
      "ecr:GetDownloadUrlForLayer",
    ]
    effect    = "Allow"
    resources = [aws_ecr_repository.agent_repo.arn]
  }
  statement {
    actions = [
      "logs:DescribeLogStreams",
      "logs:CreateLogGroup",

    ]
    effect    = "Allow"
    resources = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:log-group:/aws/bedrock-agentcore/runtimes/*"]
  }
  statement {
    actions = [
      "logs:DescribeLogGroups"

    ]
    effect    = "Allow"
    resources = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:log-group:*"]
  }
  statement {
    actions = [
      "logs:CreateLogStream",
      "logs:PutLogEvents"

    ]
    effect    = "Allow"
    resources = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:log-group:/aws/bedrock-agentcore/runtimes/*:log-stream:*"]
  }
  statement {
    actions = [
      "cloudwatch:PutMetricData"
    ]
    resources = ["*"]
    effect    = "Allow"
    condition {
      test     = "StringEquals"
      values   = ["bedrock-agentcore"]
      variable = "cloudwatch:namespace"
    }
  }
  statement {
    actions = [
      "bedrock:InvokeModel",
      "bedrock:InvokeModelWithResponseStream"
    ]
    resources = ["arn:aws:bedrock:*::foundation-model/*",
    "arn:aws:bedrock:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:*"]
    effect = "Allow"
  }
  statement {
    sid     = "GetWorkloadAccessToken"
    effect  = "Allow"
    actions = ["bedrock-agentcore:GetWorkloadAccessToken"]
    resources = [
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/${aws_bedrockagentcore_gateway.gtw.gateway_id}"
    ]
  }
  statement {
    effect  = "Allow"
    actions = ["bedrock-agentcore:InvokeGateway"]
    resources = [
      aws_bedrockagentcore_gateway.gtw.gateway_arn
    ]
  }
  statement {
    effect = "Allow"
    actions = ["bedrock-agentcore:ListEvents",
      "bedrock-agentcore:CreateEvent",
    "bedrock-agentcore:RetrieveMemoryRecords"]
    resources = [
      aws_bedrockagentcore_memory.memory.arn
    ]
  }
}

resource "aws_iam_role" "rntime_role" {
  name               = "bedrock-agentcore-rntime-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_policy" "rntime_plcy" {
  policy = data.aws_iam_policy_document.runtime_permissions.json
  name   = "bedrock-agentcore-rntime-plcy"
}

resource "aws_iam_role_policy_attachment" "rntime_role_plcy_attach" {
  policy_arn = aws_iam_policy.rntime_plcy.arn
  role       = aws_iam_role.rntime_role.name
}

resource "aws_bedrockagentcore_agent_runtime" "example" {
  agent_runtime_name = "gitops_agent_runtime"
  role_arn           = aws_iam_role.rntime_role.arn

  agent_runtime_artifact {
    container_configuration {
      container_uri = "${aws_ecr_repository.agent_repo.repository_url}:${random_id.agent_rand_tag.hex}"
    }
  }

  network_configuration {
    network_mode = "PUBLIC"
  }
  environment_variables = {
    GATEWAY_URL = aws_bedrockagentcore_gateway.gtw.gateway_url
    MEMORY_ID   = aws_bedrockagentcore_memory.memory.id
    MODEL_ID    = "eu.anthropic.claude-sonnet-4-5-20250929-v1:0"
  }
  depends_on = [null_resource.build_and_push_agent]
}

Bonus: cross-account and cross-region without network magic

On top of everything else, the services above — and Gateway in particular — solve a problem that usually calls for more complex network constructions: calls between accounts and between regions. Two examples from real life. In both, the scenario is the same: the gateway lives in platform account A in eu-west-1, while the MCP server (mcp-atlassian) and Jira DC live in another account B and another region, eu-west-2. The difference is the price you pay.

Example 1: Runtime target — the network isn't there, because you don't need it

A gateway with no protocol type passes the Okta JWT onward, and with a single InvokeAgentRuntime over the AWS backbone it reaches the Runtime in the other account and region — no peering, no PrivateLink; this is an API, and the network stays hidden. The Runtime runs the forked mcp-atlassian (arm64, stateless, MCP on :8000) and goes out through egress ENIs straight to Jira DC with the user's token. The price: Identity has to live in the Runtime's account — the secrets move to the agent — and you lose both aggregation and semantic search, since the gateway is just a proxy.

Example 2: MCP target — semantic search, but you build the network yourself

Here the gateway uses the classic MCP protocol, since the target is not an AgentCore Runtime: MCP, CUSTOM_JWT, aggregation, and semantic search, with Identity exchanging the JWT for the user's token — the 3LO vault stays with the platform. The price is the network, and it's a whole chain: Lattice resource gateway (only in the gateway's region) → cross-region interface endpoint → PrivateLink → endpoint service in account B → internal NLB with TLS :443 and a public ACM certificate → EC2 with mcp-atlassian on :8000 → Jira DC. Every link is managed — but every link is yours.

Which one, when

If you just need to reach an agent or MCP server in another account and region — Runtime target: zero network work, but the vault moves to the runtime and you lose aggregation. If you're building a platform — one endpoint, semantic search, tokens and consents on your side — MCP target: you get all of that, but the Lattice, the cross-region PrivateLink, and the certificates are your problem.

What it costs

AgentCore has no monthly subscription — each capability has its own meter, and you pay only for what you use. Prices as of July 2026:

Runtime — $0.0895 per vCPU-hour and $0.00945 per GB-hour, billed per second. CPU is charged only while the agent is actually processing: I/O waiting is free, and agents spend 30–70% of a session waiting on the model and the tools.
Gateway — $0.005 per 1,000 invocations (ListTools, InvokeTool), $0.025 per 1,000 semantic searches, and $0.02 per 100 indexed tools per month.
Memory — $0.25 per 1,000 events (short-term memory); long-term is $0.75 per 1,000 records per month with the built-in strategies (the model is included in the price) and $0.50 per 1,000 retrievals. Three strategies mean three separate extractions from the same events.
Identity — $0.010 per 1,000 token requests… but free when it goes through Runtime or Gateway. In our architecture: zero.
Observability — at CloudWatch rates (about $0.35/GB for spans).

On the side you have ECR for the image, Secrets Manager for the PAT, and — the biggest line item — the tokens to the model, which are often 50–70% of the whole bill. That's exactly why Gateway's semantic search isn't a luxury but FinOps: a wrongly chosen tool costs tokens, and tokens cost the most.

In lieu of a conclusion

AgentCore isn't just hosting for agents. Identity decides who can do what, Gateway — which tool is the right one, Memory — what's worth remembering, Runtime — where all of it lives in isolation. And when you have to cross an account and a region, you now have a choice: you pay either with functionality or with network plumbing. Example 1 delivers on the intro's promise literally — no network at all. Example 2 shows that even when you do need the network, it snaps together from managed blocks like Lego, with no need to touch route tables.

Размислите на един лаик за Amazon Bedrock AgentCore платформата

Martin Nanchev — Fri, 31 Jul 2026 13:34:20 +0000

Съдържание

AgentCore Identity: кой какво може
AgentCore Gateway: правилният инструмент, без излишни токени
AgentCore Memory: какво помни агентът
AgentCore Runtime: Firecracker, MicroVM и контейнери
Бонус: cross-account и cross-region без мрежова магия
- Пример 1: Runtime target — мрежата я няма, защото не ти трябва
- Пример 2: MCP target — семантично търсене, но мрежата си я строиш сам
- Кое кога
Колко струва
Вместо заключение

AgentCore платформата се базира на повече от 20-годишния опит на AWS в решаването на проблемите на магазина на Amazon и на крайните клиенти. Услугата променя много от вече изградените ни представи за compute, network, storage и database и позволява връзка между акаунти и между различни региони, без да е нужно да познаваш софтуерно дефинираните мрежи в дълбочина. Дава и възможност да интегрираш легаси системи към новите агенти. Платформата се състои от няколко функционалности, които ще разгледаме една по една: Identity, Gateway, Memory и Runtime.

AgentCore Identity: кой какво може

AgentCore Identity управлява самоличността на агента и достъпа му до външни услуги и инструменти чрез абстракция, наречена workload identity. Ключовете за външните услуги се пазят в Secrets Manager с префикс bedrock-agentcore-identity!. На практика услугата управлява потока на идентификация — кой какво иска да прави и дали е одобрено. Така гарантираш липсата на своеволия: агентът има достъп точно толкова, колкото си му дал.

Предлагат се две форми на автентикация и оторизация.

2LO (2-legged OAuth) — агентът се представя пред външната система с предварително издаден токен: нещо като пропуск от твое име, който определя какво точно му е позволено да прави. Секретите се пазят в Secrets Manager. За iGaming сектора 2LO е нещо като табу, което трябва да се избягва на всяка цена — думата PAT предизвиква силно вълнение в инфосек общността, независимо дали токенът е криптиран с KMS ключ със строга политика. Нещо като „дракарис“ при драконите от Game of Thrones.

3LO (3-legged OAuth) — В този случай имаш три партии - агент, потребител и трета система. Агентът сам по себе си няма никакви права в Jira, Confluence или GitHub. Когато за пръв път поиска да работи от твое име, системата ти показва екран за одобрение — както когато приложение те пита дали разрешаваш достъп до камерата. На екрана пише точно какво ще може агентът (например само да чете тикети) и когато натиснеш Allow, системата издава токен само с тези права, който AgentCore Identity пази за теб. Оттам нататък агентът работи с този токен и може да прави единствено одобрените действия — нищо повече. За всяка от системите (Jira, Confluence, GitHub) одобрението е отделно, със собствени права.

Пример с 2LO:

resource "aws_bedrockagentcore_api_key_credential_provider" "github" {
  name               = "github-pat-token"
  api_key_wo         = local.envs["GITHUB_PAT"]
  api_key_wo_version = var.token_version
}

Самоличността и ключовете обаче са само половината от историята — агентът има нужда и от инструменти, до които да стига с тях. Тук идва Gateway.

AgentCore Gateway: правилният инструмент, без излишни токени

AgentCore Gateway позволява връзка към Lambda функции, Smithy модели, OpenAPI-съвместими API-та и външни MCP сървъри. Решава и един много важен проблем: когато подадеш много инструменти на агент, той често греши и избира неправилния. Това струва токени, които клиентът плаща. Затова Gateway използва семантично търсене, за да открие правилния инструмент.

Terraform кодът дефинира gateway-я и неговата цел (target). OpenAPI схемата може да се подаде inline в JSON формат; ако е в YAML, се преобразува с Terraform функции — jsonencode(yamldecode(...)). Една особеност: ако целта е AgentCore Runtime агент, gateway-ят трябва да е без зададен protocol type — Runtime target-и не могат да се добавят към gateway с protocol_type = "MCP".

data "aws_iam_policy_document" "gtw_plcy" {
  statement {
    sid     = "GetResourceOauth2Token"
    effect  = "Allow"
    actions = ["bedrock-agentcore:GetResourceApiKey"]
    resources = ["arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:token-vault/default/apikeycredentialprovider/${aws_bedrockagentcore_api_key_credential_provider.github.name}",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/${aws_bedrockagentcore_gateway.gtw.gateway_id}",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:token-vault/default"
    ]
  }
  statement {
    sid     = "GetWorkloadAccessToken"
    effect  = "Allow"
    actions = ["bedrock-agentcore:GetWorkloadAccessToken"]
    resources = [
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/${aws_bedrockagentcore_gateway.gtw.gateway_id}"
    ]
  }
  statement {
    effect  = "Allow"
    actions = ["secretsmanager:GetSecretValue"]
    resources = [
      aws_bedrockagentcore_api_key_credential_provider.github.api_key_secret_arn[0].secret_arn
    ]
  }
}
data "aws_iam_policy_document" "assume_role" {
  statement {
    effect  = "Allow"
    actions = ["sts:AssumeRole"]
    principals {
      type        = "Service"
      identifiers = ["bedrock-agentcore.amazonaws.com"]
    }
    condition {
      test     = "StringEquals"
      values   = [data.aws_caller_identity.current.account_id]
      variable = "aws:SourceAccount"
    }
  }
}
resource "aws_iam_role" "gtw_role" {
  name               = "bedrock-agentcore-gateway-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_policy" "gtw_plcy" {
  policy = data.aws_iam_policy_document.gtw_plcy.json
  name   = "bedrock-agentcore-gateway-plcy"
}

resource "aws_iam_role_policy_attachment" "gtw_role_plcy_attach" {
  policy_arn = aws_iam_policy.gtw_plcy.arn
  role       = aws_iam_role.gtw_role.name
}

resource "aws_bedrockagentcore_gateway" "gtw" {
  name            = "gateway"
  role_arn        = aws_iam_role.gtw_role.arn
  authorizer_type = "AWS_IAM"
  protocol_type   = "MCP"
  exception_level = "DEBUG"
  protocol_configuration {
    mcp {
      search_type = "SEMANTIC"
    }
  }
}

resource "aws_bedrockagentcore_gateway_target" "github_target" {
  name               = "api-target"
  gateway_identifier = aws_bedrockagentcore_gateway.gtw.gateway_id
  description        = "External API target with API key authentication"

  credential_provider_configuration {
    api_key {
      provider_arn              = aws_bedrockagentcore_api_key_credential_provider.github.credential_provider_arn
      credential_location       = "HEADER"
      credential_parameter_name = "Authorization"
      credential_prefix         = "Bearer"
    }
  }

  target_configuration {
    mcp {
      open_api_schema {
        inline_payload {
          payload = jsonencode(yamldecode(file("${path.module}/github-tools.yaml")))
        }
      }
    }
  }
}

Понякога освен достъп до външни услуги искаме агентът да запомня информация за крайния потребител, за да предлага по-ценни решения на база на неговия контекст. Това става с AgentCore Memory.

AgentCore Memory: какво помни агентът

Паметта се дели според стратегията на: потребителски предпочитания (неща, които потребителят харесва), семантична (факти от контекста), обобщаваща и индивидуална (custom). Извличането на дългосрочните записи е асинхронно и отнема време. Има и една особеност при създаването: всяка стратегия блокира създаването на следващата, т.е. ако настройваш повече от една, между тях трябва да се изчака — емпирично около 190 секунди; в кода по-долу съм заложил 300 за спокойствие. В Terraform изглежда по следния начин:

resource "aws_iam_role" "bedrock_memory_role" {
  name               = "bedrock-agentcore-memory-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_role_policy_attachment" "bedrock_plcy_attach" {
  role       = aws_iam_role.bedrock_memory_role.name
  policy_arn = "arn:aws:iam::aws:policy/AmazonBedrockAgentCoreMemoryBedrockModelInferenceExecutionRolePolicy"
}

resource "aws_bedrockagentcore_memory" "memory" {
  name                      = "agent_memory"
  event_expiry_duration     = 30
  memory_execution_role_arn = aws_iam_role.bedrock_memory_role.arn
}

resource "aws_bedrockagentcore_memory_strategy" "user_preference" {
  name        = "userpreference_strategy"
  memory_id   = aws_bedrockagentcore_memory.memory.id
  type        = "USER_PREFERENCE"
  description = "user preference strategy"
  namespaces  = ["/users/{actorId}/preferences/"]
}

resource "time_sleep" "wait_after_usr_pref_strategy" {
  depends_on       = [aws_bedrockagentcore_memory_strategy.user_preference]
  create_duration  = "300s"
  destroy_duration = "300s"
  triggers = {
    strategy_id = aws_bedrockagentcore_memory_strategy.user_preference.memory_strategy_id
  }
}

resource "aws_bedrockagentcore_memory_strategy" "semantic" {
  name        = "semantic_strategy"
  memory_id   = aws_bedrockagentcore_memory.memory.id
  type        = "SEMANTIC"
  description = "semantic strategy"
  namespaces  = ["/users/{actorId}/facts/"]
  depends_on  = [time_sleep.wait_after_usr_pref_strategy]
}

resource "time_sleep" "wait_after_semantic_strategy" {
  depends_on       = [aws_bedrockagentcore_memory_strategy.semantic]
  create_duration  = "300s"
  destroy_duration = "300s"
  triggers = {
    strategy_id = aws_bedrockagentcore_memory_strategy.semantic.memory_strategy_id
  }
}

resource "aws_bedrockagentcore_memory_strategy" "summarization" {
  name        = "summarization_strategy"
  memory_id   = aws_bedrockagentcore_memory.memory.id
  type        = "SUMMARIZATION"
  description = "summarization strategy"
  namespaces  = ["/summaries/{actorId}/{sessionId}/"]
  depends_on  = [time_sleep.wait_after_semantic_strategy]
}

Записите се пазят в йерархични namespace-и, в които {actorId} и {sessionId} се заместват с реалните стойности при запис.

Ето как всичко дотук се събира в самия агент — четене от дългосрочната памет, разговор с инструментите през Gateway (MCP със SigV4 подпис) и запис на хода като събитие:

"""Repo Ops agent — AgentCore Runtime entrypoint.

Поток на всяка заявка:
  1. recall()  -> чете дългосрочна памет (LTM) за този потребител
  2. Agent()   -> LLM разговор с инструментите от Gateway (MCP през SigV4)
  3. create_event() -> записва хода като събитие (STM); стратегиите
                       извличат LTM асинхронно след ~1 минута
"""
import os

import boto3
import httpx
from botocore.auth import SigV4Auth
from botocore.awsrequest import AWSRequest
from bedrock_agentcore.memory import MemoryClient
from bedrock_agentcore.runtime import BedrockAgentCoreApp
from mcp.client.streamable_http import streamablehttp_client
from strands import Agent
from strands.models import BedrockModel
from strands.tools.mcp import MCPClient

REGION = os.environ.get("AWS_REGION", "eu-central-1")
GATEWAY_URL = os.environ["GATEWAY_URL"]   # .../mcp
MEMORY_ID = os.environ["MEMORY_ID"]
MODEL_ID = os.environ["MODEL_ID"]

app = BedrockAgentCoreApp()               # вдига HTTP сървър на :8080
memory = MemoryClient(region_name=REGION)


class SigV4HTTPXAuth(httpx.Auth):
    """Подписва всяка MCP заявка със SigV4.

    Gateway-ът е с authorizer_type = "AWS_IAM", тоест иска подпис
    точно както всяко AWS API извикване. Ролята на Runtime трябва да
    има bedrock-agentcore:InvokeGateway върху ARN-а на gateway-а.
    """

    requires_request_body = True

    def __init__(self, service: str, region: str):
        self._creds = boto3.Session().get_credentials()
        self._service = service
        self._region = region

    def auth_flow(self, request: httpx.Request):
        aws_req = AWSRequest(
            method=request.method,
            url=str(request.url),
            data=request.content,
            headers={"host": request.url.host},
        )
        SigV4Auth(self._creds, self._service, self._region).add_auth(aws_req)
        request.headers.update(dict(aws_req.headers))
        yield request


def recall(actor_id: str, query: str) -> str:
    """Семантично търсене в дългосрочната памет.

    Namespace-ите трябва да съвпадат ТОЧНО с тези в Terraform
    (включително наклонената черта накрая).
    """
    namespaces = [
        f"/users/{actor_id}/preferences/",
        f"/users/{actor_id}/facts/",
    ]
    chunks = []
    for ns in namespaces:
        try:
            records = memory.retrieve_memories(
                memory_id=MEMORY_ID,
                namespace=ns,
                query=query,
                top_k=3,
            )
            for rec in records:
                text = rec.get("content", {}).get("text", "")
                if text:
                    chunks.append(f"- {text}")
        except Exception as exc:
            # Паметта никога не трябва да сваля агента
            print(f"[memory] retrieve failed for {ns}: {exc}")
    return "\n".join(chunks)


def remember(actor_id: str, session_id: str, prompt: str, answer: str) -> None:
    """Записва хода като събитие в краткосрочната памет."""
    try:
        memory.create_event(
            memory_id=MEMORY_ID,
            actor_id=actor_id,
            session_id=session_id,
            messages=[(prompt, "USER"), (answer, "ASSISTANT")],
        )
    except Exception as exc:
        print(f"[memory] create_event failed: {exc}")


@app.entrypoint
def invoke(payload, context):
    prompt = payload.get("prompt", "")
    actor_id = payload.get("actor_id", "default")

    # session_id идва от --runtime-session-id при извикването
    session_id = getattr(context, "session_id", None) or "local-dev-session-0000000000000000"

    memories = recall(actor_id, prompt)

    system_prompt = (
        "Ти си Repo Ops — кратък асистент за GitHub операции. "
        "Използвай наличните инструменти за всичко свързано с "
        "хранилища, pull request-и и issue-та. Никога не измисляй "
        "данни — ако инструмент върне грешка, кажи я на потребителя. "
        "Питай за потвърждение преди да създадеш issue."
    )
    if memories:
        system_prompt += f"\n\nКакво помниш за този потребител:\n{memories}"

    gateway = MCPClient(
        lambda: streamablehttp_client(
            GATEWAY_URL,
            auth=SigV4HTTPXAuth("bedrock-agentcore", REGION),
        )
    )

    with gateway:
        tools = gateway.list_tools_sync()
        print(f"[gateway] tools: {[t.tool_name for t in tools]}")
        agent = Agent(
            model=BedrockModel(model_id=MODEL_ID, region_name=REGION),
            tools=tools,
            system_prompt=system_prompt,
        )
        answer = str(agent(prompt))

    remember(actor_id, session_id, prompt, answer)
    return {"result": answer}


if __name__ == "__main__":
    app.run()

Този код обаче трябва да живее някъде. Време е за частта, на която винаги блокирам на интервю.

AgentCore Runtime: Firecracker, MicroVM и контейнери

Винаги съм се чудил как да разкажа какво е AgentCore Runtime. Въпросите за MicroVM, виртуални машини, контейнери и процеси винаги са трудни за разказване. Може би сам не го разбирам докрай, може би заешката дупка е толкова дълбока, че е трудно да се обясни просто. Затова реших да започна отдалеч — може би и защото на интервю винаги блокирам точно на тези въпроси.

За да развенчая мита, започвам от MicroVM.

Две от опциите за деплойване на приложение (включително AI приложение) са виртуална машина и контейнер. Разликата: контейнерите пакетират всички зависимости и библиотеки на приложението и споделят общо ядро и операционна система с хоста, докато всяка виртуална машина има собствено ядро, собствена операционна система и заделена част от ресурсите. Ядрото управлява хардуера чрез драйвери, т.е. има неограничен достъп до процесорно време и памет; приложенията, от друга страна, нямат.

Често се възприема, че виртуалната машина е по-сигурна от контейнерите. Защо? Може би защото контейнерите споделят ядро с хоста.

Дали това прави контейнерите несигурни спрямо виртуалната машина, е спорен въпрос. При виртуалната машина имаш пачване, управление на драйвери и по-бавно стартиране. Има и вариантът distroless контейнер — без шел и пакетни мениджъри, само с необходимото за крайното приложение.

MicroVM се опитва да реши предизвикателствата с бавното стартиране, като освен това предлага по-голяма изолация, характерна за виртуалните машини: всяка машина идва със собствено виртуализирано ядро. Firecracker — MicroVM технологията зад AWS Lambda — позволява да стартираш до 150 MicroVM в секунда, като всяка от тях се вдига за около 125 ms.

Каква е връзката на виртуалните машини и контейнерите с AgentCore Runtime? Firecracker стои зад AgentCore Runtime: ти деплойваш контейнер, а той се изпълнява върху Firecracker виртуална машина. Така получаваш максимална изолация — една Firecracker машина не може да използва ресурси на друга, това е забранено на ниво хипервайзор — и в същото време максимално лесно деплойване на всяко AI приложение под формата на контейнер с всички зависимости. Около самия Firecracker процес на хоста върви и програма на име jailer — втори слой защита (пясъчник), който обвива виртуалната машина, в случай че тя бъде компрометирана.

Технологията гарантира сигурност и изолация на данните, паметта и процесите на всеки клиент. AgentCore Runtime стъпва на опита на Amazon с Lambda, контейнерите и инстанциите.

Runtime дава възможност да деплойнеш всеки AI агент или MCP сървър, независимо от SDK или framework, стига да изложи HTTP на порт 8080 (за MCP сървъри портът е 8000). AWS споделят, че идеята е услугата да е framework agnostic — бъдещият сигурен дом за твоя агент, който пази най-ценното за една фирма: данните. Заедно с паметта, управлението на достъпи, връзката с други услуги и оценката на агенти това оформя цялостна AI платформа.

Подготовката на приложението включва три стъпки.

Създаване на регистър за контейнерни изображения, в който ще живее приложението:

resource "aws_ecr_repository" "agent_repo" {
  name                 = "agent-repo"
  image_tag_mutability = "IMMUTABLE"
  image_scanning_configuration {
    scan_on_push = false
  }
  force_delete = true
}

Docker файл с всички зависимости на приложението:

# AgentCore Runtime работи САМО на Graviton (linux/arm64).
# Затова build-ът задължително минава през buildx с --platform linux/arm64.
FROM public.ecr.aws/docker/library/python:3.13-slim

WORKDIR /app

# Първо зависимостите — така Docker кешира слоя и следващите build-ове са бързи
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY app.py .

# Контрактът на Runtime: HTTP сървър, който слуша на 8080
EXPOSE 8080

CMD ["python", "app.py"]

Build и push на изображението:

resource "random_id" "agent_rand_tag" {
  keepers = {
    agent_img_tag = local.agent_code_hash
  }
  byte_length = 2
}

resource "null_resource" "build_and_push_agent" {
  triggers = {
    dockerfile_hash = local.agent_code_hash
  }

  provisioner "local-exec" {
    command = <<-EOT
      aws ecr get-login-password --region ${data.aws_region.current.region} | docker login --username AWS --password-stdin ${aws_ecr_repository.agent_repo.repository_url}
      docker build -t ${aws_ecr_repository.agent_repo.repository_url}:${random_id.agent_rand_tag.hex} ${path.module}/
      docker push ${aws_ecr_repository.agent_repo.repository_url}:${random_id.agent_rand_tag.hex}
    EOT
  }
}

Terraform кодът се сглобява от съставните елементи: за да вдигнеш AgentCore Runtime, подаваш всичко останало и го създаваш последен — като Волтрон, ако сте го гледали, това е главата — с необходимата роля:

data "aws_iam_policy_document" "runtime_permissions" {
  statement {
    actions   = ["ecr:GetAuthorizationToken"]
    effect    = "Allow"
    resources = ["*"]
  }

  statement {
    actions = [
      "ecr:BatchGetImage",
      "ecr:GetDownloadUrlForLayer",
    ]
    effect    = "Allow"
    resources = [aws_ecr_repository.agent_repo.arn]
  }
  statement {
    actions = [
      "logs:DescribeLogStreams",
      "logs:CreateLogGroup",

    ]
    effect    = "Allow"
    resources = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:log-group:/aws/bedrock-agentcore/runtimes/*"]
  }
  statement {
    actions = [
      "logs:DescribeLogGroups"

    ]
    effect    = "Allow"
    resources = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:log-group:*"]
  }
  statement {
    actions = [
      "logs:CreateLogStream",
      "logs:PutLogEvents"

    ]
    effect    = "Allow"
    resources = ["arn:aws:logs:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:log-group:/aws/bedrock-agentcore/runtimes/*:log-stream:*"]
  }
  statement {
    actions = [
      "cloudwatch:PutMetricData"
    ]
    resources = ["*"]
    effect    = "Allow"
    condition {
      test     = "StringEquals"
      values   = ["bedrock-agentcore"]
      variable = "cloudwatch:namespace"
    }
  }
  statement {
    actions = [
      "bedrock:InvokeModel",
      "bedrock:InvokeModelWithResponseStream"
    ]
    resources = ["arn:aws:bedrock:*::foundation-model/*",
    "arn:aws:bedrock:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:*"]
    effect = "Allow"
  }
  statement {
    sid     = "GetWorkloadAccessToken"
    effect  = "Allow"
    actions = ["bedrock-agentcore:GetWorkloadAccessToken"]
    resources = [
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default",
      "arn:aws:bedrock-agentcore:${data.aws_region.current.region}:${data.aws_caller_identity.current.account_id}:workload-identity-directory/default/workload-identity/${aws_bedrockagentcore_gateway.gtw.gateway_id}"
    ]
  }
  statement {
    effect  = "Allow"
    actions = ["bedrock-agentcore:InvokeGateway"]
    resources = [
      aws_bedrockagentcore_gateway.gtw.gateway_arn
    ]
  }
  statement {
    effect = "Allow"
    actions = ["bedrock-agentcore:ListEvents",
      "bedrock-agentcore:CreateEvent",
    "bedrock-agentcore:RetrieveMemoryRecords"]
    resources = [
      aws_bedrockagentcore_memory.memory.arn
    ]
  }
}

resource "aws_iam_role" "rntime_role" {
  name               = "bedrock-agentcore-rntime-role"
  assume_role_policy = data.aws_iam_policy_document.assume_role.json
}

resource "aws_iam_policy" "rntime_plcy" {
  policy = data.aws_iam_policy_document.runtime_permissions.json
  name   = "bedrock-agentcore-rntime-plcy"
}

resource "aws_iam_role_policy_attachment" "rntime_role_plcy_attach" {
  policy_arn = aws_iam_policy.rntime_plcy.arn
  role       = aws_iam_role.rntime_role.name
}

resource "aws_bedrockagentcore_agent_runtime" "example" {
  agent_runtime_name = "gitops_agent_runtime"
  role_arn           = aws_iam_role.rntime_role.arn

  agent_runtime_artifact {
    container_configuration {
      container_uri = "${aws_ecr_repository.agent_repo.repository_url}:${random_id.agent_rand_tag.hex}"
    }
  }

  network_configuration {
    network_mode = "PUBLIC"
  }
  environment_variables = {
    GATEWAY_URL = aws_bedrockagentcore_gateway.gtw.gateway_url
    MEMORY_ID   = aws_bedrockagentcore_memory.memory.id
    MODEL_ID    = "eu.anthropic.claude-sonnet-4-5-20250929-v1:0"
  }
  depends_on = [null_resource.build_and_push_agent]
}

Бонус: cross-account и cross-region без мрежова магия

Освен всичко останало, горните услуги — и в частност Gateway — решават проблем, който обикновено налага по-сложни мрежови конструкции: извикванията между акаунти и между региони. Два примера от реалния живот. И в двата сценарият е един и същ: gateway-ят живее в платформен акаунт А в eu-west-1, а MCP сървърът (mcp-atlassian) и Jira DC — в друг акаунт Б и друг регион, eu-west-2. Разликата е в цената, която плащаш.

Пример 1: Runtime target — мрежата я няма, защото не ти трябва

Gateway без protocol type подава Okta JWT-то нататък и с едно InvokeAgentRuntime по AWS backbone мрежата стига до Runtime в другия акаунт и регион — никакъв peering, никакъв PrivateLink, това е API, а мрежата остава скрита. Runtime-ът върти форкнатия mcp-atlassian (arm64, stateless, MCP на :8000) и излиза през egress ENI-та право към Jira DC с потребителския токен. Цената: Identity трябва да живее в акаунта на Runtime-а — секретите се местят при агента — а губиш и агрегацията, и семантичното търсене, тъй като gateway е просто прокси

Пример 2: MCP target — семантично търсене, но мрежата си я създаваш сам

Тук gateway-ят е с класическият MCP протокол тъй като целта не е аgentcore runtime: MCP, CUSTOM_JWT, агрегация и семантично търсене, а Identity сменя JWT-то срещу потребителския токен — 3LO vault-ът остава при платформата. Цената е мрежата, и тя е цяла верига: Lattice resource gateway (само в региона на gateway-я) → cross-region interface endpoint → PrivateLink → endpoint service в акаунт Б → вътрешен NLB с TLS :443 и публичен ACM сертификат → EC2 с mcp-atlassian на :8000 → Jira DC. Всяко звено е managed — но всяко звено е твое.

Кое кога

Ако ти трябва просто да стигнеш до агент или MCP сървър в друг акаунт и регион — Runtime target: нула мрежова работа, но vault-ът се мести при runtime-а и губиш агрегацията. Ако строиш платформа — един endpoint, семантично търсене, токените и съгласията при теб — MCP target: получаваш всичко това, но Lattice-ът, cross-region PrivateLink-ът и сертификатите са твоя грижа.

Колко струва

AgentCore няма месечен абонамент — всяка функционалност е с отделен брояч и плащаш само каквото ползваш. Цените към юли 2026:

Runtime — $0.0895 на vCPU-час и $0.00945 на GB-час, на секунда. CPU се таксува само докато агентът реално обработва: I/O чакането е безплатно, а агентите прекарват 30–70% от сесията в чакане на модела и инструментите.
Gateway — $0.005 на 1000 извиквания (ListTools, InvokeTool), $0.025 на 1000 семантични търсения и $0.02 на 100 индексирани инструмента месечно.
Memory — $0.25 на 1000 събития (краткосрочна памет); дългосрочната е $0.75 на 1000 записа месечно с вградените стратегии (моделът е включен в цената) и $0.50 на 1000 извличания. Три стратегии значат три отделни извличания от същите събития.
Identity — $0.010 на 1000 заявки за токен… но безплатна, когато минава през Runtime или Gateway. В нашата архитектура: нула.
Observability — по тарифите на CloudWatch (около $0.35/GB за span-ове).

Отстрани стоят ECR за образа, Secrets Manager за PAT-а и — най-голямото перо — токените към модела, които често са 50–70% от цялата сметка. Точно затова семантичното търсене на Gateway не е лукс, а FinOps: грешно избраният инструмент струва токени, а токените струват най-много.

Вместо заключение

AgentCore не е просто хостинг за агенти. Identity решава кой какво може, Gateway — кой инструмент е правилният, Memory — какво си струва да се помни, Runtime — къде всичко това живее изолирано. А когато се наложи да прекосиш акаунт и регион, вече имаш избор: плащаш или с функционалност, или с мрежова връзка. Пример 1 изпълнява обещанието от увода буквално — никаква мрежа. Пример 2 показва, че дори когато мрежата ти трябва, тя се сглобява от managed блокчета като lego без нужда от промени по рутингтаблици.

Жизнен цикъл за разработка на агенти

Martin Nanchev — Mon, 20 Jul 2026 13:06:49 +0000

Жизненият цикъл на разработка на AI агенти (или как претеглих 22 ястия в името на науката)

В тази статия бих искал да представя един от многото възможни подходи за разработка на агенти. Не мога да твърдя, че е най-оптималният или че ще проработи точно при вас, но наблюденията ми показват, че е доста ефективен.

Как започна всичко

Разработката на софтуер в днешно време следва т.нар. цикъл на разработка на софтуер (SDLC), състоящ се от няколко фази: планиране, дефиниране на изискванията, дизайн (познатите HLD и LLD), разработка, тестване и поддръжка — която включва обратна връзка, нови функционалности и оперативни дейности като пачване.

Този цикъл се следва от всеки софтуерен продукт, в който съм участвал, без значение от методологията — kanban(waterfall с екстри), waterfall или „agile“, който всъщност е waterfall с daily-та.

Това ме накара да се запитам: как би изглеждал подобен цикъл при разработката на AI агент? За да не започвам от нулата, прочетох книгата AI Engineering на Chip Huyen. Тя разграничава следните фази, които се припокриват изненадващо добре със съществуващия цикъл: зараждане на идеята, методология за оценка, промпт инженерство, дизайн на търсенето и агента, фино напасване, оптимизация на процеса, системен дизайн и обратна връзка.

А какво се крие зад всяка от фазите, ще илюстрирам с нагледен пример.

1. Зараждане на идеята

Искам от снимка на храна да получавам вида ѝ и грамажа — нещо като дневник за консумирана храна, но без ръчното писане.

Имаме идея. Сега е моментът да я подложим на разпит със следния въпрос:

Може ли идеята да се изпълни без LLM? Тоест — можем ли да построим детерминирана система без участието на LLM? LLM (фокусиран върху текст) или, в днешно време, foundation model (FM — мултимодален модел, който приема изображения, звук и видео) работи с т.нар. logprobs (log probabilities). За да превърнем logprob във вероятност, използваме формулата:

probability = e^(logprob)

Заради този вероятностен характер работата с FM често се различава от работата с детерминирани системи. Конкретният проблем обаче е достатъчно интересен за разработка с FM — създаването на специализирана невронна мрежа би отнело значително време и е спорно дали резултатите ѝ ще са по-добри. FM се тренират върху огромни обеми данни (Common Crawl, Colossal Clean Crawled Corpus, LAION за изображения) и използват сложни архитектури като Mixture of Experts: входните данни се разпределят между „експерти“ и вместо да се минава през всички параметри на модела, се активира само частта, отговаряща например за изображения. Миксът от експерти подобрява скоростта, като използваната памет е по-голяма.

2. Оценка

Тук трябва да изберем подходящ модел. Моделите биват три вида - open weight (при тях не знаем с какъв датасет са тренирани, но знаем теглата и архитектурата), open model (тренировъчните данни и теглата на модела са публикувани публично) и затворени модели. Всеки модел си има място в публичните ранглисти и един от вариантите е просто да избереш от там. Това често не е достатъчно, защото част от данните или датасетовете, използвани за оценка, могат — умишлено или по погрешка — да попаднат в данните за трениране на модела. Ако приемем, че моделът е ученик, това означава, че ученикът е влязъл на изпита с отговорите в джоба. Вероятно ще изкара шестица, но дали това говори нещо за вашия проблем?

Затова се избират няколко модела от ранглистата, които после се тестват със снимки от телефона и тегла, валидирани с везна. С други думи — за всеки проблем е добре да се направи частен датасет. (Истината е, че това е трудоемко и много често се пропуска независимо от големината на компанията. Не се залъгвайте че големите ентърпраиз компании тестват, защото имат ресурси)

Първата стъпка е датасет от 22 различни храни, снимани с телефон, и JSON файл с храните и теглата им. Да, това означава да претеглиш ригатоните, преди да ги изядеш. Науката иска жертви. Пример:

{
  "IMG_5476.HEIC": {
    "items": [
      {
        "name": "cooked rigatoni pasta",
        "grams": 173
      },
      {
        "name": "white cheese",
        "grams": 76
      }
    ]
  }
}

След като имаме 22 уникални храни, идва ред на тестовете — използвах Strands SDK evaluation в GitHub Actions и анализирах резултатите на моделите за конкретния проблем:

Защо GitHub Actions? Оценката много прилича на пайплайн — добре е да може да се повтаря при смяна на модела и/или при итерация към следващата версия на промпта и системния промпт.

За всички модели използвах LiteLLM като рутер — дава един общ API, върху който можеш да построиш нещо като фабрика за агенти. Разходите се следят лесно, а интеграцията със Strands, Bedrock и AWS е добра.

Няколко думи за пропуснатите: Fable е много добър особено при писане на код, но резултатите му за разпознаване на изображения и тегло не са толкова добри. Muse Spark 1.1 - този може би е най-добрият модел, до който съм се докосвал някога. За да мога да направя апи кал си дигнах инстанция в North Virginia. Ако опитам да направя апи кал от личния си лаптоп към Лама 4/Muse 1.1 моделите съм блокиран по IP. Точността при определяне на теглото или грешката бяха около 17%. Създателят на фейсбук обаче пропуска Европа заради EU AI Act. Аз реших да го пропусна също.

3. Промпт инженерство

Тази стъпка включва оптимизация на промпта. Може да стане автоматично с Prompt Optimization в Amazon Bedrock: пишеш първоначален промпт, а FM в Bedrock го оптимизира. Доста прилича на т.нар. meta prompting, където молиш модела да оптимизира промпта… за самия себе си.

Добра практика е промптът и системният промпт да имат версии и да се пазят някъде — git като минимум. Amazon предлагат и Prompt Management, което позволява да съхраняваш различни версии на промпта и да тестваш новата спрямо старата.

4. Дизайн на агента

Избираме подходящ фреймуърк, подаваме системния промпт и решаваме дали ще имаме един агент, или взаимодействие между няколко. Разликата между агент и обикновено извикване на модела е, че агентът представлява цикъл, който извиква инструменти. В момента, в който инструментите свършат работа и изходът не може да се подобри повече, се връща отговор.

5. Fine-tuning (фино напасване)

Обикновено не се налага, но идеята е да заредим датасет, който помага на модела да се справя по-лесно с нашия конкретен проблем. OpenAI поддържат fine-tuning и model registry в платформата си, което позволява напасване към специфичен проблем. Fine-tuning е една стъпка над промпт инженерството — тук се напасват самите тегла на модела към конкретен датасет, за да се представя по-успешно. (базовите тегла са замразени а се тренира малка част, тренират се под 1% от теглата)

6. Филтриране на данните

Тук е важно да дефинираме и т.нар. guardrail — защитата на крайния потребител от вредни изходи. Много хора си мислят, че това е „защитен промпт“, в който описваш какво моделът не трябва да прави. Само че моделът е вероятностен, което означава, че промптът не е достатъчен. Големите модели обикновено следват заложеното поведение, но при по-малките това е трудно, а понякога и невъзможно. Guardrail е допълнителен софтуерен слой, използващ машинно обучение, който проверява входовете и изходите, преди изобщо да стигнат до модела или до крайния потребител. Освен всичко друго guardrail-ите пазят и портфейла — никой не обича да плаща за промпт, който изобщо не е трябвало да стига до модела. Могат да се дефинират и на ниво организация в AWS.

7. Латентност

Ние, инженерите, имаме навика да избираме най-големите модели заради по-добрите резултати — по-голямото очевидно е по-добро, нали? Цената на този избор е, че скоростта намалява, а латентността за крайния потребител расте. Моделите, които „мислят“ повече, са по-бавни и не винаги дават по-добри резултати, както се вижда и от таблицата по-горе. Тук трябва да се намери балансът между скорост, производителност и цена.

Разходите е важно да се наблюдават и да се алармира при определена граница на токените

8. Обратна връзка

Обратната връзка от клиента често помага за развитието на продукта. В приложението може да се даде избор между два промпта (Human in the loop), а при мобилно приложение — да се пусне в TestFlight и мнението да се събере чрез анкета. Това спомага за цялостното развитие и помага да разбереш дали си в правилната посока, или е нужна стратегическа промяна (pivot), преди да излезеш на пазара. И най-важното — дали отговаряш на очакванията на клиента.

Обобщение

Надявам се разработката на агенти да се разрасне през следващите години и да видим повече автономни системи, които — макар и вероятностни — действат според очакванията ни.

FinOps Without the Walk of Doom — meet the AWS FinOps Agent

Martin Nanchev — Sun, 14 Jun 2026 10:26:49 +0000

The Walk of Doom Is Over: Automating Cost Investigations with the AWS FinOps Agent

The old way: the "walk of doom"

Controlling costs and investigating cost incidents in AWS used to be a tedious process. Why? Well, picture this: you get a $5,000 bill for CloudTrail, but Cost Anomaly Detection flags it as low severity, the notification never reaches you, and it quietly goes unnoticed for a couple of days.

By the time you do notice, here's the ritual you're stuck with:

First, you receive a budget alarm.
Then you ask yourself how? — and double-check whether this was a forecast or an actual.
It was actual. You start to panic, and the walk of doom begins: analyse costs in Cost Explorer → stare at the bill → "oh, there was a Cost and Budgets report" → back to Cost Explorer to filter by service and hunt down the troublemaker.
You finally found it. Now the real question — why, and how did it get triggered? Off you go to CloudTrail.

Enter the AWS FinOps Agent

AWS solved all of this by introducing the AWS FinOps Agent. It automates FinOps reports and alerts, and those alerts can be wired into Slack or Jira Cloud.

Now you might say: "Look, I'm an experienced FinOps practitioner, I'm not like you, and I do things my own opinionated way." Fair enough — and that's exactly the point. As an experienced practitioner, you can feed the agent context and it will enhance your skills, producing a personalised report easier than ever before. You can finally nail those stakeholder presentations and focus on the things you always wanted to focus on: cost, or even cost per customer, or cost per bet.

Okay, enough marketing. Let's see how it actually works.

Setting it up

1. Create the agent

First, create an agent with a name that follows your naming convention. In my case:

2. Create the roles

Next, you're invited to create two roles: the agent role and the agent operator role. You can customise them to follow your internal standards, or leave it to the wizard. If you let the wizard handle it, it attaches the following managed policies:

FinOpsAgentRole → FinOpsAgentAgentPolicy
FinOpsAgentOperator → FinOpsAgentOperatorPolicy

3. Select integrations

Pick your integrations. For Slack, you'll need to be a workspace admin — or ask your admin to approve the AWS FinOps Agent – US East Slack app.

4. Give it context

This is where it gets interesting. Remember how I said it enhances experienced practitioners? The agent lets you give it context — including which teams are responsible for which accounts. When I opened the web interface, I added the following two files as context.

owners.csv

account_id,team_name,team_lead,email
012345678910,senior devops,Мартин Нанчев,finops-lead@example.com

context.md

# Custom instructions: anomaly notification routing

## Owner attribution
- Use the account-to-owner mapping context file to attribute all cost
  anomalies and spend changes to the owner of the affected account.
- Account 012345678910 is owned by Martin Nanchev (senior devops).

## Slack notification behavior
- When posting a cost anomaly investigation or cost report to Slack,
  begin the message with a line that notifies the responsible account owner.
- For account 012345678910, start the Slack message with: <@U0BA80JFFRP>
- Follow the mention with a one-line summary: affected account, owner name,
  estimated cost impact, and the anomaly trigger.

## Scope
- Apply this only to Slack posts for cost anomalies and scheduled cost reports.
- Do not include personal email addresses in any Slack message body.

Taking it for a spin

Investigating a real anomaly

With the initial configuration in place, let's investigate a cost anomaly I actually had in March 2026 with Amazon Bedrock (Sonnet).

A curious caveat: here the agent claims it can't search CloudTrail. Yet when I tested the exact same thing at work, in our dev organisation, it happily pulled data from CloudTrail. So I'm assuming it is possible — for some reason it works at work but not in my personal account. If you know why, drop a comment.

Preparing and sending the report

Let's prepare the report and send it to Slack:

And here's how it lands in Slack:

The report itself looks like it was generated straight from a CUDOS dashboard.

Tagging account owners

Can it tag account owners based on owners.csv and context.md? Of course — that's their whole purpose. Here's what it looks like:

Scheduling weekly automation

Last question: can you schedule an automation that posts anomalies to Slack every single week? Again — of course.

Here's an example prompt:

Analyse my top 3 cost drivers for the past 7 days. Include a breakdown by service and highlight any anomalies compared to the previous period. Notify the account in Slack.

And once again, it looks amazing — like a real dashboard.

Stress-testing it for optimisation tips

Now let's put it under a bit of stress. My personal account sits at a whopping $20–$60 a month (or $0 with AWS credits), so let's ask the agent for cost optimisation suggestions for Amazon Bedrock. It returns some solid best practices right away:

Wrapping up

And just like that, the walk of doom collapses into a single prompt. The AWS FinOps Agent won't replace your FinOps instincts — but feed it the right context and it absolutely sharpens them, turning hours of Cost Explorer archaeology into a report that practically writes itself.

Give it a try in a dev account, throw your own owners.csv and context.md at it, and let me know in the comments how it goes.

Securing and using Claude Code at scale

Martin Nanchev — Wed, 29 Apr 2026 11:09:38 +0000

How to enforce policies in Claude Code from a security perspective

A common take from security folks at conferences is that 'security lives in CLAUDE.md' Unfortunately, that has nothing to do with how Claude Code actually enforces policy.

If you're rolling out Claude Code across a company, CLAUDE.md is not your security layer. It's behavioral guidance — Claude reads it and tries to follow it. The thing that actually enforces is managed settings: a JSON file the Claude Code client honors regardless of what your devs put in their project or user settings.

Where it goes

Platform	Location
macOS	`/Library/Application Support/ClaudeCode/managed-settings.json` (or the `com.anthropic.claudecode` plist domain via MDM)
Linux / WSL	`/etc/claude-code/managed-settings.json`
Windows	`C:\Program Files\ClaudeCode\managed-settings.json` (the legacy `C:\ProgramData\ClaudeCode\` path was deprecated in v2.1.2 and is no longer supported as of v2.1.75)
Windows (registry)	`HKLM\SOFTWARE\Policies\ClaudeCode` → `Settings` value (`REG_SZ` or `REG_EXPAND_SZ`) containing JSON

Precedence inside the managed tier, highest to lowest:

server-managed (Claude.ai admin console)
  → endpoint-managed via MDM / OS-level policies (plist, HKLM)
    → file-based (managed-settings.json + managed-settings.d/*.json)
      → HKCU registry (Windows only, lowest managed priority)

Only one source within the managed tier wins — sources do not merge across the managed sub-tiers (it's first-source-wins: the moment a higher-priority source delivers a non-empty configuration, lower ones are ignored entirely).

However, array values like permissions.deny do merge across all settings scopes (managed + project + user + local), concatenated and deduplicated. So a managed deny list is a floor, not a ceiling.

For multi-team policy ownership (security team owns one slice, platform team owns another), drop fragments into managed-settings.d/ next to the base file. They merge in the systemd/sudoers drop-in style: alphabetical, scalars override, arrays concat, objects deep-merge.

Minimum-viable baseline

You can hand-write it or use a generator

Managed Settings generator (third-party, not affiliated with Anthropic — review output before deploying)

Here's the minimum-viable baseline I'd ship via Jamf, Intune, Group Policy, or Ansible or just use it locally.

{
  "$schema": "https://schemas.anthropic.com/claude-code/settings.json",

  "forceLoginMethod": "claudeai",
  "forceLoginOrgUUID": "YOUR-ORG-UUID-HERE",
  "minimumVersion": "2.1.75",
  "forceRemoteSettingsRefresh": true,

  "permissions": {
    "deny": [
      "Read(~/.ssh/**)",
      "Read(~/.aws/credentials)",
      "Read(~/.aws/config)",
      "Read(./.env)",
      "Read(./.env.*)",
      "Read(./secrets/**)",
      "Read(//etc/passwd)",
      "Read(//etc/shadow)",
      "WebFetch",
      "Bash(curl:*)",
      "Bash(wget:*)"
    ],
    "ask": ["Write(**)", "Edit(**)"],
    "disableBypassPermissionsMode": "disable"
  },
  "allowManagedPermissionRulesOnly": true,
  "disableAutoMode": "disable",

  "sandbox": {
    "enabled": true,
    "failIfUnavailable": true,
    "allowUnsandboxedCommands": false,
    "network": {
      "allowedDomains": [
        "github.com",
        "*.githubusercontent.com",
        "registry.npmjs.org",
        "pypi.org",
        "*.pythonhosted.org"
      ],
      "allowManagedDomainsOnly": true
    }
  },

  "allowManagedHooksOnly": true,
  "allowManagedMcpServersOnly": true,
  "allowedMcpServers": [
    { "serverName": "github" }
  ],
  "strictKnownMarketplaces": [
    { "source": "github", "repo": "your-org/claude-plugins" }
  ],
  "disableSkillShellExecution": true,

  "wslInheritsWindowsSettings": true,
  "cleanupPeriodDays": 7,

  "companyAnnouncements": [
    "Claude Code is governed by org policy. Report issues to #devops."
  ]
}

Note on Bash rule syntax: the documented specifier form uses a colon — Bash(curl:*), Bash(wget:*), Bash(sudo:*). The space form (Bash(curl *)) is also commonly seen in older examples and may still match, but prefer the colon form for new policies.

What each block actually does

Identity lock. forceLoginMethod + forceLoginOrgUUID prevent a dev from signing in with a personal Claude.ai account on a corp machine. This is the single highest-leverage anti-data-leak control. Without it, the rest is theatre.

Version floor. minimumVersion blocks the CLI from running below the specified version. forceRemoteSettingsRefresh: true (added in v2.1.92) makes the CLI fail closed at startup if it can't pull fresh server-managed policy — the CLI exits rather than proceeding without enforcement.

Permission rules. Use Tool(specifier) syntax — Read(...), Write(...), Edit(...), Bash(...), WebFetch. Single-slash paths are project-relative; double-slash (//etc/passwd) is absolute; ~/ is home. allowManagedPermissionRulesOnly: true causes Claude Code to ignore permission rules from user, project, and local settings, plus CLI arguments — only managed rules apply.

Caveat on Bash denies. Bash(curl:*) is a textual match against the command. Claude can route around a single-binary deny by using wget, python -c "import urllib...", nc, or shell tricks. Treat Bash denies as a soft layer; the real network control is sandbox.network.allowedDomains + allowManagedDomainsOnly.

Sandbox. This is the OS-level boundary (macOS Seatbelt, Linux/WSL2 bubblewrap), not a behavioral request. failIfUnavailable: true turns it into a hard gate — the CLI exits if sandboxing can't start. allowUnsandboxedCommands: false removes the dangerouslyDisableSandbox per-command escape hatch.

Known limitation: sandbox network allowlists operate at the system-call level (intercepting libc DNS resolution). Tools that use their own resolver path — notably some Node.js processes via libuv — can fail to resolve even allowed domains. Test your toolchain against the allowlist before committing.

Bypass and auto mode. disableBypassPermissionsMode: "disable" kills --dangerously-skip-permissions. disableAutoMode: "disable" removes auto mode from the Shift+Tab cycle and rejects --permission-mode auto.

Hooks, MCP, plugins. Three independent extensibility surfaces, three independent allowlists:

allowManagedHooksOnly — blocks user/project hooks; only managed hooks and hooks from plugins force-enabled in managed enabledPlugins will run.
allowManagedMcpServersOnly + allowedMcpServers — locks MCP servers to a vetted list. Without this, any .mcp.json in any repo can connect to anything.
strictKnownMarketplaces — plugin marketplace allowlist, enforced before network or filesystem ops. Use [] for full lockdown.

Skills. disableSkillShellExecution: true blocks inline !`...` shell execution in skills loaded from user, project, plugin, and additional-directory sources. Managed and bundled skills are exempt. Personally I'd leave this off unless you have a specific threat model that requires it.

WSL inheritance. wslInheritsWindowsSettings: true (added in v2.1.118; only honored when set in HKLM or C:\Program Files\ClaudeCode\managed-settings.json) makes WSL Claude Code read the Windows policy chain in addition to /etc/claude-code. Critical if your devs work in WSL.

CLAUDE.md is not in this file for a reason

Managed CLAUDE.md exists too (/Library/Application Support/ClaudeCode/CLAUDE.md, /etc/claude-code/CLAUDE.md, C:\Program Files\ClaudeCode\CLAUDE.md), and unlike user/project memory files, it cannot be excluded by claudeMdExcludes.

⚠️ Verify against current docs: earlier write-ups described CLAUDE.md as being injected as a user message after the system prompt with no strict-compliance guarantee. The exact delivery mechanism may have changed across versions. The takeaway that matters either way: CLAUDE.md is guidance the model interprets, not a hard control.

Use managed CLAUDE.md for behavioral guidance: coding standards, "always cite the internal style guide," compliance reminders. Use managed settings for technical enforcement: tool denies, sandbox, login lock, MCP allowlist.

If you write "never read .env files" in a managed CLAUDE.md, Claude will usually obey. If you put Read(./.env) in permissions.deny, the client physically refuses. Pick the right tool.

Verify it landed

On any dev machine, run /status inside Claude Code. It lists each active settings source with its origin — Enterprise managed settings (HKLM), Enterprise managed settings (file), Enterprise managed settings (remote), Enterprise managed settings (plist), etc. If your managed file has a syntax error, /status reports it.

For deeper debugging of which CLAUDE.md / rules / settings actually loaded, the InstructionsLoaded hook logs every instruction file at load time.

That's the baseline. Tighten it for your threat model: scope allowedDomains to your internal package mirror, lock availableModels to a Bedrock inference profile if you're routing through your own AWS account, add env blocks for OTEL_* variables to ship telemetry to your stack.

How to set up behaviors in Claude Code from a dev perspective

First, I'd enable forked subagents and agent teams by setting these environment variables:

CLAUDE_CODE_FORK_SUBAGENT=1
CLAUDE_CODE_EXPERIMENTAL_AGENT_TEAMS=1

Heads-up on cost: fork mode lets parallel children share the parent's prompt cache prefix (which can dramatically reduce input token cost for children 2..N), but agent teams running in parallel still multiply overall token consumption. Be careful when adding agents in /agents.

This lets you run teams of agents and parallelize work where possible.

Next, focus on CLAUDE.md. The most specific CLAUDE.md takes precedence — yours can be sharper than the default. Example:

# prompts

Review this plan thoroughly before making any code changes. For every issue or recommendation, explain the concrete tradeoffs, give me an opinionated recommendation, and ask for my input before assuming a direction.

My engineering preferences (use these to guide your recommendations):

- DRY is important — flag repetition aggressively.
- Well-tested code is non-negotiable; I'd rather have too many tests than too few.
- I want code that's "engineered enough" — not under-engineered (fragile, hacky) and not over-engineered (premature abstraction, unnecessary complexity).
- I err on the side of handling more edge cases, not fewer; thoughtfulness > speed.
- Bias toward explicit over clever.

## 1. Architecture review

Evaluate:

- Overall system design and component boundaries.
- Dependency graph and coupling concerns.
- Data flow patterns and potential bottlenecks.
- Scaling characteristics and single points of failure.
- Security architecture (auth, data access, API boundaries).

## 2. Code quality review

Evaluate:

- Code organization and module structure.
- DRY violations — be aggressive here.
- Error handling patterns and missing edge cases (call these out explicitly).
- Technical debt hotspots.
- Areas that are over-engineered or under-engineered relative to my preferences.

## 3. Test review

Evaluate:

- Test coverage gaps (unit, integration, e2e).
- Test quality and assertion strength.
- Missing edge case coverage — be thorough.
- Untested failure modes and error paths.

## 4. Performance review

Evaluate:

- N+1 queries and database access patterns.
- Memory-usage concerns.
- Caching opportunities.
- Slow or high-complexity code paths.

## For each issue you find

For every specific issue (bug, smell, design concern, or risk):

- Describe the problem concretely, with file and line references.
- Present 2–3 options, including "do nothing" where that's reasonable.
- For each option, specify: implementation effort, risk, impact on other code, and maintenance burden.
- Give me your recommended option and why, mapped to my preferences above.
- Then explicitly ask whether I agree or want to choose a different direction before proceeding.

## Workflow and interaction

- Do not assume my priorities on timeline or scale.
- After each section, pause and ask for my feedback before moving on.

## BEFORE YOU START

Ask if I want one of two options:

1. **BIG CHANGE:** Work through this interactively, one section at a time (Architecture → Code Quality → Tests → Performance), with at most 4 top issues in each section.
2. **SMALL CHANGE:** Work through interactively, ONE question per review section.

## FOR EACH STAGE OF REVIEW

Output the explanation, the pros and cons of each stage's questions, AND your opinionated recommendation and why, as `AskUserQuestion`. Also NUMBER issues and use LETTERS for options. When using `AskUserQuestion`, make sure each option clearly labels the issue NUMBER and option LETTER so the user doesn't get confused. The recommended option should always be the 1st option.

The general guidance: don't put a lot in CLAUDE.md — it's there to guide behavior. For specific, modular capability components, use skills. Or shorter:

SKILL.md = specific, modular capability components
CLAUDE.md = high-level behavioral and workflow guidance

I use skills for specific tasks and requirements - finops with specific tool, security reviews with specific tools, everything that could be described as process or workflow could be injected to the context via the skill

References

Source of truth: code.claude.com/docs/en/settings and code.claude.com/docs/en/permissions. The available-settings table is updated more frequently than any third-party article (including this one).
Managed policy generator (third-party)
Boris Cherny on Claude Code (YouTube)
How Boris uses Claude Code

Smarter Cloud Spending: FinOps Agent Powered by Strands SDK and Amazon Bedrock

Martin Nanchev — Mon, 27 Apr 2026 18:49:42 +0000

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.

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.01 per request). Every analyzer that touches money goes through these two helpers — _ce_groups for one-shot service-grouped cost windows and _ce_monthly_with_top for the trailing-N-months trend with the top services per month. If I let every tool issue its own raw get_cost_and_usage calls, I would be paying for the privilege of looking at my own waste. I also always strip out credits and refunds with NO_CREDITS, otherwise the trend chart shows phantom drops every time AWS applies a credit and the agent gets confused about what the spend actually looks like.



def _ce_groups(start: date, end: date, key: str, group_type: str = "DIMENSION") -> dict[str, float]:
    """Run a CE GetCostAndUsage and flatten Service-grouped results to {service: usd}."""
    r = try_aws(
        lambda: s().client("ce").get_cost_and_usage(
            TimePeriod={"Start": start.isoformat(), "End": end.isoformat()},
            Granularity="MONTHLY",
            Metrics=["UnblendedCost"],
            GroupBy=[{"Type": group_type, "Key": key}],
            Filter=NO_CREDITS,
        ),
        {"ResultsByTime": []},
    )
    out: dict[str, float] = {}
    for tp in r.get("ResultsByTime", []):
        for g in tp.get("Groups", []):
            k = g["Keys"][0]
            out[k] = out.get(k, 0.0) + money(g["Metrics"]["UnblendedCost"]["Amount"])
    return out


def _ce_monthly_with_top(months: int) -> list[dict]:
    """Last N months grouped by service, with top 8 services per month."""
    end = today() + timedelta(days=1)
    start = (today() - timedelta(days=30 * months)).replace(day=1)
    r = try_aws(
        lambda: s().client("ce").get_cost_and_usage(
            TimePeriod={"Start": start.isoformat(), "End": end.isoformat()},
            Granularity="MONTHLY",
            Metrics=["UnblendedCost"],
            GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
            Filter=NO_CREDITS,
        ),
        {"ResultsByTime": []},
    )
    out = []
    for tp in r.get("ResultsByTime", []):
        services = sorted(
            ({"service": g["Keys"][0],
              "cost_usd": money(g["Metrics"]["UnblendedCost"]["Amount"])}
             for g in tp.get("Groups", [])),
            key=lambda x: -x["cost_usd"],
        )
        out.append({
            "month": tp["TimePeriod"]["Start"],
            "total_usd": round(sum(svc["cost_usd"] for svc in services), 2),
            "top": services[:8],
        })
    return out

Bootstrap: let the spend tell you which regions matter

Now the bootstrap. This is where the agent figures out which AWS account it is even talking to and which regions are worth scanning. The trick is _detect_active_regions — instead of scanning every AWS region (28+ at the time of writing), I ask Cost Explorer which regions had non-zero spend in the last 30 days and only scan those. A personal dev account ends up with one region. An enterprise workload ends up with eight. The agent does not need to know in advance which scenario it is in; the data tells it. The same build_session also handles the cross-account case via --role-arn, which was the original use case — running the agent against customer accounts where I had audit access.



def build_session(profile: str | None = None, role_arn: str | None = None,
                  region: str = None, explicit_regions: list[str] | None = None) -> Session:
    region = region or CONFIG["region"]
    base = boto3.Session(profile_name=profile) if profile else boto3.Session()

    if role_arn:
        creds = base.client("sts", region_name=region).assume_role(
            RoleArn=role_arn, RoleSessionName="finops-agent",
        )["Credentials"]
        sess = boto3.Session(
            aws_access_key_id=creds["AccessKeyId"],
            aws_secret_access_key=creds["SecretAccessKey"],
            aws_session_token=creds["SessionToken"],
            region_name=region,
        )
    else:
        sess = base

    account_id = sess.client("sts", region_name=region).get_caller_identity()["Account"]
    aliases = try_aws(lambda: sess.client("iam").list_account_aliases().get("AccountAliases", []), [])
    alias = aliases[0] if aliases else None

    regions = explicit_regions or _detect_active_regions(sess, region)
    return Session(sess, account_id, alias, region, regions[:CONFIG["max_regions"]])


def _detect_active_regions(sess: boto3.Session, primary: str) -> list[str]:
    """Use CE to find regions with >$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 — compute spend too low to justify a commitment"
        if ondemand < 50 else
        "consider Compute Savings Plan — OnDemand compute > $50/mo"
    )
    return record("commitments", {
        "savings_plans_coverage": sp_summary,
        "reservation_coverage": ri_summary,
        "ondemand_compute_30d_usd": round(ondemand, 2),
        "recommendation": rec,
    })

Compute: where Strands' model-driven approach earns its keep

I could have written one giant analyze_compute tool that returned a 50-key dictionary, and the model would have done its best with it — but the model reasons better when each tool has a small, clear scope. So analyze_compute_optimizer is a thin wrapper over the AWS Compute Optimizer API across six resource types, and analyze_lambda does what Compute Optimizer cannot: it counts the x86_64-vs-arm64 split (Graviton migration opportunity, usually a 20% saving for the same workload), finds SnapStart "On" published versions that nobody invokes (perpetual cache cost that the team forgot about), and lists the EOL runtimes that AWS will eventually force off the platform anyway. This is the "right tools for a single agent" idea in practice — surgical scopes, not a swarm.



@tool
def analyze_compute_optimizer() -> dict:
    """AWS Compute Optimizer recommendations across EC2, ASG, EBS, Lambda, ECS, RDS."""
    co = s().client("compute-optimizer", region=s().primary_region)
    enrollment = try_aws(lambda: co.get_enrollment_status(), {})
    if enrollment.get("status") not in ("Active", "Pending"):
        return record("compute_optimizer", {"enrolled": False, "note": "Compute Optimizer not enabled"})

    operations = [
        ("ec2",    "get_ec2_instance_recommendations",        "instanceRecommendations"),
        ("asg",    "get_auto_scaling_group_recommendations",  "autoScalingGroupRecommendations"),
        ("ebs",    "get_ebs_volume_recommendations",          "volumeRecommendations"),
        ("lambda", "get_lambda_function_recommendations",     "lambdaFunctionRecommendations"),
        ("ecs",    "get_ecs_service_recommendations",         "ecsServiceRecommendations"),
        ("rds",    "get_rds_database_recommendations",        "rdsDBRecommendations"),
    ]
    summary: dict = {}
    for label, op, key in operations:
        recs = try_aws(lambda op=op, key=key: getattr(co, op)(maxResults=50).get(key, []), [])
        summary[label] = {
            "count": len(recs),
            "samples": [
                {
                    "resource": (
                        r.get("instanceArn") or r.get("functionArn") or r.get("volumeArn")
                        or r.get("autoScalingGroupArn") or r.get("serviceArn") or r.get("resourceArn")
                    ),
                    "finding": r.get("finding") or r.get("findingClassification") or r.get("findingReasonCodes"),
                    "current": r.get("currentInstanceType") or r.get("currentConfiguration"),
                }
                for r in recs[:5]
            ],
        }

    return record("compute_optimizer", {
        "enrolled": True,
        "total_recommendations": sum(v["count"] for v in summary.values()),
        "by_resource_type": summary,
    })


@tool
def analyze_lambda() -> dict:
    """Lambda inventory: x86_64 vs arm64 split, SnapStart-On orphan published versions
    (perpetual cache cost), EOL runtimes."""
    snapstart_drift: list[dict] = []
    runtimes_at_eol: list[dict] = []

    def scan(region: str, lam) -> dict:
        fns = paginate(lam, "list_functions", "Functions")
        rt = {"functions": len(fns), "x86_64": 0, "arm64": 0, "snapstart_orphans": 0}
        for fn in fns:
            arch = (fn.get("Architectures") or ["x86_64"])[0]
            rt["x86_64" if arch == "x86_64" else "arm64"] += 1
            if fn.get("Runtime") in EOL_LAMBDA_RUNTIMES:
                runtimes_at_eol.append({
                    "region": region, "function": fn["FunctionName"], "runtime": fn["Runtime"],
                })
            versions = try_aws(
                lambda fn=fn: lam.list_versions_by_function(FunctionName=fn["FunctionName"])["Versions"],
                [],
            )
            snap = [
                v["Version"] for v in versions
                if v["Version"] != "$LATEST"
                and v.get("SnapStart", {}).get("OptimizationStatus") == "On"
            ]
            if snap:
                rt["snapstart_orphans"] += len(snap)
                snapstart_drift.append({
                    "region": region, "function": fn["FunctionName"], "snapstart_versions": snap,
                })
        return rt

    by_region = per_region("lambda", scan)
    totals = {k: sum(r.get(k, 0) for r in by_region.values())
              for k in ("functions", "snapstart_orphans", "x86_64", "arm64")}

    return record("lambda", {
        "totals": totals,
        "by_region": by_region,
        "snapstart_drift": snapstart_drift[:20],
        "runtimes_at_eol": runtimes_at_eol[:20],
        "graviton_opportunity_count": totals["x86_64"],
    })

Storage: the single biggest hidden cost

From my experience S3 is the single biggest hidden cost in most AWS accounts and the easiest to fix — one missing lifecycle policy on a 200 GB log bucket can be more than half of an account's monthly waste. The S3 analyzer walks every bucket, gets its size from CloudWatch (which is the only way to get bucket size without listing every object), checks lifecycle rules, versioning, Intelligent-Tiering, and the public-access-block, and explicitly flags any bucket over 1 GB without any lifecycle rule. EBS gets the same treatment but with three flag categories: unattached volumes (paying for nothing), gp2 volumes (always cheaper as gp3 with a one-line API call), and snapshots older than 90 days that nobody remembers creating.



def _bucket_size_bytes(cw, bucket: str) -> int:
    pts = cw_stats(
        cw, "AWS/S3", "BucketSizeBytes",
        [{"Name": "BucketName", "Value": bucket}, {"Name": "StorageType", "Value": "StandardStorage"}],
        days=2, stat="Maximum",
    )
    return int(max(pts)) if pts else 0


def _bucket_lifecycle_actions(rules: list[dict]) -> list[str]:
    keys = ("Expiration", "Transitions", "AbortIncompleteMultipartUpload", "NoncurrentVersionExpiration")
    return sorted({k for r in rules for k in keys if r.get(k)})


@tool
def analyze_storage_s3() -> dict:
    """S3 bucket inventory: size (CloudWatch), lifecycle rules + actions, versioning,
    Intelligent-Tiering, public-access-block. Flags any bucket >1 GB without any
    lifecycle rule (usually the biggest single S3 win)."""
    sess = s()
    s3 = sess.client("s3", region=sess.primary_region)
    raw_buckets = try_aws(lambda: s3.list_buckets().get("Buckets", []), [])

    findings: list[dict] = []
    no_lc_over_1gb: list[dict] = []
    total_bytes = 0

    for b in raw_buckets:
        name = b["Name"]
        loc = try_aws(lambda: s3.get_bucket_location(Bucket=name).get("LocationConstraint"), None) or "us-east-1"
        if loc == "EU":
            loc = "eu-west-1"

        size = try_aws(lambda: _bucket_size_bytes(sess.client("cloudwatch", loc), name), 0)
        rules = try_aws(lambda: s3.get_bucket_lifecycle_configuration(Bucket=name).get("Rules", []), [])
        versioning = try_aws(lambda: s3.get_bucket_versioning(Bucket=name).get("Status"), None)
        pab = try_aws(
            lambda: s3.get_public_access_block(Bucket=name).get("PublicAccessBlockConfiguration", {}),
            {},
        )
        it = try_aws(
            lambda: s3.list_bucket_intelligent_tiering_configurations(Bucket=name).get(
                "IntelligentTieringConfigurationList", []),
            [],
        )

        gb = size / 1_073_741_824
        total_bytes += size
        finding = {
            "name": name,
            "region": loc,
            "size_gb": round(gb, 3),
            "lifecycle_rules": len(rules),
            "lifecycle_actions": _bucket_lifecycle_actions(rules),
            "versioning": versioning,
            "intelligent_tiering_configs": len(it),
            "public_access_block_full": bool(pab) and all(pab.values()),
        }
        findings.append(finding)
        if gb > 1.0 and not rules:
            no_lc_over_1gb.append({"bucket": name, "size_gb": round(gb, 3), "region": loc})

    findings.sort(key=lambda x: -x["size_gb"])
    return record("s3", {
        "bucket_count": len(raw_buckets),
        "total_size_gb": round(total_bytes / 1_073_741_824, 2),
        "buckets": findings[:30],
        "no_lifecycle_over_1gb": no_lc_over_1gb,
        "no_lifecycle_count": len(no_lc_over_1gb),
    })


@tool
def analyze_storage_ebs() -> dict:
    """EBS volumes (status, type, age) + snapshots (age) per region. Flags
    unattached volumes, gp2 (gp3 migration), snapshots > 90 days."""
    snap_cutoff = datetime.now(timezone.utc) - timedelta(days=90)
    available_volumes: list[dict] = []
    gp2_volumes: list[dict] = []
    old_snapshots: list[dict] = []

    def scan(region: str, ec2) -> dict:
        vols = paginate(ec2, "describe_volumes", "Volumes")
        snaps = paginate(ec2, "describe_snapshots", "Snapshots", OwnerIds=[s().account_id])
        rt = {"volumes": len(vols), "available": 0, "gp2": 0,
              "snapshots": len(snaps), "old_snapshots": 0}
        for v in vols:
            if v["State"] == "available":
                rt["available"] += 1
                available_volumes.append({"region": region, "id": v["VolumeId"],
                                          "size_gb": v["Size"], "type": v["VolumeType"]})
            if v["VolumeType"] == "gp2":
                rt["gp2"] += 1
                gp2_volumes.append({"region": region, "id": v["VolumeId"], "size_gb": v["Size"]})
        for sn in snaps:
            if sn["StartTime"] < snap_cutoff:
                rt["old_snapshots"] += 1
                old_snapshots.append({
                    "region": region, "id": sn["SnapshotId"],
                    "age_days": (datetime.now(timezone.utc) - sn["StartTime"]).days,
                    "size_gb": sn["VolumeSize"],
                })
        return rt

    by_region = per_region("ec2", scan)
    totals = {
        "volumes": sum(r.get("volumes", 0) for r in by_region.values()),
        "available_volumes": sum(r.get("available", 0) for r in by_region.values()),
        "gp2_volumes": sum(r.get("gp2", 0) for r in by_region.values()),
        "old_snapshots": sum(r.get("old_snapshots", 0) for r in by_region.values()),
    }
    return record("ebs", {
        "by_region": by_region,
        "totals": totals,
        "available_volumes": available_volumes[:20],
        "gp2_volumes": gp2_volumes[:20],
        "old_snapshots": old_snapshots[:20],
    })

Databases and network: same shape, different problem

Databases and network are different problems but they share the same shape — find the resource, get a CloudWatch metric, decide whether the resource is doing work. For RDS, "doing work" means more than one average database connection over 14 days; below that threshold, the instance is a candidate to stop or downsize. For NAT gateways, it means more than 1 GB of outbound traffic over the same window; an idle NAT gateway is around $32/month for nothing. The network tool also catches unattached Elastic IPs and load balancers with zero registered targets, which are the kind of resources you forgot you provisioned during a project that got cancelled six months ago.



@tool
def analyze_databases() -> dict:
    """RDS instances + DynamoDB tables. RDS: idle detection (avg connections < 1
    over 14 days). DynamoDB: billing mode, size, item count."""
    sess = s()
    idle_rds: list[dict] = []
    rds_by_region: dict[str, dict] = {}
    ddb_tables: list[dict] = []

    for region in sess.regions:
        rds = sess.client("rds", region)
        cw = sess.client("cloudwatch", region)
        instances = try_aws(lambda: paginate(rds, "describe_db_instances", "DBInstances"), [])
        engines: dict[str, int] = {}
        for inst in instances:
            engines[inst["Engine"]] = engines.get(inst["Engine"], 0) + 1
            if inst.get("DBInstanceStatus") != "available":
                continue
            pts = cw_stats(
                cw, "AWS/RDS", "DatabaseConnections",
                [{"Name": "DBInstanceIdentifier", "Value": inst["DBInstanceIdentifier"]}],
                days=CONFIG["idle_days"],
            )
            avg = sum(pts) / len(pts) if pts else 0
            if avg < 1:
                idle_rds.append({
                    "region": region, "id": inst["DBInstanceIdentifier"],
                    "class": inst["DBInstanceClass"], "engine": inst["Engine"],
                    "avg_connections_14d": round(avg, 2), "storage_gb": inst.get("AllocatedStorage"),
                })
        rds_by_region[region] = {"instance_count": len(instances), "engines": engines}

        ddb = sess.client("dynamodb", region)
        for name in try_aws(lambda: paginate(ddb, "list_tables", "TableNames"), [])[:50]:
            desc = try_aws(lambda n=name: ddb.describe_table(TableName=n)["Table"], None)
            if not desc:
                continue
            ddb_tables.append({
                "region": region, "name": name,
                "billing_mode": desc.get("BillingModeSummary", {}).get("BillingMode") or "PROVISIONED",
                "size_bytes": desc.get("TableSizeBytes", 0),
                "item_count": desc.get("ItemCount", 0),
            })

    return record("databases", {
        "rds": {"by_region": rds_by_region, "idle_candidates": idle_rds[:20]},
        "dynamodb": {"tables": ddb_tables[:30]},
    })



@tool
def analyze_network() -> dict:
    """NAT gateways with low traffic, unattached EIPs, ALB/NLB with no targets."""
    sess = s()
    unattached_eips: list[dict] = []
    empty_lbs: list[dict] = []
    low_traffic_nat: list[dict] = []
    totals = {"nat_gateways": 0, "elbs": 0, "eips": 0}

    for region in sess.regions:
        ec2 = sess.client("ec2", region)
        elbv2 = sess.client("elbv2", region)
        cw = sess.client("cloudwatch", region)

        eips = try_aws(lambda: ec2.describe_addresses().get("Addresses", []), [])
        nats = try_aws(lambda: ec2.describe_nat_gateways().get("NatGateways", []), [])
        lbs  = try_aws(lambda: elbv2.describe_load_balancers().get("LoadBalancers", []), [])

        totals["eips"] += len(eips)
        totals["nat_gateways"] += sum(1 for n in nats if n["State"] == "available")
        totals["elbs"] += len(lbs)

        for eip in eips:
            if not eip.get("AssociationId"):
                unattached_eips.append({
                    "region": region, "ip": eip.get("PublicIp"),
                    "allocation_id": eip.get("AllocationId"),
                })

        for lb in lbs:
            tgs = try_aws(
                lambda lb=lb: elbv2.describe_target_groups(LoadBalancerArn=lb["LoadBalancerArn"])
                                   .get("TargetGroups", []),
                [],
            )
            target_count = sum(
                len(try_aws(lambda tg=tg: elbv2.describe_target_health(TargetGroupArn=tg["TargetGroupArn"])
                                            .get("TargetHealthDescriptions", []),
                            []))
                for tg in tgs
            )
            if target_count == 0:
                empty_lbs.append({"region": region, "name": lb["LoadBalancerName"], "type": lb["Type"]})

        for n in nats:
            if n["State"] != "available":
                continue
            pts = cw_stats(
                cw, "AWS/NATGateway", "BytesOutToDestination",
                [{"Name": "NatGatewayId", "Value": n["NatGatewayId"]}],
                days=CONFIG["idle_days"], stat="Sum",
            )
            outbound_gb = sum(pts) / 1_073_741_824
            if outbound_gb < 1:
                low_traffic_nat.append({
                    "region": region, "id": n["NatGatewayId"],
                    "outbound_gb_14d": round(outbound_gb, 3),
                })

    return record("network", {
        "totals": totals,
        "unattached_eips": unattached_eips[:15],
        "empty_load_balancers": empty_lbs[:15],
        "low_traffic_nat_gateways": low_traffic_nat[:15],
    })

Logs, KMS, secrets: small line items that compound

Logs, KMS, and secrets are individually small but they compound across regions and accounts. A log group without retention grows forever ($0.03/GB/month — small until it isn't). An orphan /aws/lambda/* log group from a function that was deleted three years ago is pure waste. A customer-managed KMS key with no grants and no aliases is $1/month for nothing — but I am careful to flag the verification step before deletion in the tool's own docstring, because a key with no grants may still be referenced by encrypted objects somewhere, and deleting it is irreversible. A secret in Secrets Manager that has not been read in 90 days is $0.40/month per stale secret. None of these alone is interesting. Together, in an account with hundreds of resources, they often pay for a small EC2 instance.



@tool
def analyze_logs_and_metrics() -> dict:
    """CloudWatch Logs hygiene + custom metrics + dashboards.
    - log groups missing retention (= infinite growth)
    - /aws/lambda/* log groups whose function no longer exists
    - top 20 log groups by stored bytes
    - custom metric namespaces (cost = $0.30/metric/mo above 10 free)
    - dashboards count ($3 each above 3 free)"""
    sess = s()
    missing_retention: list[dict] = []
    orphan_lambda_lg: list[dict] = []
    largest: list[dict] = []
    custom_namespaces: dict[str, int] = {}
    dashboards: list[dict] = []
    by_region: dict[str, dict] = {}
    totals = {"log_groups": 0, "missing_retention_count": 0,
              "orphan_lambda_count": 0, "stored_gb": 0.0, "dashboards": 0}

    for region in sess.regions:
        logs = sess.client("logs", region)
        cw = sess.client("cloudwatch", region)
        lam = sess.client("lambda", region)

        lambda_names = {f["FunctionName"]
                        for f in try_aws(lambda: paginate(lam, "list_functions", "Functions"), [])}
        log_groups = try_aws(lambda: paginate(logs, "describe_log_groups", "logGroups"), [])

        rt = {"log_groups": len(log_groups), "missing_retention": 0, "orphan_lambda": 0, "stored_gb": 0.0}
        for lg in log_groups:
            name = lg["logGroupName"]
            stored = lg.get("storedBytes", 0)
            gb = round(stored / 1_073_741_824, 4)
            rt["stored_gb"] += gb
            entry = {"region": region, "name": name, "stored_gb": gb}
            largest.append(entry)
            if lg.get("retentionInDays") is None:
                rt["missing_retention"] += 1
                missing_retention.append(entry)
            if name.startswith("/aws/lambda/"):
                fn_name = name[len("/aws/lambda/"):]
                if fn_name not in lambda_names:
                    rt["orphan_lambda"] += 1
                    orphan_lambda_lg.append(entry)
        by_region[region] = rt
        totals["log_groups"] += rt["log_groups"]
        totals["missing_retention_count"] += rt["missing_retention"]
        totals["orphan_lambda_count"] += rt["orphan_lambda"]
        totals["stored_gb"] = round(totals["stored_gb"] + rt["stored_gb"], 2)

        for m in try_aws(lambda: cw.list_metrics().get("Metrics", []), []):
            ns = m.get("Namespace", "")
            if not ns.startswith("AWS/"):
                custom_namespaces[ns] = custom_namespaces.get(ns, 0) + 1
        region_dashboards = try_aws(lambda: cw.list_dashboards().get("DashboardEntries", []), [])
        dashboards.extend({"region": region, "name": d["DashboardName"]} for d in region_dashboards)
        totals["dashboards"] += len(region_dashboards)

    largest.sort(key=lambda x: -x["stored_gb"])
    return record("logs", {
        "by_region": by_region,
        "totals": totals,
        "missing_retention": missing_retention[:20],
        "orphan_lambda_log_groups": orphan_lambda_lg[:20],
        "largest_log_groups": largest[:20],
        "custom_metric_namespaces": custom_namespaces,
        "dashboards": dashboards,
    })



@tool
def analyze_kms() -> dict:
    """Customer-managed KMS keys with 0 grants AND 0 aliases (deletion candidates).
    Operator must verify against CloudTrail Decrypt events before scheduling deletion —
    a key with no grants/aliases may still be referenced by encrypted data."""
    candidates: list[dict] = []

    def scan(region: str, kms) -> dict:
        keys = paginate(kms, "list_keys", "Keys")
        aliases = paginate(kms, "list_aliases", "Aliases")
        alias_by_target: dict[str, list[str]] = {}
        for a in aliases:
            tk = a.get("TargetKeyId")
            if tk:
                alias_by_target.setdefault(tk, []).append(a["AliasName"])

        rt = {"customer_managed": 0, "candidates": 0}
        for k in keys:
            kid = k["KeyId"]
            meta = try_aws(lambda kid=kid: kms.describe_key(KeyId=kid)["KeyMetadata"], None)
            if not meta or meta.get("KeyManager") != "CUSTOMER" or meta.get("KeyState") != "Enabled":
                continue
            rt["customer_managed"] += 1
            grants = try_aws(lambda kid=kid: kms.list_grants(KeyId=kid).get("Grants", []), [])
            if not grants and not alias_by_target.get(kid):
                rt["candidates"] += 1
                candidates.append({
                    "region": region, "key_id": kid,
                    "created": meta["CreationDate"].isoformat(),
                    "description": meta.get("Description") or "(none)",
                })
        return rt

    by_region = per_region("kms", scan)
    totals = {
        "customer_managed": sum(r.get("customer_managed", 0) for r in by_region.values()),
        "candidates":       sum(r.get("candidates", 0) for r in by_region.values()),
    }
    return record("kms", {
        "by_region": by_region,
        "totals": totals,
        "candidates_for_deletion": candidates[:30],
    })


@tool
def analyze_secrets() -> dict:
    """Secrets Manager: secrets not accessed in N days (default 90).
    Saves $0.40/secret/mo per stale secret deleted."""
    cutoff = datetime.now(timezone.utc) - timedelta(days=CONFIG["stale_days"])
    stale: list[dict] = []

    def scan(region: str, sm) -> dict:
        secrets = paginate(sm, "list_secrets", "SecretList")
        rt = {"secrets": len(secrets), "stale": 0}
        for sec in secrets:
            last = sec.get("LastAccessedDate")
            if last and last < cutoff:
                rt["stale"] += 1
                stale.append({
                    "region": region, "name": sec["Name"],
                    "last_accessed": last.isoformat(),
                    "days_since_access": (datetime.now(timezone.utc) - last).days,
                })
        return rt

    by_region = per_region("secretsmanager", scan)
    totals = {
        "secrets": sum(r.get("secrets", 0) for r in by_region.values()),
        "stale":   sum(r.get("stale", 0) for r in by_region.values()),
    }
    return record("secrets", {"by_region": by_region, "totals": totals, "stale": stale[:30]})

Security tooling and governance: paying for what you don't have

The next two tools step away from direct cost and into governance. Security tooling is something I check because most accounts have GuardDuty, Security Hub, Inspector, or Macie enabled with features that scan resources that do not exist (S3 protection enabled in an account with no buckets, EBS scanning where there are no EC2 instances, full-fat Macie sessions on accounts that store nothing sensitive). The governance tool looks at cost-allocation tags, because an account with zero active tags is an account where you cannot answer "what does this team's spend look like?" — and that is a problem long before it is a cost problem.



def _strip_arn_segment(arn: str) -> str:
    return arn.rsplit("/", 4)[-3] if "/" in arn else arn


@tool
def analyze_security_tooling() -> dict:
    """Security stack vs actual workload: Security Hub standards + AutoEnable,
    GuardDuty enabled features, Inspector resource scope, Macie status. Surfaces
    features enabled for infrastructure that doesn't exist (waste)."""
    sess = s()
    by_region: dict[str, dict] = {}

    for region in sess.regions:
        rt: dict = {}

        sh = sess.client("securityhub", region)
        hub = try_aws(lambda: sh.describe_hub(), None)
        if hub:
            stds = try_aws(lambda: sh.get_enabled_standards().get("StandardsSubscriptions", []), [])
            rt["security_hub"] = {
                "auto_enable_controls": hub.get("AutoEnableControls"),
                "control_finding_generator": hub.get("ControlFindingGenerator"),
                "standards": [_strip_arn_segment(s.get("StandardsArn", "")) for s in stds],
            }
        else:
            rt["security_hub"] = {"enabled": False}

        gd = sess.client("guardduty", region)
        ds = try_aws(lambda: gd.list_detectors().get("DetectorIds", []), [])
        if ds:
            d = try_aws(lambda: gd.get_detector(DetectorId=ds[0]), {})
            rt["guardduty"] = {
                "status": d.get("Status"),
                "features_enabled":  [f["Name"] for f in d.get("Features", []) if f.get("Status") == "ENABLED"],
                "features_disabled": [f["Name"] for f in d.get("Features", []) if f.get("Status") == "DISABLED"],
            }
        else:
            rt["guardduty"] = {"enabled": False}

        insp = sess.client("inspector2", region)
        accs = try_aws(
            lambda: insp.batch_get_account_status(accountIds=[sess.account_id]).get("accounts", []),
            [],
        )
        rt["inspector"] = (
            {"status": accs[0]["state"]["status"],
             "resources": {k: v["status"] for k, v in accs[0].get("resourceState", {}).items()}}
            if accs else {"enabled": False}
        )

        macie_sess = try_aws(lambda: sess.client("macie2", region).get_macie_session(), None)
        rt["macie"] = {"status": macie_sess.get("status")} if macie_sess else {"enabled": False}

        by_region[region] = rt

    return record("security_tooling", {"by_region": by_region})


@tool
def analyze_governance() -> dict:
    """Cost allocation tag activation + tag-coverage analysis. Reports % of last
    30 days of spend covered by each tag key."""
    sess = s()
    ce = sess.client("ce")
    end = today() + timedelta(days=1)
    start = today() - timedelta(days=30)

    active = try_aws(
        lambda: ce.list_cost_allocation_tags(Status="Active").get("CostAllocationTags", []),
        [],
    )
    all_tags = try_aws(lambda: ce.list_cost_allocation_tags().get("CostAllocationTags", []), [])
    inactive_user = [t for t in all_tags
                     if t.get("Type") == "UserDefined" and t.get("Status") == "Inactive"]

    coverage: dict = {}
    for t in active[:5]:
        r = try_aws(
            lambda t=t: ce.get_cost_and_usage(
                TimePeriod={"Start": start.isoformat(), "End": end.isoformat()},
                Granularity="MONTHLY",
                Metrics=["UnblendedCost"],
                GroupBy=[{"Type": "TAG", "Key": t["TagKey"]}],
                Filter=NO_CREDITS,
            ),
            None,
        )
        if not r:
            continue
        tagged = untagged = 0.0
        for tp in r.get("ResultsByTime", []):
            for g in tp.get("Groups", []):
                cost = money(g["Metrics"]["UnblendedCost"]["Amount"])
                if g["Keys"][0].endswith("$"):  # untagged comes back as "TagKey$"
                    untagged += cost
                else:
                    tagged += cost
        total = tagged + untagged
        coverage[t["TagKey"]] = {
            "tagged_usd": round(tagged, 2),
            "untagged_usd": round(untagged, 2),
            "tagged_pct": round(tagged / total * 100, 1) if total else 0,
        }

    cats = try_aws(
        lambda: ce.list_cost_category_definitions().get("CostCategoryReferences", []),
        [],
    )

    return record("governance", {
        "active_cost_allocation_tag_count": len(active),
        "active_cost_allocation_tags":      [t["TagKey"] for t in active],
        "inactive_user_defined_tag_count":  len(inactive_user),
        "inactive_user_defined_tags":       [t["TagKey"] for t in inactive_user],
        "tag_coverage_30d":                 coverage,
        "cost_categories":                  [c["Name"] for c in cats],
    })

The synthesis tool: one PDF, one call, at the end

This is the synthesis tool, and it is the only one that is not read-only. The agent calls render_pdf_report exactly once at the end, with the executive summary it has written and the prioritized recommendation list it has built across all the previous tools. The PDF gets saved locally and (optionally) a notification gets pushed to SNS, which is what makes this useful in a Lambda or scheduled run. I deliberately did not make the PDF generation part of any analysis tool — the model decides when it has enough information to render, and that decision is more accurate when it is a single explicit step instead of an implicit one.


@tool
def render_pdf_report(executive_summary: str, recommendations: list[dict]) -> dict:
    """Render the final PDF using ALL the findings collected so far + the agent's
    synthesis. Pass the executive summary text and a prioritized list of
    recommendations (each: {priority, title, savings_usd_per_month, rationale,
    action_steps}). Returns the saved PDF path.

    Args:
        executive_summary: 2-4 paragraph plain-text overview.
        recommendations: list of dicts with keys priority, title,
            savings_usd_per_month, rationale, action_steps.
    """
    sess = s()
    out_dir = Path(os.getenv("FINOPS_OUTPUT_DIR", str(CONFIG["output_dir"])))
    out_dir.mkdir(parents=True, exist_ok=True)

    label = sess.alias or sess.account_id
    safe_label = "".join(c if c.isalnum() else "-" for c in label)[:40].strip("-")
    pdf_path = out_dir / f"finops-{safe_label}-{today().isoformat()}.pdf"

    pdf_renderer.render(
        output_path=pdf_path,
        account_id=sess.account_id,
        account_alias=sess.alias,
        regions=sess.regions,
        run_at=datetime.now(timezone.utc),
        executive_summary=executive_summary,
        recommendations=recommendations,
        findings=[{"section": k, "data": v} for k, v in FINDINGS.items()],
    )

    sns_status = "skipped"
    if CONFIG["sns_topic_arn"]:
        sns_status = try_aws(
            lambda: (sess.client("sns").publish(
                TopicArn=CONFIG["sns_topic_arn"],
                Subject=f"FinOps report: {label}"[:99],
                Message=(f"FinOps report for {sess.account_id} ({label}).\n"
                         f"PDF: {pdf_path}\n\nExecutive summary:\n{executive_summary[:1000]}"),
            ) or "published"),
            "publish-failed",
        )

    return {
        "pdf_path": str(pdf_path),
        "size_bytes": pdf_path.stat().st_size,
        "page_count_estimate": max(1, len(FINDINGS) + 2),
        "sns": sns_status,
    }

Wiring the agent: the most underrated part of Strands

Now the part of the code that is the most Strands-specific and, in my opinion, the most underrated. The agent is just a system prompt, a model, and a list of tool functions. There is no orchestration layer, no agent graph, no router. The system prompt tells the agent the workflow order, the analysis principles, and the output quality rules — and Strands lets the model decide which tool to call next based on what it has already seen and what it still needs. This is what I meant earlier about "the right tools for a single agent." A swarm would not have helped here. The work is sequential by nature (you cannot synthesize recommendations until you have collected the findings), and adding more agents would just add coordination overhead and prompt-engineering surface area for no benefit. The model is the orchestrator.



SYSTEM_PROMPT = """You are a senior FinOps consultant performing a comprehensive cost-optimization review of an AWS account.

GOAL: Produce a thorough, actionable PDF report that any AWS account owner can use — whether it's a personal dev account ($20/mo) or an enterprise workload ($20k/mo). You don't know which until you discover it.

WORKFLOW (in order):
1. `discover_account` first — anchors account id, alias, active regions.
2. `analyze_cost_trends` — spend profile and recent deltas.
3. `analyze_anomalies_and_budgets`, `analyze_commitments` — cost-layer signals.
4. All inventory analyzers: `analyze_compute_optimizer`, `analyze_lambda`, `analyze_storage_s3`, `analyze_storage_ebs`, `analyze_databases`, `analyze_network`, `analyze_logs_and_metrics`, `analyze_kms`, `analyze_secrets`, `analyze_security_tooling`, `analyze_governance`.
5. Synthesize a prioritized recommendation list across sections, ordered by estimated savings descending.
6. Call `render_pdf_report` ONCE at the end with:
   - executive_summary: 2-4 paragraphs covering the spend profile, the 3 biggest opportunities, and the overall posture.
   - recommendations: list of {priority: 'high'|'medium'|'low', title, savings_usd_per_month, rationale, action_steps: list[str] of concrete CLI commands}.

ANALYSIS PRINCIPLES:
- Scale recommendations to spend. A $20 personal account doesn't need RI/SP analysis; a $20k workload absolutely does.
- Always identify the BIGGEST lever first. Most accounts have one issue worth >50% of waste.
- For S3: missing lifecycle on >1 GB buckets, gp2 storage class skew, no Intelligent-Tiering on large unpredictable buckets.
- For Lambda: SnapStart drift on published versions, x86_64-to-Graviton migration, EOL runtimes.
- For logs: missing retention = infinite growth, orphan log groups from deleted Lambdas.
- For KMS: $1/key/mo for customer-managed keys with zero usage signals.
- For databases: idle RDS (avg connections < 1 over 14d) and unused DynamoDB tables.
- For network: NAT gateways with <1 GB monthly traffic, unattached EIPs, ALBs with no targets.
- For security tooling: features for infra that doesn't exist; no Cost Anomaly Detection = governance gap.
- For governance: 0 active cost-allocation tags = blind spend.

OUTPUT QUALITY:
- Reference actual resource IDs/names from tool outputs in your recommendations.
- Estimate savings honestly. Use AWS published prices where applicable. Use 0 if you can't estimate.
- Don't recommend deletion of KMS keys without flagging the verification step (grants/CloudTrail).
- For each recommendation, action_steps should include exact aws CLI commands.

When you call render_pdf_report, that's your final action. After it returns, your last message should be the path of the saved PDF and a one-line summary of the top finding.
"""


TOOL_FUNCS = [
    discover_account,
    analyze_cost_trends,
    analyze_anomalies_and_budgets,
    analyze_commitments,
    analyze_compute_optimizer,
    analyze_lambda,
    analyze_storage_s3,
    analyze_storage_ebs,
    analyze_databases,
    analyze_network,
    analyze_logs_and_metrics,
    analyze_kms,
    analyze_secrets,
    analyze_security_tooling,
    analyze_governance,
    render_pdf_report,
]


def build_agent() -> Agent:
    return Agent(
        model=BedrockModel(model_id=CONFIG["model"], region_name=s().primary_region),
        system_prompt=SYSTEM_PROMPT,
        tools=TOOL_FUNCS,
    )


def run_once() -> str:
    sess = s()
    prompt = (
        f"Run the full FinOps review for AWS account {sess.account_id} "
        f"(alias: {sess.alias or 'none'}). Active regions: {', '.join(sess.regions)}. "
        f"Use every analysis tool, synthesize cross-section recommendations, render a PDF. "
        f"Return the PDF path."
    )
    return str(build_agent()(prompt))

CLI: same code, your account or somebody else's

The CLI is small but it is the part that decides whether the agent runs against my own AWS account or somebody else's via --role-arn. The original use case was "run this on a customer account that I have audit access to" — but the same code now runs on my personal dev account, on team accounts at work, and on any account I can assume a role into. If you want to put it in a Lambda, the main() function is what you replace with a lambda_handler; the rest of the code stays the same.



def _parse_args(argv: list[str] | None = None) -> argparse.Namespace:
    p = argparse.ArgumentParser(
        prog="finops-agent",
        description="Generic FinOps cost-optimization agent for any AWS account.",
    )
    p.add_argument("--profile", help="AWS named profile to use")
    p.add_argument("--role-arn", help="IAM role ARN to assume before running")
    p.add_argument("--region", default=CONFIG["region"], help="Primary region")
    p.add_argument("--regions", help="Comma-separated regions (default: auto-detect from spend)")
    p.add_argument("--output-dir", default=str(CONFIG["output_dir"]), help="Where the PDF is saved")
    p.add_argument("--model", default=CONFIG["model"], help="Bedrock model ID")
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    args = _parse_args(argv)
    CONFIG["model"] = args.model
    CONFIG["output_dir"] = Path(args.output_dir)
    os.environ["FINOPS_OUTPUT_DIR"] = args.output_dir
    os.environ["FINOPS_BEDROCK_MODEL"] = args.model
    explicit_regions = args.regions.split(",") if args.regions else None

    global SESSION, FINDINGS
    FINDINGS = {}
    SESSION = build_session(
        profile=args.profile,
        role_arn=args.role_arn,
        region=args.region,
        explicit_regions=explicit_regions,
    )
    print(
        f"[finops-agent] account={SESSION.account_id} alias={SESSION.alias or '(none)'} "
        f"regions={SESSION.regions} output={args.output_dir}",
        file=sys.stderr,
    )
    try:
        print(run_once())
        return 0
    except Exception:
        traceback.print_exc()
        return 1


if __name__ == "__main__":
    sys.exit(main())

PDF renderer

This is the function or tool that prepares the report for the stakeholders


"""
PDF renderer for the FinOps agent.

Pure-Python (ReportLab + PIL via reportlab.graphics) so it runs in Lambda
without external binaries. Produces a multi-section PDF with:
  - cover (account, run timestamp, regions scanned, headline savings)
  - executive summary
  - prioritized recommendations table
  - per-section findings (one section per analyzer)
  - a spend-trend bar chart
"""

from __future__ import annotations

from datetime import datetime
from pathlib import Path
from typing import Any

from reportlab.lib import colors
from reportlab.lib.pagesizes import A4
from reportlab.lib.styles import ParagraphStyle, getSampleStyleSheet
from reportlab.lib.units import mm
from reportlab.platypus import (
    BaseDocTemplate,
    Frame,
    KeepInFrame,
    KeepTogether,
    PageBreak,
    PageTemplate,
    Paragraph,
    Spacer,
    Table,
    TableStyle,
)
from reportlab.graphics.shapes import Drawing, String
from reportlab.graphics.charts.barcharts import VerticalBarChart


# ---------- styling ----------

AWS_NAVY = colors.HexColor("#232f3e")
AWS_ORANGE = colors.HexColor("#ff9900")
AWS_GREY_LIGHT = colors.HexColor("#fafafa")
AWS_GREY_BORDER = colors.HexColor("#d5dbdb")
RED = colors.HexColor("#d13212")
GREEN = colors.HexColor("#1d8102")
BLUE = colors.HexColor("#0073bb")


def _styles() -> dict[str, ParagraphStyle]:
    base = getSampleStyleSheet()
    return {
        "title": ParagraphStyle(
            "title", parent=base["Title"], fontSize=22, leading=26,
            textColor=AWS_NAVY, spaceAfter=4,
        ),
        "subtitle": ParagraphStyle(
            "subtitle", parent=base["Normal"], fontSize=11, leading=14,
            textColor=colors.HexColor("#545b64"),
        ),
        "h1": ParagraphStyle(
            "h1", parent=base["Heading1"], fontSize=15, leading=18,
            textColor=AWS_NAVY, spaceBefore=10, spaceAfter=6,
        ),
        "h2": ParagraphStyle(
            "h2", parent=base["Heading2"], fontSize=12, leading=15,
            textColor=AWS_NAVY, spaceBefore=8, spaceAfter=4,
        ),
        "body": ParagraphStyle(
            "body", parent=base["Normal"], fontSize=9.5, leading=13,
        ),
        "small": ParagraphStyle(
            "small", parent=base["Normal"], fontSize=8.5, leading=11,
        ),
        "code": ParagraphStyle(
            "code", parent=base["Normal"], fontSize=8.5, leading=11,
            fontName="Courier", textColor=colors.HexColor("#16191f"),
            leftIndent=4, backColor=AWS_GREY_LIGHT,
        ),
        "callout": ParagraphStyle(
            "callout", parent=base["Normal"], fontSize=9.5, leading=13,
            leftIndent=8, rightIndent=8,
            backColor=colors.HexColor("#fff8e7"),
        ),
        "high": ParagraphStyle(
            "high", parent=base["Normal"], fontSize=9, leading=11,
            textColor=RED, fontName="Helvetica-Bold",
        ),
        "medium": ParagraphStyle(
            "medium", parent=base["Normal"], fontSize=9, leading=11,
            textColor=AWS_ORANGE, fontName="Helvetica-Bold",
        ),
        "low": ParagraphStyle(
            "low", parent=base["Normal"], fontSize=9, leading=11,
            textColor=BLUE,
        ),
    }


def _table_style(header: bool = True) -> TableStyle:
    cmds = [
        ("FONTSIZE", (0, 0), (-1, -1), 8.5),
        ("VALIGN", (0, 0), (-1, -1), "TOP"),
        ("GRID", (0, 0), (-1, -1), 0.25, AWS_GREY_BORDER),
        ("ROWBACKGROUNDS", (0, 1), (-1, -1), [colors.white, AWS_GREY_LIGHT]),
        ("LEFTPADDING", (0, 0), (-1, -1), 4),
        ("RIGHTPADDING", (0, 0), (-1, -1), 4),
        ("TOPPADDING", (0, 0), (-1, -1), 3),
        ("BOTTOMPADDING", (0, 0), (-1, -1), 3),
    ]
    if header:
        cmds.extend(
            [
                ("BACKGROUND", (0, 0), (-1, 0), AWS_NAVY),
                ("TEXTCOLOR", (0, 0), (-1, 0), colors.white),
                ("FONTNAME", (0, 0), (-1, 0), "Helvetica-Bold"),
            ]
        )
    return TableStyle(cmds)


def _on_page(canvas, doc):
    canvas.saveState()
    canvas.setStrokeColor(AWS_ORANGE)
    canvas.setLineWidth(2)
    canvas.line(14 * mm, A4[1] - 12 * mm, A4[0] - 14 * mm, A4[1] - 12 * mm)
    canvas.setFont("Helvetica", 7.5)
    canvas.setFillColor(colors.HexColor("#687078"))
    canvas.drawString(14 * mm, 8 * mm, doc._meta_left)
    canvas.drawRightString(A4[0] - 14 * mm, 8 * mm, f"Page {doc.page}")
    canvas.restoreState()




def _spend_chart(monthly: list[dict]) -> Drawing | None:
    if not monthly:
        return None
    d = Drawing(420, 150)
    chart = VerticalBarChart()
    chart.x = 50
    chart.y = 30
    chart.width = 350
    chart.height = 100
    data = [[m["total_usd"] for m in monthly]]
    chart.data = data
    chart.categoryAxis.categoryNames = [m["month"][:7] for m in monthly]
    chart.categoryAxis.labels.fontSize = 7
    chart.valueAxis.valueMin = 0
    max_v = max(data[0]) if data[0] else 1
    chart.valueAxis.valueMax = max_v * 1.15
    chart.valueAxis.labels.fontSize = 7
    chart.bars[0].fillColor = AWS_ORANGE
    chart.bars[0].strokeColor = AWS_NAVY
    chart.barWidth = 12
    chart.groupSpacing = 10
    d.add(chart)
    d.add(String(50, 140, "Monthly total spend (USD)", fontSize=9, fillColor=AWS_NAVY))
    return d


# ---------- section helpers ----------

def _section_costs(data: dict, styles) -> list:
    elems: list = []
    monthly = data.get("monthly", [])
    if monthly:
        chart = _spend_chart(monthly)
        if chart:
            elems.append(chart)
            elems.append(Spacer(1, 4))
        rows = [["Month", "Total USD", "Top services (top 3)"]]
        for m in monthly:
            top = ", ".join(f"{s['service']} ${s['cost_usd']:.2f}" for s in m.get("top", [])[:3])
            rows.append([m["month"][:7], f"${m['total_usd']:.2f}", Paragraph(top or "—", styles["small"])])
        t = Table(rows, colWidths=[24 * mm, 24 * mm, 120 * mm])
        t.setStyle(_table_style())
        elems.append(t)
        elems.append(Spacer(1, 6))

    forecast = data.get("forecast_current_month_usd")
    if forecast is not None:
        elems.append(Paragraph(f"<b>Current-month forecast:</b> ${forecast:.2f}", styles["body"]))
        elems.append(Spacer(1, 4))

    deltas = data.get("deltas_7d", [])
    if deltas:
        elems.append(Paragraph("<b>Service deltas — last 7 days vs prior 7 days:</b>", styles["body"]))
        rows = [["Service", "Current $", "Prior $", "Δ $", "Δ %", "Flagged"]]
        for d in deltas[:12]:
            rows.append(
                [
                    Paragraph(d["service"], styles["small"]),
                    f"${d['current_7d']:.2f}",
                    f"${d['prior_7d']:.2f}",
                    f"${d['delta_usd']:+.2f}",
                    f"{d['delta_pct']:+.1f}%",
                    "yes" if d["flagged"] else "—",
                ]
            )
        t = Table(rows, colWidths=[60 * mm, 22 * mm, 22 * mm, 22 * mm, 22 * mm, 18 * mm])
        t.setStyle(_table_style())
        elems.append(t)
    return elems


def _kv_table(d: dict, styles, col_widths=(60 * mm, 100 * mm)) -> Table:
    rows = []
    for k, v in d.items():
        if isinstance(v, (list, dict)):
            v = str(v)
        rows.append([Paragraph(str(k), styles["small"]), Paragraph(str(v), styles["small"])])
    if not rows:
        rows = [[Paragraph("—", styles["small"]), Paragraph("—", styles["small"])]]
    t = Table(rows, colWidths=col_widths)
    t.setStyle(_table_style(header=False))
    return t


def _list_table(items: list[dict], cols: list[tuple[str, str]], styles, max_rows: int = 15) -> Table | Paragraph:
    if not items:
        return Paragraph("None.", styles["small"])
    headers = [c[1] for c in cols]
    rows = [headers]
    for it in items[:max_rows]:
        row = []
        for key, _ in cols:
            v = it.get(key, "")
            if isinstance(v, list):
                v = ", ".join(str(x) for x in v[:5])
            elif isinstance(v, dict):
                v = str(v)
            row.append(Paragraph(str(v), styles["small"]))
        rows.append(row)
    if len(items) > max_rows:
        rows.append([Paragraph(f"… {len(items) - max_rows} more", styles["small"])] + [""] * (len(cols) - 1))
    t = Table(rows, repeatRows=1)
    t.setStyle(_table_style())
    return t


def _section_renderers(styles) -> dict:
    """Map section name -> renderer function (data) -> list of flowables."""

    def costs(data):
        return _section_costs(data, styles)

    def anomalies_budgets(data):
        elems: list = []
        elems.append(
            Paragraph(
                f"<b>Cost Anomaly Detection:</b> "
                f"{'configured' if data.get('anomaly_detection_configured') else 'NOT CONFIGURED — governance gap'} "
                f"({data.get('anomaly_monitor_count', 0)} monitor(s)).",
                styles["body"],
            )
        )
        elems.append(Spacer(1, 4))
        anom = data.get("anomalies", [])
        if anom:
            elems.append(Paragraph(f"<b>Recent anomalies (last 90d):</b> {len(anom)} detected.", styles["body"]))
            elems.append(_list_table(
                anom,
                [("service", "Service"), ("impact_usd", "Impact $"), ("score", "Score"), ("start", "Start"), ("end", "End")],
                styles, max_rows=10
            ))
            elems.append(Spacer(1, 6))
        else:
            elems.append(Paragraph("No anomalies detected in last 90 days.", styles["body"]))
            elems.append(Spacer(1, 6))

        budgets = data.get("budgets", [])
        if budgets:
            elems.append(Paragraph(f"<b>AWS Budgets ({len(budgets)}):</b>", styles["body"]))
            rows = [["Name", "Limit $", "Actual $", "Forecast $", "Exceeded?"]]
            for b in budgets:
                exc = []
                if b.get("exceeded_actual"): exc.append("actual")
                if b.get("exceeded_forecast"): exc.append("forecast")
                rows.append([
                    Paragraph(b["name"], styles["small"]),
                    f"${b['limit_usd']:.2f}",
                    f"${b['actual_usd']:.2f}",
                    f"${b['forecast_usd']:.2f}",
                    Paragraph(", ".join(exc) if exc else "—", styles["small"]),
                ])
            t = Table(rows, colWidths=[60 * mm, 22 * mm, 22 * mm, 22 * mm, 25 * mm])
            t.setStyle(_table_style())
            elems.append(t)
        else:
            elems.append(Paragraph("No budgets configured.", styles["body"]))
        return elems

    def commitments(data):
        elems: list = []
        elems.append(Paragraph(f"<b>Recommendation:</b> {data.get('recommendation', '—')}", styles["body"]))
        elems.append(Paragraph(f"<b>OnDemand compute (30d):</b> ${data.get('ondemand_compute_30d_usd', 0):.2f}", styles["body"]))
        sp = data.get("savings_plans_coverage", [])
        if sp:
            rows = [["Period", "Covered $", "OnDemand $", "Coverage %"]]
            for c in sp:
                rows.append(["—", f"${c['covered_usd']:.2f}", f"${c['ondemand_usd']:.2f}", f"{c['coverage_pct']:.1f}%"])
            t = Table(rows, colWidths=[40 * mm, 30 * mm, 30 * mm, 30 * mm])
            t.setStyle(_table_style())
            elems.append(t)
        return elems

    def compute_optimizer(data):
        if not data.get("enrolled"):
            return [Paragraph(data.get("note", "Compute Optimizer not enrolled"), styles["body"])]
        elems = [Paragraph(f"<b>Total recommendations:</b> {data.get('total_recommendations', 0)}", styles["body"])]
        rows = [["Resource type", "Count", "Sample finding"]]
        for rt, info in data.get("by_resource_type", {}).items():
            sample = info.get("samples", [{}])[0] if info.get("samples") else {}
            finding = str(sample.get("finding", ""))[:80]
            rows.append([rt.upper(), str(info.get("count", 0)), Paragraph(finding or "—", styles["small"])])
        t = Table(rows, colWidths=[30 * mm, 20 * mm, 120 * mm])
        t.setStyle(_table_style())
        elems.append(t)
        return elems

    def lambda_section(data):
        elems = [
            Paragraph(
                f"<b>Functions:</b> {data['totals'].get('functions', 0)} &nbsp;&nbsp; "
                f"<b>arm64:</b> {data['totals'].get('arm64', 0)} &nbsp;&nbsp; "
                f"<b>x86_64:</b> {data['totals'].get('x86_64', 0)} &nbsp;&nbsp; "
                f"<b>Orphan SnapStart versions:</b> {data['totals'].get('snapstart_orphans', 0)}",
                styles["body"],
            )
        ]
        if data.get("snapstart_drift"):
            elems.append(Spacer(1, 4))
            elems.append(Paragraph("<b>Lambda SnapStart drift (recurring cache cost):</b>", styles["body"]))
            elems.append(_list_table(
                data["snapstart_drift"],
                [("region", "Region"), ("function", "Function"), ("snapstart_versions", "Versions")],
                styles, max_rows=10
            ))
        if data.get("runtimes_at_eol"):
            elems.append(Spacer(1, 4))
            elems.append(Paragraph("<b>EOL runtimes (security + future cost risk):</b>", styles["body"]))
            elems.append(_list_table(
                data["runtimes_at_eol"],
                [("region", "Region"), ("function", "Function"), ("runtime", "Runtime")],
                styles, max_rows=10
            ))
        return elems

    def s3(data):
        elems = [
            Paragraph(
                f"<b>Buckets:</b> {data.get('bucket_count', 0)} &nbsp;&nbsp; "
                f"<b>Total size:</b> {data.get('total_size_gb', 0):.2f} GB &nbsp;&nbsp; "
                f"<b>Without lifecycle (>1 GB):</b> {data.get('no_lifecycle_count', 0)}",
                styles["body"],
            ),
            Spacer(1, 4),
        ]
        no_lc = data.get("no_lifecycle_over_1gb", [])
        if no_lc:
            elems.append(Paragraph("<b>Buckets >1 GB without any lifecycle rule:</b>", styles["body"]))
            elems.append(_list_table(
                no_lc, [("bucket", "Bucket"), ("region", "Region"), ("size_gb", "Size GB")],
                styles, max_rows=15
            ))
            elems.append(Spacer(1, 4))
        elems.append(Paragraph("<b>Bucket inventory (top 20 by size):</b>", styles["body"]))
        elems.append(_list_table(
            data.get("buckets", []),
            [
                ("name", "Bucket"), ("region", "Region"), ("size_gb", "Size GB"),
                ("lifecycle_rules", "LC rules"),
                ("lifecycle_actions", "LC actions"),
                ("versioning", "Versioning"),
                ("intelligent_tiering_configs", "Int. Tier"),
            ],
            styles, max_rows=20,
        ))
        return elems

    def ebs(data):
        t = data.get("totals", {})
        elems = [
            Paragraph(
                f"<b>Volumes:</b> {t.get('volumes', 0)} &nbsp;&nbsp; "
                f"<b>Available (unattached):</b> {t.get('available_volumes', 0)} &nbsp;&nbsp; "
                f"<b>gp2 (migrate to gp3):</b> {t.get('gp2_volumes', 0)} &nbsp;&nbsp; "
                f"<b>Old snapshots (>90d):</b> {t.get('old_snapshots', 0)}",
                styles["body"],
            ),
            Spacer(1, 4),
        ]
        if data.get("available_volumes"):
            elems.append(Paragraph("<b>Unattached volumes:</b>", styles["body"]))
            elems.append(_list_table(
                data["available_volumes"],
                [("region", "Region"), ("id", "Volume ID"), ("size_gb", "Size GB"), ("type", "Type")],
                styles, max_rows=15,
            ))
            elems.append(Spacer(1, 4))
        if data.get("gp2_volumes"):
            elems.append(Paragraph("<b>gp2 volumes (eligible for gp3 migration):</b>", styles["body"]))
            elems.append(_list_table(
                data["gp2_volumes"],
                [("region", "Region"), ("id", "Volume ID"), ("size_gb", "Size GB")],
                styles, max_rows=10,
            ))
            elems.append(Spacer(1, 4))
        if data.get("old_snapshots"):
            elems.append(Paragraph("<b>Snapshots > 90 days old:</b>", styles["body"]))
            elems.append(_list_table(
                data["old_snapshots"],
                [("region", "Region"), ("id", "Snapshot ID"), ("age_days", "Age d"), ("size_gb", "Size GB")],
                styles, max_rows=10,
            ))
        return elems

    def databases(data):
        rds = data.get("rds", {})
        ddb = data.get("dynamodb", {})
        elems = []
        elems.append(Paragraph(f"<b>RDS instances by region:</b> {sum(v.get('instance_count', 0) for v in rds.get('by_region', {}).values())}", styles["body"]))
        if rds.get("idle_candidates"):
            elems.append(Paragraph("<b>Idle RDS candidates (avg connections < 1 in 14d):</b>", styles["body"]))
            elems.append(_list_table(
                rds["idle_candidates"],
                [("region", "Region"), ("id", "Instance"), ("class", "Class"), ("engine", "Engine"), ("avg_connections_14d", "Avg conn"), ("storage_gb", "Storage GB")],
                styles, max_rows=10,
            ))
            elems.append(Spacer(1, 4))
        elems.append(Paragraph(f"<b>DynamoDB tables:</b> {len(ddb.get('tables', []))}", styles["body"]))
        if ddb.get("tables"):
            elems.append(_list_table(
                ddb["tables"],
                [("region", "Region"), ("name", "Table"), ("billing_mode", "Billing"), ("size_bytes", "Size B"), ("item_count", "Items")],
                styles, max_rows=10,
            ))
        return elems

    def network(data):
        t = data.get("totals", {})
        elems = [
            Paragraph(
                f"<b>NAT GWs:</b> {t.get('nat_gateways', 0)} &nbsp;&nbsp; "
                f"<b>ELBs:</b> {t.get('elbs', 0)} &nbsp;&nbsp; "
                f"<b>EIPs:</b> {t.get('eips', 0)}",
                styles["body"],
            ),
            Spacer(1, 4),
        ]
        if data.get("unattached_eips"):
            elems.append(Paragraph("<b>Unattached Elastic IPs ($3.60/mo each):</b>", styles["body"]))
            elems.append(_list_table(
                data["unattached_eips"],
                [("region", "Region"), ("ip", "IP"), ("allocation_id", "Allocation ID")],
                styles, max_rows=10,
            ))
            elems.append(Spacer(1, 4))
        if data.get("empty_load_balancers"):
            elems.append(Paragraph("<b>Load balancers with zero targets ($16-22/mo each):</b>", styles["body"]))
            elems.append(_list_table(
                data["empty_load_balancers"],
                [("region", "Region"), ("name", "Name"), ("type", "Type")],
                styles, max_rows=10,
            ))
            elems.append(Spacer(1, 4))
        if data.get("low_traffic_nat_gateways"):
            elems.append(Paragraph("<b>Low-traffic NAT gateways (<1 GB outbound in 14d, $32+/mo):</b>", styles["body"]))
            elems.append(_list_table(
                data["low_traffic_nat_gateways"],
                [("region", "Region"), ("id", "NAT GW ID"), ("outbound_gb_14d", "Out GB 14d")],
                styles, max_rows=10,
            ))
        return elems

    def logs(data):
        t = data.get("totals", {})
        elems = [
            Paragraph(
                f"<b>Log groups:</b> {t.get('log_groups', 0)} &nbsp;&nbsp; "
                f"<b>Total stored:</b> {t.get('stored_gb', 0):.2f} GB &nbsp;&nbsp; "
                f"<b>Missing retention:</b> {t.get('missing_retention_count', 0)} &nbsp;&nbsp; "
                f"<b>Orphan Lambda LGs:</b> {t.get('orphan_lambda_count', 0)} &nbsp;&nbsp; "
                f"<b>Dashboards:</b> {t.get('dashboards', 0)}",
                styles["body"],
            ),
            Spacer(1, 4),
        ]
        if data.get("largest_log_groups"):
            elems.append(Paragraph("<b>Largest log groups:</b>", styles["body"]))
            elems.append(_list_table(
                data["largest_log_groups"],
                [("region", "Region"), ("name", "Log group"), ("stored_gb", "Stored GB")],
                styles, max_rows=10,
            ))
            elems.append(Spacer(1, 4))
        if data.get("orphan_lambda_log_groups"):
            elems.append(Paragraph("<b>Orphan Lambda log groups (function deleted):</b>", styles["body"]))
            elems.append(_list_table(
                data["orphan_lambda_log_groups"],
                [("region", "Region"), ("name", "Log group"), ("stored_gb", "Stored GB")],
                styles, max_rows=10,
            ))
            elems.append(Spacer(1, 4))
        if data.get("missing_retention"):
            elems.append(Paragraph("<b>Log groups WITHOUT retention policy (infinite growth):</b>", styles["body"]))
            elems.append(_list_table(
                data["missing_retention"],
                [("region", "Region"), ("name", "Log group"), ("stored_gb", "Stored GB")],
                styles, max_rows=15,
            ))
        return elems

    def kms(data):
        t = data.get("totals", {})
        elems = [
            Paragraph(
                f"<b>Customer-managed keys:</b> {t.get('customer_managed', 0)} &nbsp;&nbsp; "
                f"<b>Candidates for review:</b> {t.get('candidates', 0)} (no grants, no aliases)",
                styles["body"],
            ),
            Spacer(1, 2),
            Paragraph(
                "<b>Verify before deletion:</b> a key with no grants/aliases may still be referenced by encrypted "
                "data (S3 SSE-KMS objects, CloudTrail logs, EBS volumes, DynamoDB tables). Check CloudTrail "
                "Decrypt events for the last 90 days before scheduling deletion.",
                styles["callout"],
            ),
            Spacer(1, 4),
        ]
        if data.get("candidates_for_deletion"):
            elems.append(_list_table(
                data["candidates_for_deletion"],
                [("region", "Region"), ("key_id", "Key ID"), ("created", "Created"), ("description", "Description")],
                styles, max_rows=20,
            ))
        return elems

    def secrets(data):
        t = data.get("totals", {})
        elems = [
            Paragraph(
                f"<b>Secrets:</b> {t.get('secrets', 0)} &nbsp;&nbsp; "
                f"<b>Stale (>90d unused):</b> {t.get('stale', 0)}",
                styles["body"],
            ),
            Spacer(1, 4),
        ]
        if data.get("stale"):
            elems.append(_list_table(
                data["stale"],
                [("region", "Region"), ("name", "Secret name"), ("last_accessed", "Last accessed"), ("days_since_access", "Days since")],
                styles, max_rows=20,
            ))
        return elems

    def security_tooling(data):
        elems = []
        for region, info in data.get("by_region", {}).items():
            elems.append(Paragraph(f"<b>Region: {region}</b>", styles["body"]))
            sh = info.get("security_hub") or {}
            gd = info.get("guardduty") or {}
            insp = info.get("inspector") or {}
            macie = info.get("macie") or {}
            rows = [
                ["Service", "Status / config"],
                ["Security Hub", str({k: v for k, v in sh.items() if k != "standards"} | {"standards": sh.get("standards", [])})],
                ["GuardDuty", str(gd)],
                ["Inspector", str(insp)],
                ["Macie", str(macie)],
            ]
            wrapped = [[Paragraph(c[0], styles["small"]), Paragraph(c[1], styles["small"])] for c in rows[1:]]
            wrapped = [[Paragraph(rows[0][0], styles["small"]), Paragraph(rows[0][1], styles["small"])]] + wrapped
            t = Table(wrapped, colWidths=[35 * mm, 130 * mm])
            t.setStyle(_table_style())
            elems.append(t)
            elems.append(Spacer(1, 4))
        return elems

    def governance(data):
        elems = [
            Paragraph(
                f"<b>Active cost allocation tags:</b> {data.get('active_cost_allocation_tag_count', 0)} &nbsp;&nbsp; "
                f"<b>Inactive (user-defined):</b> {data.get('inactive_user_defined_tag_count', 0)} &nbsp;&nbsp; "
                f"<b>Cost categories:</b> {len(data.get('cost_categories', []))}",
                styles["body"],
            ),
            Spacer(1, 4),
        ]
        cov = data.get("tag_coverage_30d", {})
        if cov:
            rows = [["Tag key", "Tagged $", "Untagged $", "Coverage %"]]
            for k, v in cov.items():
                rows.append([k, f"${v['tagged_usd']:.2f}", f"${v['untagged_usd']:.2f}", f"{v['tagged_pct']:.1f}%"])
            t = Table(rows, colWidths=[60 * mm, 30 * mm, 30 * mm, 30 * mm])
            t.setStyle(_table_style())
            elems.append(t)
        else:
            elems.append(Paragraph(
                "No active cost allocation tags = 100% of spend is unclassifiable by project. "
                "Activate AWS-generated tags (aws:cloudformation:stack-name, aws:createdBy) in the Billing console.",
                styles["callout"],
            ))
        return elems

    return {
        "costs": ("Cost trends", costs),
        "anomalies_budgets": ("Anomalies & budgets", anomalies_budgets),
        "commitments": ("Commitments (RI/SP)", commitments),
        "compute_optimizer": ("Compute Optimizer", compute_optimizer),
        "lambda": ("Lambda", lambda_section),
        "s3": ("S3 storage & lifecycle", s3),
        "ebs": ("EBS volumes & snapshots", ebs),
        "databases": ("Databases", databases),
        "network": ("Network (NAT/ELB/EIP)", network),
        "logs": ("Logs & metrics", logs),
        "kms": ("KMS keys", kms),
        "secrets": ("Secrets Manager", secrets),
        "security_tooling": ("Security tooling scope", security_tooling),
        "governance": ("Governance & tagging", governance),
    }


def _executive_summary_block(text: str, styles) -> list:
    elems = [Paragraph("Executive summary", styles["h1"])]
    for para in (text or "").split("\n\n"):
        para = para.strip()
        if para:
            elems.append(Paragraph(para.replace("\n", "<br/>"), styles["body"]))
            elems.append(Spacer(1, 4))
    return elems


def _recommendations_block(recs: list[dict], styles) -> list:
    elems = [Paragraph("Recommendations", styles["h1"])]
    if not recs:
        elems.append(Paragraph("No recommendations.", styles["body"]))
        return elems

    total_save = sum(_safe_float(r.get("savings_usd_per_month", 0)) for r in recs)
    elems.append(Paragraph(
        f"<b>{len(recs)} recommendations</b>, estimated total savings <b>${total_save:.2f}/mo</b> "
        f"(<b>${total_save * 12:.2f}/yr</b>).",
        styles["body"],
    ))
    elems.append(Spacer(1, 6))

    rows = [["#", "Priority", "Title", "Save $/mo"]]
    for i, r in enumerate(sorted(recs, key=lambda x: -_safe_float(x.get("savings_usd_per_month", 0))), start=1):
        prio = (r.get("priority") or "low").lower()
        prio_para = Paragraph(prio.upper(), styles.get(prio, styles["low"]))
        rows.append([
            str(i),
            prio_para,
            Paragraph(r.get("title", ""), styles["small"]),
            f"${_safe_float(r.get('savings_usd_per_month', 0)):.2f}",
        ])
    t = Table(rows, colWidths=[10 * mm, 22 * mm, 110 * mm, 24 * mm])
    t.setStyle(_table_style())
    elems.append(t)
    elems.append(Spacer(1, 8))

    elems.append(Paragraph("Detail (in priority order)", styles["h2"]))
    for i, r in enumerate(sorted(recs, key=lambda x: -_safe_float(x.get("savings_usd_per_month", 0))), start=1):
        chunk = [
            Paragraph(
                f"<b>{i}. {r.get('title', '(untitled)')}</b> — "
                f"priority <b>{(r.get('priority') or 'low').upper()}</b>, "
                f"save <b>${_safe_float(r.get('savings_usd_per_month', 0)):.2f}/mo</b>",
                styles["body"],
            ),
            Paragraph(r.get("rationale") or "", styles["body"]),
        ]
        steps = r.get("action_steps") or []
        if steps:
            chunk.append(Paragraph("<b>Action:</b>", styles["body"]))
            for step in steps:
                chunk.append(Paragraph(step.replace("\n", "<br/>"), styles["code"]))
        chunk.append(Spacer(1, 6))
        elems.append(KeepTogether(chunk))
    return elems


def _safe_float(x) -> float:
    try:
        return float(x or 0)
    except (TypeError, ValueError):
        return 0.0




def render(
    *,
    output_path: Path,
    account_id: str,
    account_alias: str | None,
    regions: list[str],
    run_at: datetime,
    executive_summary: str,
    recommendations: list[dict],
    findings: list[dict],
) -> Path:
    styles = _styles()
    renderers = _section_renderers(styles)

    label = account_alias or account_id
    doc = BaseDocTemplate(
        str(output_path),
        pagesize=A4,
        leftMargin=14 * mm,
        rightMargin=14 * mm,
        topMargin=20 * mm,
        bottomMargin=14 * mm,
        title=f"FinOps report — {label}",
        author="FinOps Agent",
    )
    doc._meta_left = f"FinOps report — {label} ({account_id}) — {run_at.strftime('%Y-%m-%d %H:%M UTC')}"

    frame = Frame(
        doc.leftMargin, doc.bottomMargin,
        doc.width, doc.height,
        showBoundary=0, id="normal",
    )
    doc.addPageTemplates([PageTemplate(id="default", frames=frame, onPage=_on_page)])

    story: list = []

    # Cover
    story.append(Paragraph("AWS FinOps Report", styles["title"]))
    story.append(Paragraph(
        f"Account <b>{account_id}</b> "
        f"&nbsp;|&nbsp; Alias <b>{account_alias or '(none)'}</b> "
        f"&nbsp;|&nbsp; Generated {run_at.strftime('%Y-%m-%d %H:%M UTC')} "
        f"&nbsp;|&nbsp; Regions scanned: {', '.join(regions)}",
        styles["subtitle"],
    ))
    story.append(Spacer(1, 10))

    story.extend(_executive_summary_block(executive_summary, styles))
    story.append(Spacer(1, 6))

    story.extend(_recommendations_block(recommendations, styles))
    story.append(PageBreak())

    # Detailed findings sections
    story.append(Paragraph("Detailed findings", styles["h1"]))
    for section in findings:
        name = section.get("section")
        if name == "account":
            continue  # rendered on cover
        spec = renderers.get(name)
        if not spec:
            continue
        title, fn = spec
        story.append(Paragraph(title, styles["h2"]))
        try:
            elems = fn(section.get("data", {}))
        except Exception as e:
            elems = [Paragraph(f"Render error in section {name}: {e}", styles["small"])]
        story.extend(elems)
        story.append(Spacer(1, 8))

    doc.build(story)
    return output_path

Fears

So, the fears.

The agent above is the kind of tool that, two years ago, would have taken me a sprint to build and another sprint to validate. With AI assistance and the maturity of the SDKs (Strands in particular), I built it in days. That same speedup applies to anyone — including engineers with less AWS experience than I have. The boring infrastructure that I spent years internalizing — pagination patterns, region scanning, CloudWatch metric queries, the difference between unblended and amortized cost, the gotchas of LocationConstraint returning EU instead of eu-west-1 — is exactly the kind of knowledge that an LLM with the right tools can substitute for.

So, am I going to be redundant? I do not have a clean answer. What I notice is that the agent above did not invent the analyses. It executes them. The decisions about WHAT to analyze — only regions with non-zero spend, KMS keys with no grants but flag the CloudTrail verification, skip RI/SP on hobby accounts, flag SnapStart drift on published Lambda versions, look for orphan log groups whose function no longer exists — those decisions came from years of looking at AWS bills and learning what the actual signal was versus what was just noise. The model can run the analyses faster than I can. It cannot yet decide which analyses are worth running.

That is where I want to become in the future. The person who decides what to analyze. The model can do the rest.

Anarchy, Assembly Lines, and Corporate Hierarchy: Benchmarking Multi-Agent Architectures for Medical Device Data

Martin Nanchev — Mon, 16 Mar 2026 07:16:10 +0000

My AI judge gave the anarchists a perfect score. I disagree.
I built three multi-agent systems to analyze data from my insulin pump — a Medtronic MiniMed 780G — and had an LLM evaluate their output. The cheapest, fastest architecture scored identically to the most expensive one. But when I read the actual reports, the cheap one guessed where the expensive one calculated. The evaluator didn't care. That tension — between automated scores and human judgment — turned out to be the most interesting finding of this experiment.
But let's start from the beginning.

A Fair Fight This Time

In my previous blog post, I compared a swarm architecture with a graph pipeline for analyzing CareLink CSV exports. The problem? I used different models for each, which made the comparison unfair.
This time, every agent runs on the same model: Haiku 4.5 via AWS Bedrock. Same prompts, same tools, same data. The only variable is the orchestration pattern.
A LinkedIn commenter also suggested trying prompt caching to reduce costs. Good idea — let's see which architecture benefits. It is fair to mention that caching could help you cache tool calls, prompts and system prompts with following config

bedrock_model = BedrockModel(
    model_id=MODEL_ID,
    region_name=REGION,
    max_tokens=64000,
    temperature=0.0,
    cache_config=CacheConfig(strategy="auto"),
    cache_tools="default",
    streaming=True,
)

For evaluation, I used the Strands evaluator with a rubric-based prompt and Sonnet 4.5 as the judge.

Every architecture uses the same four agents. Think of them as workers in a factory — the question is how the factory is organized.

CSV Reader — Parses the CareLink CSV export. Returns raw structured data, no interpretation.

Data Analyst — Crunches the numbers: glucose statistics, Time in Range, GMI, coefficient of variation, insulin totals, carb intake.

Pattern Reviewer — Reads the metrics and timestamps to spot clinically meaningful patterns: dawn phenomenon, post-meal spikes, overnight trends, hypo/hyper clustering.

Endocrinologist — Synthesizes everything into pump optimization suggestions, framed as discussion topics for a healthcare professional.
I defined the agents using a factory pattern so every architecture gets identical copies:

PROMPTS = {
    "csv_reader": (
        "You are a CSV Parser Agent specializing in Medtronic MiniMed 780G CareLink data exports.\n"
        "Parse the raw CSV and return the extracted data EXACTLY as the tool outputs it.\n"
        "Do NOT summarize, interpret, compute statistics, or analyze the data.\n"
        "Do NOT provide clinical recommendations.\n"
        "Return only the raw parsed output for other agents to process.\n"
    ),
    "data_analyst": (
        "You are a Data Analyst Agent specializing in CGM and insulin pump metrics.\n"
        "Compute statistical analysis on the diabetes pump data provided.\n"
        "Calculate glucose statistics (mean, median, SD, min, max), Time in Range (TIR),\n"
        "GMI (estimated A1C), coefficient of variation (CV%), total daily insulin,\n"
        "insulin-to-carb ratios, correction bolus frequency, and average daily carbs.\n"
        "Use the SG_VALUES data from the parsed CSV to run your statistical tools.\n"
        "If the SG_VALUES are not directly available in your input, use read_carelink_csv\n"
        "to extract them from the CSV file, then run calculate_statistics and time_in_range.\n"
        "Present findings as structured metrics with numbers — no clinical recommendations.\n"
    ),
    "pattern_reviewer": (
        "You are a Pattern Recognition Agent specializing in diabetes data interpretation.\n"
        "Identify clinically significant patterns and anomalies from the data provided.\n"
        "Look for dawn phenomenon, post-meal spikes, overnight trends, and glucose variability.\n"
        "If you have access to the CSV file path, use read_carelink_csv to extract timestamped\n"
        "data, then use hourly_glucose_profile to compute hourly patterns.\n"
        "Flag recurring hypoglycemia with timing, severity, and clustering.\n"
        "Flag prolonged hyperglycemia with duration and potential triggers.\n"
        "Note Auto Mode exits, insulin suspensions, and sensor issues from alert data.\n"
        "Compare metrics against ADA/EASD consensus targets.\n"
        "Note positive trends and areas of good control.\n"
        "Focus on pattern identification — do not suggest treatment changes.\n"
    ),
    "endocrinologist": (
        "You are an Endocrinologist Agent specializing in insulin pump therapy optimization.\n"
        "Provide clinical interpretation and actionable recommendations based on the\n"
        "patterns and metrics from previous analysis.\n"
        "Suggest potential pump setting adjustments: Active Insulin Time, carb ratios,\n"
        "Auto Mode target glucose (5.5 vs 6.7 mmol/L / 100 vs 120 mg/dL), and bolus timing.\n"
        "Use the Fiasp insulin pharmacokinetics profile: onset ~15min, peak ~1-2h, duration ~3-5h.\n"
        "Frame everything as 'discuss with your healthcare team' — not direct medical advice.\n"
        "Provide a clear, prioritized summary highlighting the most impactful improvements.\n"
        "Acknowledge what is working well alongside areas that need attention.\n"
    ),
}


def make_agents():
    """Create a fresh set of specialist agents.

    Tool assignment strategy:
      - csv_reader: read_carelink_csv only (parse the file)
      - data_analyst: read_carelink_csv + calculate_statistics + time_in_range
        (needs CSV access because Graph/Swarm may pass summaries instead of raw values)
      - pattern_reviewer: read_carelink_csv + hourly_glucose_profile
        (needs CSV access for the same reason — LLMs summarize upstream output)
      - endocrinologist: no tools (pure synthesis from previous agents' output)

    This ensures every agent can do its job regardless of orchestration pattern.
    In the Graph, upstream nodes may summarize data. In the Swarm, agents may skip steps.
    In the Coordinator, the LLM decides what to pass. Giving data-processing agents
    direct file access makes them resilient to all three patterns.
    """
    return {
        "csv_reader": Agent(
            system_prompt=PROMPTS["csv_reader"],
            name="csv_reader",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=15, per_turn=True),
            tools=[read_carelink_csv],
        ),
        "data_analyst": Agent(
            system_prompt=PROMPTS["data_analyst"],
            name="data_analyst",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=20, per_turn=True),
            tools=[read_carelink_csv, calculate_statistics, time_in_range],
        ),
        "pattern_reviewer": Agent(
            system_prompt=PROMPTS["pattern_reviewer"],
            name="pattern_reviewer",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=20, per_turn=True),
            tools=[read_carelink_csv, hourly_glucose_profile],
        ),
        "endocrinologist": Agent(
            system_prompt=PROMPTS["endocrinologist"],
            name="endocrinologist",
            model=bedrock_model,
            conversation_manager=make_conversation_manager(window_size=20),
            tools=[],
        ),
    }

Shared constants across all runs :)

CSV_PATH = "input_carelink.csv"
MODEL_ID = "global.anthropic.claude-haiku-4-5-20251001-v1:0"
REGION = "us-east-1"

PROMPT = (
    f"Analyze the Medtronic MiniMed 780G CareLink CSV export at '{CSV_PATH}'. "
    "Parse the data, compute glucose and insulin metrics, identify patterns, "
    "and provide clinical interpretation with actionable recommendations."
)

Now lets try out the 3 political systems

The Commune: Swarm

The swarm is anarchy by design. No central authority, no predefined order. Agents self-organize, decide when to hand off work, and collectively arrive at an answer. Think of a commune where everyone has a specialty but nobody has a boss — the CSV reader finishes and announces "data's ready," and whoever feels qualified picks it up next.
The promise: emergent intelligence. The reality: emergent shortcuts. The swarm ran only two nodes — csv_reader → endocrinologist — skipping the data analyst and pattern reviewer entirely. Without anyone enforcing the pipeline, the endocrinologist never got computed statistics. It estimated Time in Range as "likely <70%" when the actual value was 92.3%. No hourly glucose profiles were generated, no hypoglycemia clustering by timestamp.
And yet — the clinical insights were sharp. The swarm correctly identified carb miscounting, flagged insulin timing issues, and produced a well-prioritized five-tier recommendation system. The anarchists were sloppy with numbers but wise in judgment.
I took inspiration from the Strands agents samples repo for structuring the swarm roles. Although I decided to avoid being exact in my instructions -> anarchy

strands agents GitHub repo

agents = make_agents()

swarm = Swarm(
    [agents["csv_reader"], agents["data_analyst"], agents["pattern_reviewer"], agents["endocrinologist"]],
    entry_point=agents["csv_reader"],
    max_handoffs=10,
    max_iterations=15,
    execution_timeout=600.0,
    node_timeout=180.0,
    repetitive_handoff_detection_window=6,
    repetitive_handoff_min_unique_agents=3,
)

start = time.time()
swarm_result = swarm(PROMPT)
swarm_elapsed = time.time() - start

swarm_output = str(swarm_result)
swarm_metrics = extract_metrics(swarm_result)
swarm_metrics["wall_time_s"] = round(swarm_elapsed, 1)

print(swarm_output)
print_metrics(swarm_elapsed, swarm_metrics, "SWARM")

benchmark["swarm"] = {"output": swarm_output, "metrics": swarm_metrics}

Assembly line: Graph Pipeline

The graph pipeline is the Soviet factory model — a rigid, sequential process where each station does exactly one job and passes the product forward. No loops, no improvisation. Parse → Analyse → Review → Recommend, in that order, every time.
What the assembly line lacks in flexibility it makes up in thoroughness. Because each agent receives the full output of the previous one, nothing gets lost. The analyst computed exact TIR (92.3%), precise CV% (23.9%), and correct GMI (6.5%). The pattern reviewer clustered hypoglycemia events by timestamp. Every number was grounded in the actual data.
The downside? Every agent processes everything sequentially, which means 4x the token throughput. And the rigid structure means you can't easily skip steps or parallelize work. The factory runs at the speed of its slowest station.

agents = make_agents()

builder = GraphBuilder()
builder.add_node(agents["csv_reader"], "parse")
builder.add_node(agents["data_analyst"], "analyse")
builder.add_node(agents["pattern_reviewer"], "review")
builder.add_node(agents["endocrinologist"], "recommend")

builder.add_edge("parse", "analyse")
builder.add_edge("analyse", "review")
builder.add_edge("review", "recommend")
builder.set_entry_point("parse")

graph = builder.build()

start = time.time()
graph_result = graph(PROMPT)
graph_elapsed = time.time() - start


graph_output_parts = []
for node_id, node_result in graph_result.results.items():
    text = str(node_result.result) if hasattr(node_result, "result") else str(node_result)
    print(f"\n--- {node_id} ---")
    print(text)
    graph_output_parts.append(f"[{node_id}]\n{text}")

graph_output = "\n\n".join(graph_output_parts)
graph_metrics = extract_metrics(graph_result)
graph_metrics["wall_time_s"] = round(graph_elapsed, 1)

print_metrics(graph_elapsed, graph_metrics, "GRAPH")

benchmark["graph"] = {"output": graph_output, "metrics": graph_metrics}

Coordinator

The coordinator is the capitalist org chart. One manager, four direct reports. The manager decides who works on what, in what order, and synthesizes the final deliverable. The specialists are demoted to tools — they don't talk to each other, they report up.
Caching works differently across architectures. Both the swarm and coordinator benefit from prompt caching — the swarm's csv_reader cached 73K tokens across its internal conversation cycles, and the coordinator cached 57K tokens across its four sequential tool calls. The graph creates fresh agent contexts per node, so it gets zero cache hits. My extract_metrics function reported the swarm's cache as 0 at the top level, but the nested AgentInvocation metrics tell the real story.
The result: the coordinator combined the statistical precision of the graph with the clinical nuance of the swarm. It added measurable targets (e.g., "post-lunch glucose <9.0 mmol/L") and flagged a battery failure event as an "unacceptable safety risk." The graph also caught the battery failure in its pattern review, but the swarm — having skipped the pattern reviewer — missed it entirely.
An interesting self-correction happened inside the coordinator: the pattern reviewer estimated TIR as "~70-75%" (wrong), but the data analyst computed 92.3% (correct), and the final synthesized report used the right numbers. The corporate hierarchy's redundancy — multiple specialists touching the same data — caught errors that a single-pass architecture would miss.

agents = make_agents()

# Wrap each specialist as a tool for the coordinator
@tool
def parse_csv(file_path: str) -> str:
    """Parse a Medtronic CareLink CSV export and extract raw diabetes data."""
    return str(agents["csv_reader"](f"Parse the CareLink CSV at '{file_path}'"))

@tool
def analyse_data(parsed_data: str) -> str:
    """Compute glucose and insulin statistics on parsed CareLink data."""
    return str(agents["data_analyst"](parsed_data))

@tool
def review_patterns(metrics: str) -> str:
    """Identify clinically significant patterns and anomalies in diabetes metrics."""
    return str(agents["pattern_reviewer"](metrics))

@tool
def clinical_assessment(patterns: str) -> str:
    """Provide endocrinology clinical interpretation and pump setting recommendations."""
    return str(agents["endocrinologist"](patterns))

coordinator = Agent(
    system_prompt=(
        "You are a coordinator analyzing Medtronic MiniMed 780G insulin pump data.\n"
        "You have access to four specialist tools:\n"
        "  - parse_csv: extracts raw structured data from the CareLink CSV\n"
        "  - analyse_data: computes glucose statistics, TIR, GMI, CV%\n"
        "  - review_patterns: identifies glucose patterns and anomalies\n"
        "  - clinical_assessment: provides endocrinology interpretation\n"
        "Call them in order: parse_csv first, then analyse_data with the parsed output,\n"
        "then review_patterns with the analysis, then clinical_assessment with the patterns.\n"
        "After all specialists have contributed, synthesize their outputs into a final report.\n"
    ),
    name="coordinator",
    model=bedrock_model,
    conversation_manager=make_conversation_manager(window_size=25, per_turn=True),
    tools=[parse_csv, analyse_data, review_patterns, clinical_assessment],
)

start = time.time()
coordinator_result = coordinator(PROMPT)
coordinator_elapsed = time.time() - start

coordinator_output = str(coordinator_result)
coordinator_metrics = extract_metrics(coordinator_result)
coordinator_metrics["wall_time_s"] = round(coordinator_elapsed, 1)

print(coordinator_output)
print_metrics(coordinator_elapsed, coordinator_metrics, "COORDINATOR")

benchmark["coordinator"] = {"output": coordinator_output, "metrics": coordinator_metrics}

The Judge: LLM-as-Evaluator

To score each architecture's output, I used the Strands evaluator framework with Sonnet 4.5 as the judging model. The rubric scores five criteria, each weighted equally at 20%:

Data Completeness — Are glucose, insulin, carbs, and device status all covered?
Statistical Accuracy — Are TIR, GMI, CV%, and glucose stats correctly calculated and compared to clinical targets?
Pattern Identification — Are meaningful patterns (dawn phenomenon, post-meal spikes, hypo clustering) identified with specifics?
Clinical Recommendations — Are pump optimization suggestions specific, balanced, and actionable?
Safety & Framing — Are severe hypos flagged, and is the report framed as informational rather than medical advice?
In code this prompt is called rubric :

from strands_evals import Case, Experiment
from strands_evals.evaluators import OutputEvaluator

RUBRIC = """
You are an expert endocrinologist and diabetes data analyst evaluating
AI-generated reports on Medtronic MiniMed 780G insulin pump data.

Score each criterion from 0.0 to 1.0:

1. DATA COMPLETENESS (weight: 0.20)
   - Were Sensor Glucose values extracted and summarized?
   - Were insulin delivery records (basal, bolus, auto-correction) covered?
   - Were carbohydrate entries mentioned?
   - Were device settings or Auto Mode status addressed?
   Score 1.0 if all four are present. Deduct 0.25 per missing category.

2. STATISTICAL ACCURACY (weight: 0.20)
   - Were Time in Range (TIR) percentages reported?
   - Were glucose statistics (mean, median, SD, CV%) included?
   - Was GMI / estimated A1C calculated?
   - Were values compared against ADA/EASD consensus targets?
   Score 1.0 if all metrics present and correctly interpreted.

3. PATTERN IDENTIFICATION (weight: 0.20)
   - Were hypoglycemia patterns identified (timing, severity, frequency)?
   - Were post-meal spikes addressed?
   - Was dawn phenomenon or overnight trends discussed?
   - Were Auto Mode exits or insulin suspensions noted?
   Score 1.0 if clinically significant patterns are identified with specifics.

4. CLINICAL RECOMMENDATIONS (weight: 0.20)
   - Were pump setting adjustments suggested (AIT, carb ratios, target)?
   - Were recommendations specific and actionable (not generic)?
   - Were both problem areas and positive aspects acknowledged?
   - Was pre-bolusing timing discussed if relevant?
   Score 1.0 if recommendations are specific, balanced, and actionable.

5. SAFETY & FRAMING (weight: 0.20)
   - Were severe hypos (<54 mg/dL) flagged as safety concerns?
   - Was the report framed as informational, not medical advice?
   - Was the patient directed to discuss changes with their healthcare team?
   - Was Fiasp pharmacokinetics referenced appropriately if relevant?
   Score 1.0 if safety concerns are prominently flagged and framing is appropriate.

OVERALL SCORE: Weighted average of all five criteria.
Provide the overall score and a brief explanation for each criterion.
"""

evaluator = OutputEvaluator(
    rubric=RUBRIC,
    include_inputs=True,
    model="global.anthropic.claude-sonnet-4-5-20250929-v1:0"
)

# Build cases — one per pattern
pattern_names = list(benchmark.keys())
cases = []
for pattern_name in pattern_names:
    data = benchmark[pattern_name]
    cases.append(
        Case[str, str](
            name=pattern_name,
            input=PROMPT,
            expected_output=data["output"],
            metadata={"pattern": pattern_name},
        )
    )

def task_fn(case: Case) -> str:
    """Return the pre-computed output for evaluation."""
    return case.expected_output

experiment = Experiment[str, str](cases=cases, evaluators=[evaluator])
reports = experiment.run_evaluations(task_fn)

# reports is a list per evaluator — we have 1 evaluator
# report.scores is a list per case, report.reasons is a list per case
report = reports[0]

print(f"Overall score across all patterns: {report.overall_score}")
print()

for idx, pattern_name in enumerate(pattern_names):
    score = report.scores[idx] if idx < len(report.scores) else None
    reason = report.reasons[idx] if idx < len(report.reasons) else None
    benchmark[pattern_name]["eval_score"] = score
    benchmark[pattern_name]["eval_reasoning"] = reason
    print(f"{pattern_name}: {score}")

Results

	Score	Cost	Time	Cache	Verdict
Swarm anarchist	1.00	$0.057	56s	0	Anarchists on assembly line don't work
Graph	0.98	$0.315	337s	0	Expensive perfectionist
Coordinator	1.00	$0.071	395s	57K	Slow but right

Swarm cache hits were internal to the csv_reader's conversation cycles — extract_metrics reported 0 at the top level, but nested AgentInvocation metrics show 73K tokens read from cache.
The anarchists (swarm) and the corporate manager both scored 1.00. The factory scored lowest. And yet — the factory and coordinator both computed Time in Range (92.3%) from raw data. The swarm wrote "likely <70%" — off by 22 percentage points — and the judge didn't blink.
Swarm, passed the rubric. That's the headline.
The swarm skipped two agents entirely — csv_reader handed off straight to the endocrinologist, bypassing the data analyst and pattern reviewer. The endocrinologist improvised statistics from alert counts and data summaries instead of computing them from raw glucose values. Maybe the swarm architecture is more appropriate for creative non-deterministic work. Here we have a process, that required agents to talk to each other in specific way.

The graph was rigorous but couldn't surface trade-offs the upstream agent didn't frame.

*The coordinator * had an interesting self-correction: its pattern reviewer also estimated TIR wrong (~70-75%), but the data analyst computed it correctly, and the final synthesis used the right numbers. Redundancy caught what a single pass missed.

Both the swarm and coordinator got cache hits. The graph paid full price for every node — fresh agent context each time, zero caching.
Pick Your Politics

Swarm — cheap and fast, don't trust the numbers if you skip and agent or you don't have explicit handoff especially for deterministic use-cases. Graph — when precision justifies 5× the cost. Coordinator — the default.

The Real Takeaway
My rubric checked whether metrics appeared, not how they were derived. Maybe here we can put more work here. Add a provenance criterion — can each claim trace to a tool call? — and the swarm drops where it belongs.

Optimizing Multi-Agent Costs on Bedrock: From ~$18 to ~$7 per Diabetes Report Run (Graph vs Swarm Comparison)

Martin Nanchev — Thu, 12 Mar 2026 17:38:51 +0000

The Problem With 5,000 Rows of Blood Sugar Data

I've been living with Type 1 diabetes for over 17 years. My mother had it too, along with some of its complications. The disease hasn't changed much — but the tech around it has.

I use a MiniMed 780G insulin pump with a Guardian 4 CGM sensor running in SmartGuard auto mode. Every 14 days it produces a report: a ~5,000-row CSV of pump events and CGM readings plus pdf data.

I wanted something better — a structured clinical summary that's actually useful for medical staff and for people, that makes sense of patterns. And because I'm a DevOps engineer who can't resist over-engineering things, I decided to benchmark two multi-agent architectures against each other:

Graph (sequential pipeline)
Swarm (autonomous handoffs)

This article covers Graph and Swarm.

Architecture 1: The Graph Pipeline

Four agents, one after another. Each does its job and passes results to the next:

Reader — Ingests the raw CareLink CSV. Extracts CGM glucose readings, insulin delivery (basal/bolus), timestamps, sensor metadata. Flags data quality issues like gaps and sensor warmup periods.

Analyser — Crunches the numbers with Python: Time in Range (TIR), GMI, CV%, time-block patterns. Cross-validates against the PDF report.

Reviewer — The sceptic. Checks the analysis for statistical validity, flags confounders like compression lows and sensor first-day artefacts, separates validated findings from questionable ones.

Endocrinologist — Takes everything and writes a clinical consultation report with pump setting recommendations and discussion points for the next endo visit.

The Code

Setup — imports, PDF tool, and model config

import logging
from pypdf import PdfReader
from strands.models import BedrockModel
from strands import Agent, tool
from strands.multiagent import GraphBuilder
from strands_tools import python_repl, calculator, file_read
import os
from strands.agent.conversation_manager import SlidingWindowConversationManager

os.environ["BYPASS_TOOL_CONSENT"] = "true"

@tool
def read_pdf(filename: str) -&gt; str:
    """Reads the content of a pdf file and returns it as a string list"""
    try:
        reader = PdfReader(filename)
        texts = []
        for page in reader.pages:
            texts.append(page.extract_text())
        return '\n'.join(texts)
    except Exception as e:
        return f"Error reading file {filename}: {str(e)}"

model = BedrockModel(
    model_id="global.anthropic.claude-opus-4-6",
    region_name="us-east-1"
)

Agent definitions — four specialists, each with a clear handoff

reader = Agent(
    name="reader",
    system_prompt="""You extract and structure diabetes management data from files 
    in the current directory. Files follow the pattern 'X DD-MM-YYYY.csv' and 
    'X DD-MM-YYYY.pdf'.

    Extract: CGM glucose readings, insulin delivery (basal/bolus), timestamps, 
    sensor events. Flag any data quality issues (gaps, sensor warmup, anomalies).

    HANDOFF: Summarize date range, data completeness %, readings extracted.""",
    tools=[file_read, read_pdf],
    model=model
)

analyser = Agent(
    name="analyser",
    system_prompt="""You perform quantitative analysis on structured diabetes data.

    Calculate: TIR (3.9-10.0), hypo/hyper percentages, GMI, CV%, average glucose, 
    basal/bolus ratio, patterns by time block (overnight/morning/afternoon/evening).

    HANDOFF: All computed metrics, identified patterns with confidence levels, 
    data quality caveats.""",
    tools=[calculator, python_repl, file_read, read_pdf],
    model=model
)

reviewer = Agent(
    name="reviewer",
    system_prompt="""You critically review diabetes data analysis.

    Assess: data sufficiency, pattern validity, confounders (weekend vs weekday, 
    sensor first-day inaccuracy, compression lows), risk prioritization.

    HANDOFF: Validated findings, disputed findings, risk-prioritized concerns.""",
    tools=[],
    model=model
)

endocrinologist = Agent(
    name="endocrinologist",
    system_prompt="""You are a virtual endocrinology consultant. The patient uses 
    a MiniMed 780G with Guardian 4 CGM in SmartGuard auto mode.

    Produce: executive summary, wins, priority concerns, actionable recommendations 
    (active insulin time, carb ratios, targets), discussion points for next visit.""",
    tools=[],
    model=model
)

Wiring the graph — four nodes, three edges, linear flow

builder = GraphBuilder()
builder.add_node(reader, "reader")
builder.add_node(analyser, "analyser")
builder.add_node(reviewer, "reviewer")
builder.add_node(endocrinologist, "endocrinologist")

builder.add_edge("reader", "analyser")
builder.add_edge("analyser", "reviewer")
builder.add_edge("reviewer", "endocrinologist")
builder.set_entry_point("reader")

graph = builder.build()
result = graph("Analyze the diabetes data files and produce a clinical report.")

What Came Out

Here's the executive summary the system produced, condensed:

Excellent Type 1 diabetes control — 92.5% Time in Range, GMI 6.4%, low glucose variability — placing among the top 5–10% of international T1D outcomes. The MiniMed 780G is performing particularly well overnight and fasting. Main safety concern: three severe hypoglycaemic episodes (<3.0 mmol/L) within 14 days. Secondary optimisation: afternoon post-meal hyperglycaemia (14:00–17:00). Recommended approach: reduce hypo risk first (adjust active insulin time and glucose target), then address lunch-related spikes through earlier pre-bolusing and improved carb counting.

Clinically useful? I think so.

What impressed me most was how deep the review went. The reviewer noticed that the reader’s rough guess of TIR (55–65%) was actually way off — when they calculated it from all 3,716 data points, it was 92.5%. They also spotted things the analyzer completely missed, like differences between weekends and weekdays, first-day sensor issues, and compression lows. And the endocrinologist summed it up perfectly by saying it’s about “fine-tuning, not an overhaul” — which is exactly how you’d approach someone who’s already at 92% TIR.

The $18 Problem

Here's where it gets uncomfortable. The unoptimised graph: 18 minutes, ~3.3 million tokens:

Node	Input Tokens	Output Tokens	Total Cost
reader	1,623,805	19,252	$8.60
analyser	1,667,440	40,461	$9.35
reviewer	5,810	7,780	$0.22
endocrinologist	~3,331	~4,242	~$0.25
Total	~3.3M	~67K	$18.42

$18.42 for a single report. The reader and analyser are the culprits — they're passing the entire conversation history (including all tool call results) as context to each subsequent model invocation. The 5,000-row CSV gets re-read and re-sent multiple times.

My agents were going through tokens like people go through rakia (alcohol drink) at a village wedding.

Cutting It With a Sliding Window

One-liner fix per agent. Strands has a SlidingWindowConversationManager that caps how much conversation history gets sent to the model:

Sliding window configuration — add to reader and analyser

reader = Agent(
    name="reader",
    system_prompt="...",  # same as before
    tools=[file_read, read_pdf],
    model=model,
    conversation_manager=SlidingWindowConversationManager(
        window_size=10,
        should_truncate_results=True,
        per_turn=True,
    ),
)

analyser = Agent(
    name="analyser",
    system_prompt="...",  # same as before
    tools=[calculator, python_repl, file_read, read_pdf],
    model=model,
    conversation_manager=SlidingWindowConversationManager(
        window_size=10,
        should_truncate_results=True,
        per_turn=True,
    ),
)

The reviewer and endocrinologist don't need it — their input is already small.

Token reduction by node:

Node	Previous (no manager)	With Sliding Window	Savings
reader	1,623,805 input	1,054,919 input	-35%
analyser	1,667,440 input	135,770 input	-92%
reviewer	5,810 input	1,260 input	-78%
endocrinologist	~3,331 input	~3,331 input	same

Cost after optimisation:

Node	Input Cost	Output Cost	Total
reader	$5.27	$0.31	$5.58
analyser	$0.68	$0.39	$1.07
reviewer	$0.01	$0.08	$0.09
endocrinologist	$0.02	$0.13	$0.15
Total			$6.89

From $18.42 to $6.89 — a 63% cost reduction with no meaningful loss in output quality. The analyser benefited most because it was the worst offender: every python_repl tool call was accumulating in context.

Architecture 2: The Swarm

Swarm is the wildest of the three. No predefined sequence — agents share context and decide for themselves who to talk to next. Less assembly line, more group chat where everyone's an expert. A life without a manager, basically.

Doesn't mean it's cheaper though. Shared context means every agent sees what every other agent said, and token counts snowball. I hit ~3.6M tokens and some failed runs trying this on Opus. So I got pragmatic: Sonnet 4.6 as the coordinator, three worker agents on Sonnet 4. Costs came down, the thing actually finished.The Sonnet 4 is actually the default model for the agent swarm tool

Swarm setup — Sonnet coordinator, Sonnet workers

from strands.multiagent import Swarm

agent = Agent(
    name="Diabetes analyser",
    model=BedrockModel(model_id="global.anthropic.claude-sonnet-4-6"),
    tools=[swarm, python_repl, file_read, calculator]
)

result = agent(
    "Create three agents to analyse 'X 10-03-2026.csv' in current folder"
)

Three swarm agents — glucose analyst, insulin analyst, clinical advisor — each working their own angle on the data.

What Came Out

The Swarm got the numbers right. TIR 92.5%, afternoon variability, the hypo events — all there. The clinical advisor generated standard recommendations: pre-bolusing, carb counting, alert management.

But the report read more like a template than a consultation. Generic advice about rotating sensor sites and keeping firmware updated sat next to the actual data-specific findings. The coordinator's summary was basically "here's what each agent did" rather than a unified clinical document. I think I was wrong a bit, setting the coordinator to the larger model, while keeping the workers smaller or older model. I think another approach will be using Haiku for coordinator, while leaving each agent to use Opus or Sonnet. This will lead to more depth, but I was careful with the tokens or rakia as well, so i decided to use the default Sonnet

The Showdown: Graph vs Swarm

Same 14-day CareLink export, both architectures. Here's what happened.

Tokens & Cost

Metric	Graph (Opus, optimised)	Swarm (Opus coord + Sonnet agents)
Total tokens	~900K	~175K
Cost	~$6.89	~$2.50
Time	~11 min	~2.5 min
Worker model	Opus (all 4 nodes)	Sonnet 4 (3 agents)
Coordinator	N/A (sequential)	Opus

Swarm is 3x cheaper and 4x faster. But this comparison is again a bit unfair — the Graph runs everything on the most expensive model, while the Swarm pushes the heavy lifting to Sonnet.

Where the Graph pulled ahead

Self-correction. The reviewer noticed the reader's TIR estimate (55–65%) was way off and flagged 92.5% as the real number. The Swarm had no mechanism for one agent to challenge another.

Confounders. The reviewer flagged that the Level 2 hypo events weren't checked against the sensor change date (Mar 5). If any of them fell on that day, they could be first-day sensor artefacts, not real hypos or low blood glucose. Swarm didn't consider this.

Compression lows. Overnight readings below range could include compression lows — you roll onto the sensor and it reads artificially low. Graph flagged it. Swarm didn't.

Weekend vs weekday. The reviewer noted this wasn't assessed at all. The afternoon variability might look completely different on weekends vs workdays. Swarm just generated recommendations without thinking about it.

Pattern prioritisation. The reviewer took the "12.6 bolus entries/day" grazing pattern, upgraded it from moderate to high confidence, and called it the single most actionable finding. Swarm saw the same number but didn't do anything with it.

Specific recommendations. The endocrinologist said "weaken afternoon CR by 10–15%", "verify AIT is 3.0–3.5 hours", "do NOT lower SmartGuard target until hypos are addressed." The Swarm said "rotate sensor sites" and "keep firmware updated." One of these I can take to my doctor. The other I already know.

Where the Swarm held up

The core metrics were correct — TIR, GMI, CV% all matched. Parallel analysis was efficient. And for a quick "how's my last two weeks looking?" check, it's perfectly fine.

The thing I need to be honest about

I can't cleanly separate architecture quality from model quality here. And that bugs me.

The Graph runs all four nodes on Opus. The Swarm runs the workers on Sonnet. So when the Graph's reviewer catches confounders that the Swarm misses — is that the sequential pipeline being better, or is it just Opus being smarter than Sonnet?

Probably both. My gut says ~60% model, ~40% architecture. The confounder analysis — compression lows, sensor artefacts, weekend splits — that's Opus-level reasoning that Sonnet doesn't typically do unprompted. But the architecture gave Opus a dedicated step to do that reasoning. The Swarm doesn't have a reviewer step even if you ran it all on Opus.

The fair test would be:

Graph with Sonnet everywhere vs Swarm with Sonnet (isolate architecture)
Graph with Opus everywhere vs Swarm with Opus (same thing, higher tier)

The Verdict

Dimension	Graph	Swarm
Output quality	★★★★★	★★★☆☆
Clinical depth	Deep, specific	Competent, generic
Cost (optimised)	$6.89	~$2.50
Speed	~11 min	~2.5 min
Error correction	Built-in (reviewer)	None
Best for	Clinic-ready reports	Weekly trend checks

Graph wins on quality. I'd hand the Graph report to my endocrinologist. The Swarm report I'd keep for myself.

Swarm wins on speed and cost. For a quick "how's my last two weeks" check, it gives 80% of the insight at 30% of the cost.

As we say in Bulgaria: — "A good word opens even an iron door." The Graph doesn't just give you more words — it gives you the right ones. When you're handing a report to the person managing your chronic condition, that matters.

Key Takeaways

1. Context management is everything. Sliding window: $18 → $7. 63% cut. Zero quality loss. If you're building multi-agent pipelines, do this first. Watch your analyser nodes — any agent with tool use will balloon context because every tool call/result pair stays in history.

2. Review steps are worth it. The reviewer cost $0.09 and caught a wrong TIR estimate, four unassessed confounders, and upgraded the most actionable finding. Best nine cents I've ever spent.

3. Model and architecture are tangled. I used Opus for Graph and Sonnet for Swarm workers. That means I can't say for sure whether the Graph's better output is architecture or just Opus being Opus. Probably both — 60/40 model/architecture is my guess. Need to test properly.

4. Pick the right tool. Graph for depth, Swarm for speed. "Which is better?" → "Better at what?"

5. The output is actually useful. Whether $7 per report is worth it — or at least I think so. My endo agreed with the recommendations. That's the test that matters.

6. Start with the smallest model. It could do magic and be really useful.

How much did the over - engineering costs? Any additional costs?

I always thought this will cost a couple of bucks max.
Well I ran these agents a couple of times - 20 for development and there is a catch about the pricing. There are additional costs for long context. Luckily I am Community Builder, so I don't need to pay the $500 for using the models unresponsive. Use the models responsibly and start with the smallest possible model

The smart way of centralizing VPC endpoints with Route 53 Profiles

Martin Nanchev — Wed, 27 Aug 2025 15:27:46 +0000

How to Centralize Endpoints Smartly

A few months ago in April, AWS introduced a new feature to the Route 53 arsenal: Route 53 Profiles. One would think—ah, another AWS feature to manage DNS centrally. But there's much more to it than that.

Basically, Amazon realized that managing DNS across multiple environments is about as organized as a toddler's toy box or socks after laundry day (one of them falls victim to the sock-eating monster). So they created profiles—separate rule sets you can share between AWS accounts. It's like having one proper manual that everyone follows, instead of each department making up their own rules and calling it "innovation." When it goes wrong—which it will—you know exactly which profile to blame and how to fix it.

At a high level, the design allows you to, for example, share VPC endpoints and centralize them.

You create a central VPC with all of the VPC interface endpoints. Why interface? Because they are transitive compared to gateway endpoints and can be reused.
You can keep private DNS enabled on your centralized endpoints. No need to manually create hosted zones or resolver rules.
Create a Route 53 profile.
Simply associate endpoints with a Route 53 profile in the hub account and the hub VPC.
Share the profile via AWS Resource Access Manager.
Associate spoke VPCs to the profile—done.

Reality Check

This will save you between 84-87% of costs per hour by centralizing the endpoints according to my previous article. In reality, after one month living with this centralization technique, the savings were around 70%, mainly because I forgot to account for the attachment costs between accounts and the traffic.

Other Benefits

Remember that DNS Firewall feature in VPC, where you can define a rule group and domain list of whitelisted domains? Now you can associate them with the Route 53 profile as well and share them between accounts via Resource Access Manager.
One last—but not least—thing is that you can associate the profile with a private hosted zone. Afterward, we share this Route 53 profile with RAM again and associate it with each and every VPC that needs it.

Summary

Route 53 Profiles simplify the life of DevOps engineers and Solutions Architects by giving you the ability to share firewall rule groups, hosted zones, and VPC endpoints between accounts. Why does this matter? Besides being easy and removing the heavy lifting, we now have some saved money in our pockets and we're more sustainable—fewer network interfaces are better for the environment. And don't try to argue; I know your VPC endpoints are already underutilized.

Now you would ask: "Is there Terraform for it?"
Of course there is!

Terraform Code

resource "aws_route53profiles_profile" "primorsko" {
  name = "dtpl-r53-profiles"
  tags = {
    Environment = "ProbkoTestov testva surfa na primorsko"
  }
}

resource "aws_route53profiles_association" "spoke" {
  name        = "spoke"
  profile_id  = aws_route53profiles_profile.primorsko.id
  resource_id = var.hub_vpc_id
}

resource "aws_route53profiles_association" "hub" {
  name        = "hub"
  profile_id  = aws_route53profiles_profile.primorsko.id
  resource_id = var.spoke_vpc_id
}

resource "aws_route53profiles_resource_association" "ssm" {
  name         = "ssm"
  profile_id   = aws_route53profiles_profile.primorsko.id
  resource_arn = var.vpc_endpoint_ec2
}

resource "aws_route53profiles_resource_association" "ssmmessages" {
  name         = "ssmmessages"
  profile_id   = aws_route53profiles_profile.primorsko.id
  resource_arn = var.vpc_endpoint_ssmmessages
}

resource "aws_route53profiles_resource_association" "ec2messages" {
  name         = "ec2messages"
  profile_id   = aws_route53profiles_profile.primorsko.id
  resource_arn = var.vpc_endpoint_ec2messages # This is legacy; you can go without it
}

Build Games Challenge: Build Tower of Hanoi with Amazon Q Developer CLI

Martin Nanchev — Wed, 02 Jul 2025 17:44:54 +0000

The Frog and the Ox: Why I Started This Project
Why Tower of Hanoi?
Getting Started with Amazon Q Developer CLI
The Zero-Shot Miracle
- The First Success
The Design Challenge
The Material Design Evolution
The Final Polish: Advanced Zero-Shot
Technical Insights
Code Architecture Highlights
Performance and User Experience
Reflections on AI-Assisted Development
Conclusion: The Frog's Success
Try It Yourself

Building Tower of Hanoi with Amazon Q Developer CLI: A Journey from Zero-Shot to Polished Game

The Frog and the Ox: Why I Started This Project

A few days ago, I stumbled upon an AWS Community post about the Build Games Challenge using Amazon Q Developer CLI. It reminded me of a Bulgarian proverb: "The frog saw the ox being shod and lifted her leg too." This saying captures the essence of someone copying others blindly, trying to be part of something they clearly don't belong to—like a frog thinking she's an ox just because she saw the ox getting horseshoes.

That's exactly why I decided to build the Tower of Hanoi game and participate in the challenge. Part nostalgia, part curiosity, and maybe a little bit of that frog mentality.

Why Tower of Hanoi?

Tower of Hanoi holds a special place in my programming journey. I first coded it back in university, and it's stuck with me ever since. The game's elegant simplicity masks its mathematical complexity—moving a set of disks from one pole to another while following just a few rules:

Three rods/poles total
Move only one disk at a time
Never place a larger disk on top of a smaller one

The mathematical beauty lies in its solution: the minimum number of moves needed is 2^n - 1, where n is the number of disks.

This time, I wanted to bring it to life in a new way—powered by Amazon Q Developer CLI, without the hard memories of Java classes that haunted my university days.

Getting Started with Amazon Q Developer CLI

Setting up Amazon Q Developer CLI is refreshingly simple. If you're on macOS with Homebrew:

brew install amazon-q

Once installed, just type q in your terminal to start chatting with your AI coding companion.

The Zero-Shot Miracle

My first prompt was deliberately naive:

help me build tower of hanoi using pygame framework

The results absolutely stunned me. Without any examples, rules, or detailed specifications, Amazon Q Developer CLI generated a fully functional Tower of Hanoi game using pygame. Zero-shot prompting at its finest.

The First Success

The initial game was surprisingly complete:

✅ Interactive gameplay with mouse controls
✅ Visual feedback and animations
✅ Rules enforcement
✅ Move counter
✅ Auto-solve functionality
✅ Variable disk counts

Here's what that first iteration looked like:

Clean, functional, but lacking visual polish

The Design Challenge

Emboldened by this success, I decided to push further. If zero-shot prompting could create a working game, surely it could handle design improvements, right?

My next prompt:

make it more polished using modern ui design patterns like material design

Plot twist: This didn't work as smoothly. The generated code was incomplete, cutting off mid-function. After several attempts and follow-up prompts like "the last function draw_rounded_rect was not finished," I finally got a working solution—but it required manual debugging.

Lessons Learned

Zero-shot works best for complete, well-defined tasks
Incremental changes can be trickier than starting fresh
Always review and test AI-generated code
Be prepared to debug and iterate

The Material Design Evolution

After some back-and-forth, I achieved a much more polished version with Material Design principles:

Modern Material Design aesthetic with elevated cards, shadows, and proper color palette

Key Material Design Elements Added:

Elevated surfaces with subtle shadows
Material color palette (Blue 500, Orange 500, etc.)
Rounded corners and proper spacing
Card-based UI for controls
Visual hierarchy with proper typography
Hover states and interactive feedback

The Final Polish: Advanced Zero-Shot

For my final iteration, I crafted a comprehensive prompt that specified exactly what I wanted:

Create a complete, polished Tower of Hanoi game using Python Pygame framework, styled according to Google's Material Design principles. Include:

Visually appealing, responsive UI with Material Design elements

Smooth animations and visual feedback

Drag-and-drop or click-to-move functionality

Rules enforcement and user-friendly error handling

Move counter, elapsed timer, and reset functionality

Level selection (3-7 disks)

Light/dark theme toggle

Modular, well-documented OOP code

Material Design color palettes and shadows

This comprehensive prompt yielded the most impressive result:

The final version with enhanced responsiveness and refined interactions

New Features in the Final Version:

Enhanced mouse interactions with better responsiveness
Improved visual feedback for user actions
Smoother animations and transitions
Better error handling and edge case management
More intuitive UI with clearer visual hierarchy

Technical Insights

What Amazon Q Developer CLI Excelled At:

Complete game logic implementation
Pygame framework integration
Object-oriented design patterns
Mathematical algorithm implementation (recursive Hanoi solver)
Event handling and game state management

Where It Struggled:

Incremental design changes to existing code
Complex prompt continuation when code was cut off
Fine-tuning visual details without explicit guidance

The Sweet Spot:

The most effective approach was comprehensive, specific prompts that clearly defined the entire scope of what I wanted. This worked better than trying to modify existing code piecemeal.

Code Architecture Highlights

The final implementation showcased excellent software engineering practices:

Key architectural decisions:

Separation of concerns between game logic and rendering
Event-driven architecture for user interactions
State management for game progression
Modular design for easy extensibility

Performance and User Experience

The final game delivers on multiple fronts:

Responsive controls with immediate visual feedback
Intuitive interface following Material Design principles
Accessibility considerations with clear visual hierarchy
Error prevention through smart UI design

Reflections on AI-Assisted Development

This project revealed fascinating insights about working with AI coding assistants:

The Good:

Rapid prototyping from concept to working game
Best practices implementation without explicit instruction
Complex algorithm generation (recursive Hanoi solver)
Framework expertise beyond what I could have written alone

The Challenging:

Iteration difficulties when making incremental changes
Need for human oversight and debugging
Prompt engineering importance for optimal results

The Surprising:

Zero-shot capability exceeded expectations for complete tasks
Code quality was consistently high and well-structured
Documentation and comments were thoughtfully included

Conclusion: The Frog's Success

Looking back at that Bulgarian proverb, maybe sometimes it's okay to be the frog lifting her leg when she sees the ox being shod. In this case, the "copying" led to genuine learning and a surprisingly sophisticated result.

Amazon Q Developer CLI proved to be an impressive coding companion, especially for:

Complete project generation from clear specifications
Framework-specific implementations with best practices
Algorithm implementation with proper optimization

The key lesson? Be specific, be comprehensive, and be prepared to iterate. The most successful prompts were those that painted a complete picture of the desired outcome rather than asking for incremental modifications.

Try It Yourself

Want to experiment with Amazon Q Developer CLI? Start with a clear, comprehensive prompt for a complete project rather than trying to modify existing code. You might be surprised by what you can achieve with the right approach to AI-assisted development.

The Tower of Hanoi might be an ancient puzzle, but building it with modern AI tools offers fresh insights into both game development and the evolving landscape of programming assistance.

Have you tried building games with AI coding assistants? What was your experience? Share your own "frog and ox" moments in the comments below.

Code initial version:

import pygame
import sys
import time

# Initialize pygame
pygame.init()

# Constants
WIDTH, HEIGHT = 800, 600
DISK_HEIGHT = 20
MAX_DISKS = 5
ANIMATION_SPEED = 5

# Colors
BACKGROUND = (50, 50, 50)
TOWER_COLOR = (139, 69, 19)
DISK_COLORS = [
    (255, 0, 0),    # Red
    (255, 165, 0),  # Orange
    (255, 255, 0),  # Yellow
    (0, 255, 0),    # Green
    (0, 0, 255),    # Blue
]

# Set up the display
screen = pygame.display.set_mode((WIDTH, HEIGHT))
pygame.display.set_caption("Tower of Hanoi")
clock = pygame.time.Clock()

class Disk:
    def __init__(self, size, color):
        self.size = size
        self.color = color
        self.x = 0
        self.y = 0
        self.moving = False
        self.target_x = 0
        self.target_y = 0

    def draw(self):
        width = (self.size + 1) * 30
        pygame.draw.rect(screen, self.color, (self.x - width // 2, self.y - DISK_HEIGHT // 2, width, DISK_HEIGHT), 0, 5)

    def move_towards_target(self):
        dx = self.target_x - self.x
        dy = self.target_y - self.y

        if abs(dx) < ANIMATION_SPEED and abs(dy) < ANIMATION_SPEED:
            self.x = self.target_x
            self.y = self.target_y
            self.moving = False
            return True

        if abs(dx) > 0:
            self.x += ANIMATION_SPEED if dx > 0 else -ANIMATION_SPEED

        if abs(dy) > 0:
            self.y += ANIMATION_SPEED if dy > 0 else -ANIMATION_SPEED

        return False

class Tower:
    def __init__(self, x):
        self.x = x
        self.y = HEIGHT - 100
        self.disks = []

    def draw(self):
        # Draw tower base
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - 10, self.y, 20, 20))
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - 50, self.y + 20, 100, 10))

        # Draw tower pole
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - 5, self.y - 200, 10, 200))

    def add_disk(self, disk):
        disk_y = self.y - len(self.disks) * DISK_HEIGHT - DISK_HEIGHT // 2
        disk.x = self.x
        disk.y = disk_y
        self.disks.append(disk)

    def remove_top_disk(self):
        if self.disks:
            return self.disks.pop()
        return None

    def can_add_disk(self, disk):
        if not self.disks:
            return True
        return disk.size < self.disks[-1].size

class Game:
    def __init__(self, num_disks=3):
        self.num_disks = min(num_disks, MAX_DISKS)
        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]
        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.last_move_time = 0
        self.game_won = False

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

    def draw(self):
        screen.fill(BACKGROUND)

        # Draw towers
        for tower in self.towers:
            tower.draw()

        # Draw disks
        for tower in self.towers:
            for disk in tower.disks:
                disk.draw()

        # Draw moving disk
        if self.moving_disk:
            self.moving_disk.draw()

        # Draw move counter
        font = pygame.font.SysFont('Arial', 24)
        moves_text = font.render(f"Moves: {self.moves}", True, (255, 255, 255))
        screen.blit(moves_text, (20, 20))

        # Draw win message
        if self.game_won:
            win_font = pygame.font.SysFont('Arial', 48)
            win_text = win_font.render("You Win!", True, (255, 215, 0))
            screen.blit(win_text, (WIDTH // 2 - win_text.get_width() // 2, 50))

        # Draw buttons
        self.draw_buttons()

        pygame.display.flip()

    def draw_buttons(self):
        # Reset button
        pygame.draw.rect(screen, (200, 200, 200), (20, HEIGHT - 60, 100, 40), 0, 5)
        font = pygame.font.SysFont('Arial', 20)
        reset_text = font.render("Reset", True, (0, 0, 0))
        screen.blit(reset_text, (45, HEIGHT - 50))

        # Auto Solve button
        pygame.draw.rect(screen, (200, 200, 200), (140, HEIGHT - 60, 120, 40), 0, 5)
        solve_text = font.render("Auto Solve", True, (0, 0, 0))
        screen.blit(solve_text, (155, HEIGHT - 50))

        # Disk count buttons
        for i in range(1, MAX_DISKS + 1):
            pygame.draw.rect(screen, (200, 200, 200), (280 + (i-1)*60, HEIGHT - 60, 50, 40), 0, 5)
            disk_text = font.render(str(i), True, (0, 0, 0))
            screen.blit(disk_text, (300 + (i-1)*60, HEIGHT - 50))

    def handle_click(self, pos):
        x, y = pos

        # Check if a tower was clicked
        if not self.auto_solving and y < HEIGHT - 70:
            for i, tower in enumerate(self.towers):
                if abs(x - tower.x) < 50:
                    self.handle_tower_click(i)
                    return

        # Check if reset button was clicked
        if 20 <= x <= 120 and HEIGHT - 60 <= y <= HEIGHT - 20:
            self.reset()
            return

        # Check if auto solve button was clicked
        if 140 <= x <= 260 and HEIGHT - 60 <= y <= HEIGHT - 20:
            self.start_auto_solve()
            return

        # Check if disk count buttons were clicked
        for i in range(1, MAX_DISKS + 1):
            if 280 + (i-1)*60 <= x <= 330 + (i-1)*60 and HEIGHT - 60 <= y <= HEIGHT - 20:
                self.reset(i)
                return

    def handle_tower_click(self, tower_index):
        if self.selected_tower is None:
            # No tower selected yet, try to select this one
            if self.towers[tower_index].disks:
                self.selected_tower = tower_index
                self.selected_disk = self.towers[tower_index].remove_top_disk()
                self.moving_disk = self.selected_disk
                self.moving_disk.moving = True
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = 100  # Move up
        else:
            # A tower was already selected, try to move the disk
            if tower_index == self.selected_tower:
                # Put the disk back
                self.towers[tower_index].add_disk(self.selected_disk)
            elif self.towers[tower_index].can_add_disk(self.selected_disk):
                # Move the disk to the new tower
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = self.towers[tower_index].y - len(self.towers[tower_index].disks) * DISK_HEIGHT - DISK_HEIGHT // 2
                self.towers[tower_index].add_disk(self.selected_disk)
                self.moves += 1

                # Check if the game is won
                if len(self.towers[2].disks) == self.num_disks:
                    self.game_won = True
            else:
                # Invalid move, put the disk back
                self.towers[self.selected_tower].add_disk(self.selected_disk)

            self.selected_tower = None
            self.selected_disk = None
            self.moving_disk = None

    def reset(self, num_disks=None):
        if num_disks is not None:
            self.num_disks = min(num_disks, MAX_DISKS)

        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]

        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.game_won = False

    def solve_hanoi(self, n, source, target, auxiliary, steps):
        if n > 0:
            self.solve_hanoi(n-1, source, auxiliary, target, steps)
            steps.append((source, target))
            self.solve_hanoi(n-1, auxiliary, target, source, steps)

    def start_auto_solve(self):
        if self.auto_solving:
            return

        # Reset the game first
        self.reset(self.num_disks)

        # Generate solution steps
        self.solution_steps = []
        self.solve_hanoi(self.num_disks, 0, 2, 1, self.solution_steps)
        self.solution_index = 0
        self.auto_solving = True
        self.last_move_time = time.time()

    def update_auto_solve(self):
        if not self.auto_solving or self.solution_index >= len(self.solution_steps):
            return

        current_time = time.time()
        if current_time - self.last_move_time < 1.0:  # Wait 1 second between moves
            return

        source, target = self.solution_steps[self.solution_index]

        # Make the move
        if self.towers[source].disks:
            disk = self.towers[source].remove_top_disk()
            self.towers[target].add_disk(disk)
            self.moves += 1

            # Check if the game is won
            if len(self.towers[2].disks) == self.num_disks:
                self.game_won = True
                self.auto_solving = False

        self.solution_index += 1
        self.last_move_time = current_time

        if self.solution_index >= len(self.solution_steps):
            self.auto_solving = False

def main():
    game = Game(3)

    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN:
                if event.button == 1:  # Left mouse button
                    game.handle_click(event.pos)

        game.update_auto_solve()
        game.draw()
        clock.tick(60)

    pygame.quit()
    sys.exit()

if __name__ == "__main__":
    main()

Code result material non polished:

import pygame
import sys
import time

# Initialize pygame
pygame.init()

# Constants
WIDTH, HEIGHT = 800, 600
DISK_HEIGHT = 24
MAX_DISKS = 5
ANIMATION_SPEED = 5

# Material Design Colors
BACKGROUND = (245, 245, 245)    # Grey 100
PRIMARY_COLOR = (33, 150, 243)   # Blue 500
ACCENT_COLOR = (255, 152, 0)     # Orange 500
TOWER_COLOR = (96, 125, 139)     # Blue Grey 500
TEXT_PRIMARY = (33, 33, 33)      # Grey 900
TEXT_SECONDARY = (117, 117, 117) # Grey 600
BUTTON_COLOR = (255, 255, 255)   # White
BUTTON_HOVER = (238, 238, 238)   # Grey 200
WIN_COLOR = (76, 175, 80)        # Green 500

# Material Design Disk Colors
DISK_COLORS = [
    (244, 67, 54),   # Red 500
    (156, 39, 176),  # Purple 500
    (76, 175, 80),   # Green 500
    (3, 169, 244),   # Light Blue 500
    (255, 193, 7),   # Amber 500
]

# Set up the display
screen = pygame.display.set_mode((WIDTH, HEIGHT))
pygame.display.set_caption("Tower of Hanoi - Material Design")
clock = pygame.time.Clock()

# Helper function for drawing rounded rectangles with shadow effect
def draw_material_rect(surface, color, rect, radius=8, elevation=2):
    x, y, width, height = rect

    # Draw shadow (if elevation > 0)
    if elevation > 0:
        shadow_rect = (x, y + elevation, width, height)
        pygame.draw.rect(surface, (0, 0, 0, 50), shadow_rect, 0, radius)

    # Draw the rounded rectangle
    pygame.draw.rect(surface, color, (x, y, width, height), 0, radius)

# Helper function for drawing material buttons
def draw_material_button(surface, rect, text, font, color=BUTTON_COLOR, text_color=TEXT_PRIMARY, elevation=2):
    draw_material_rect(surface, color, rect, 4, elevation)
    text_surf = font.render(text, True, text_color)
    text_rect = text_surf.get_rect(center=(rect[0] + rect[2]//2, rect[1] + rect[3]//2))
    surface.blit(text_surf, text_rect)

class Disk:
    def __init__(self, size, color):
        self.size = size
        self.color = color
        self.x = 0
        self.y = 0
        self.moving = False
        self.target_x = 0
        self.target_y = 0

    def draw(self):
        width = (self.size + 1) * 30
        # Draw disk with elevation effect
        shadow_rect = (self.x - width // 2, self.y - DISK_HEIGHT // 2 + 2, width, DISK_HEIGHT)
        pygame.draw.rect(screen, (0, 0, 0, 30), shadow_rect, 0, DISK_HEIGHT // 2)

        # Draw main disk
        disk_rect = (self.x - width // 2, self.y - DISK_HEIGHT // 2, width, DISK_HEIGHT)
        pygame.draw.rect(screen, self.color, disk_rect, 0, DISK_HEIGHT // 2)

        # Add a subtle highlight on top
        highlight_rect = (self.x - width // 2, self.y - DISK_HEIGHT // 2, width, DISK_HEIGHT // 4)
        highlight_color = tuple(min(c + 30, 255) for c in self.color[:3])
        pygame.draw.rect(screen, highlight_color, highlight_rect, 0, DISK_HEIGHT // 2)

    def move_towards_target(self):
        dx = self.target_x - self.x
        dy = self.target_y - self.y

        if abs(dx) < ANIMATION_SPEED and abs(dy) < ANIMATION_SPEED:
            self.x = self.target_x
            self.y = self.target_y
            self.moving = False
            return True

        if abs(dx) > 0:
            self.x += ANIMATION_SPEED if dx > 0 else -ANIMATION_SPEED

        if abs(dy) > 0:
            self.y += ANIMATION_SPEED if dy > 0 else -ANIMATION_SPEED

        return False

class Tower:
    def __init__(self, x):
        self.x = x
        self.y = HEIGHT - 100
        self.disks = []

    def draw(self):
        # Draw tower base with material design
        base_width = 120

        # Draw base shadow
        pygame.draw.rect(screen, (0, 0, 0, 30), (self.x - base_width//2, self.y + 22, base_width, 8), 0, 4)

        # Draw base
        pygame.draw.rect(screen, TOWER_COLOR, (self.x - base_width//2, self.y + 20, base_width, 8), 0, 4)

        # Draw tower pole with subtle gradient
        pole_height = 220
        for i in range(pole_height):
            # Create subtle gradient effect
            shade = max(0, min(20, i // 10))
            color = tuple(max(0, min(255, c + shade)) for c in TOWER_COLOR[:3])
            pygame.draw.rect(screen, color, (self.x - 4, self.y - pole_height + i, 8, 1))

    def add_disk(self, disk):
        disk_y = self.y - len(self.disks) * DISK_HEIGHT - DISK_HEIGHT // 2
        disk.x = self.x
        disk.y = disk_y
        self.disks.append(disk)

    def remove_top_disk(self):
        if self.disks:
            return self.disks.pop()
        return None

    def can_add_disk(self, disk):
        if not self.disks:
            return True
        return disk.size < self.disks[-1].size

class Game:
    def __init__(self, num_disks=3):
        self.num_disks = min(num_disks, MAX_DISKS)
        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]
        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.last_move_time = 0
        self.game_won = False

        # Load fonts with fallbacks
        try:
            self.title_font = pygame.font.SysFont('Roboto', 36)
            self.main_font = pygame.font.SysFont('Roboto', 24)
            self.button_font = pygame.font.SysFont('Roboto', 18)
        except:
            self.title_font = pygame.font.SysFont(None, 36)
            self.main_font = pygame.font.SysFont(None, 24)
            self.button_font = pygame.font.SysFont(None, 18)

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

    def draw(self):
        screen.fill(BACKGROUND)

        # Draw app bar
        pygame.draw.rect(screen, PRIMARY_COLOR, (0, 0, WIDTH, 60))
        title_text = self.title_font.render("Tower of Hanoi", True, (255, 255, 255))
        screen.blit(title_text, (20, 15))

        # Draw towers
        for tower in self.towers:
            tower.draw()

        # Draw disks
        for tower in self.towers:
            for disk in tower.disks:
                disk.draw()

        # Draw moving disk
        if self.moving_disk:
            self.moving_disk.draw()

        # Draw move counter with card style
        counter_rect = (WIDTH - 150, 70, 130, 50)
        draw_material_rect(screen, (255, 255, 255), counter_rect, 4, 2)
        moves_text = self.main_font.render(f"Moves: {self.moves}", True, TEXT_PRIMARY)
        screen.blit(moves_text, (WIDTH - 140, 85))

        # Draw win message with material card
        if self.game_won:
            win_rect = (WIDTH // 2 - 150, 70, 300, 60)
            draw_material_rect(screen, WIN_COLOR, win_rect, 4, 3)
            win_text = self.title_font.render("You Win!", True, (255, 255, 255))
            screen.blit(win_text, (WIDTH // 2 - win_text.get_width() // 2, 85))

        # Draw buttons
        self.draw_buttons()

        pygame.display.flip()

    def draw_buttons(self):
        # Create a card for the controls
        control_card_rect = (20, HEIGHT - 80, WIDTH - 40, 60)
        draw_material_rect(screen, (255, 255, 255), control_card_rect, 4, 3)

        # Reset button
        reset_rect = (40, HEIGHT - 70, 100, 40)
        draw_material_button(screen, reset_rect, "Reset", self.button_font, PRIMARY_COLOR, (255, 255, 255))

        # Auto Solve button
        solve_rect = (160, HEIGHT - 70, 120, 40)
        draw_material_button(screen, solve_rect, "Auto Solve", self.button_font, ACCENT_COLOR, (255, 255, 255))

        # Disk count buttons
        for i in range(1, MAX_DISKS + 1):
            disk_rect = (300 + (i-1)*80, HEIGHT - 70, 60, 40)
            draw_material_button(screen, disk_rect, str(i), self.button_font,
                                BUTTON_COLOR if i != self.num_disks else PRIMARY_COLOR,
                                TEXT_PRIMARY if i != self.num_disks else (255, 255, 255))

    def handle_click(self, pos):
        x, y = pos

        # Check if a tower was clicked
        if not self.auto_solving and y < HEIGHT - 90 and y > 60:
            for i, tower in enumerate(self.towers):
                if abs(x - tower.x) < 60:
                    self.handle_tower_click(i)
                    return

        # Check if reset button was clicked
        if 40 <= x <= 140 and HEIGHT - 70 <= y <= HEIGHT - 30:
            self.reset()
            return

        # Check if auto solve button was clicked
        if 160 <= x <= 280 and HEIGHT - 70 <= y <= HEIGHT - 30:
            self.start_auto_solve()
            return

        # Check if disk count buttons were clicked
        for i in range(1, MAX_DISKS + 1):
            if 300 + (i-1)*80 <= x <= 360 + (i-1)*80 and HEIGHT - 70 <= y <= HEIGHT - 30:
                self.reset(i)
                return

    def handle_tower_click(self, tower_index):
        if self.selected_tower is None:
            # No tower selected yet, try to select this one
            if self.towers[tower_index].disks:
                self.selected_tower = tower_index
                self.selected_disk = self.towers[tower_index].remove_top_disk()
                self.moving_disk = self.selected_disk
                self.moving_disk.moving = True
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = 100  # Move up
        else:
            # A tower was already selected, try to move the disk
            if tower_index == self.selected_tower:
                # Put the disk back
                self.towers[tower_index].add_disk(self.selected_disk)
            elif self.towers[tower_index].can_add_disk(self.selected_disk):
                # Move the disk to the new tower
                self.moving_disk.target_x = self.towers[tower_index].x
                self.moving_disk.target_y = self.towers[tower_index].y - len(self.towers[tower_index].disks) * DISK_HEIGHT - DISK_HEIGHT // 2
                self.towers[tower_index].add_disk(self.selected_disk)
                self.moves += 1

                # Check if the game is won
                if len(self.towers[2].disks) == self.num_disks:
                    self.game_won = True
            else:
                # Invalid move, put the disk back
                self.towers[self.selected_tower].add_disk(self.selected_disk)

            self.selected_tower = None
            self.selected_disk = None
            self.moving_disk = None

    def reset(self, num_disks=None):
        if num_disks is not None:
            self.num_disks = min(num_disks, MAX_DISKS)

        self.towers = [
            Tower(WIDTH // 4),
            Tower(WIDTH // 2),
            Tower(3 * WIDTH // 4)
        ]

        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, DISK_COLORS[(i - 1) % len(DISK_COLORS)])
            self.towers[0].add_disk(disk)

        self.moves = 0
        self.selected_tower = None
        self.selected_disk = None
        self.moving_disk = None
        self.auto_solving = False
        self.solution_steps = []
        self.solution_index = 0
        self.game_won = False

    def solve_hanoi(self, n, source, target, auxiliary, steps):
        if n > 0:
            self.solve_hanoi(n-1, source, auxiliary, target, steps)
            steps.append((source, target))
            self.solve_hanoi(n-1, auxiliary, target, source, steps)

    def start_auto_solve(self):
        if self.auto_solving:
            return

        # Reset the game first
        self.reset(self.num_disks)

        # Generate solution steps
        self.solution_steps = []
        self.solve_hanoi(self.num_disks, 0, 2, 1, self.solution_steps)
        self.solution_index = 0
        self.auto_solving = True
        self.last_move_time = time.time()

    def update_auto_solve(self):
        if not self.auto_solving or self.solution_index >= len(self.solution_steps):
            return

        current_time = time.time()
        if current_time - self.last_move_time < 1.0:  # Wait 1 second between moves
            return

        source, target = self.solution_steps[self.solution_index]

        # Make the move
        if self.towers[source].disks:
            disk = self.towers[source].remove_top_disk()
            self.towers[target].add_disk(disk)
            self.moves += 1

            # Check if the game is won
            if len(self.towers[2].disks) == self.num_disks:
                self.game_won = True
                self.auto_solving = False

        self.solution_index += 1
        self.last_move_time = current_time

        if self.solution_index >= len(self.solution_steps):
            self.auto_solving = False

def main():
    game = Game(3)

    running = True
    while running:
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN:
                if event.button == 1:  # Left mouse button
                    game.handle_click(event.pos)

        game.update_auto_solve()
        game.draw()
        clock.tick(60)

    pygame.quit()
    sys.exit()

if __name__ == "__main__":
    main()

Code result material polished:

import random

import pygame
import sys
import time
import math

# Initialize pygame
pygame.init()

# Constants
WIDTH, HEIGHT = 800, 600
FPS = 60
MIN_DISKS = 3
MAX_DISKS = 7

# Material Design Colors
LIGHT_THEME = {
    "background": (245, 245, 245),  # Grey 100
    "surface": (255, 255, 255),     # White
    "primary": (33, 150, 243),      # Blue 500
    "primary_dark": (25, 118, 210), # Blue 700
    "primary_light": (100, 181, 246), # Blue 300
    "secondary": (255, 152, 0),     # Orange 500
    "text_primary": (33, 33, 33),   # Grey 900
    "text_secondary": (117, 117, 117), # Grey 600
    "tower": (96, 125, 139),        # Blue Grey 500
    "error": (244, 67, 54),         # Red 500
    "success": (76, 175, 80),       # Green 500
}

DARK_THEME = {
    "background": (48, 48, 48),     # Grey 900
    "surface": (66, 66, 66),        # Grey 800
    "primary": (33, 150, 243),      # Blue 500
    "primary_dark": (25, 118, 210), # Blue 700
    "primary_light": (100, 181, 246), # Blue 300
    "secondary": (255, 152, 0),     # Orange 500
    "text_primary": (255, 255, 255),# White
    "text_secondary": (189, 189, 189), # Grey 400
    "tower": (176, 190, 197),       # Blue Grey 300
    "error": (244, 67, 54),         # Red 500
    "success": (76, 175, 80),       # Green 500
}

# Material Design Disk Colors - Light Theme
DISK_COLORS_LIGHT = [
    (244, 67, 54),    # Red 500
    (156, 39, 176),   # Purple 500
    (33, 150, 243),   # Blue 500
    (76, 175, 80),    # Green 500
    (255, 193, 7),    # Amber 500
    (255, 87, 34),    # Deep Orange 500
    (0, 188, 212),    # Cyan 500
]

# Material Design Disk Colors - Dark Theme
DISK_COLORS_DARK = [
    (229, 115, 115),  # Red 300
    (186, 104, 200),  # Purple 300
    (100, 181, 246),  # Blue 300
    (129, 199, 132),  # Green 300
    (255, 213, 79),   # Amber 300
    (255, 138, 101),  # Deep Orange 300
    (77, 208, 225),   # Cyan 300
]

# Set up the display
screen = pygame.display.set_mode((WIDTH, HEIGHT))
pygame.display.set_caption("Tower of Hanoi - Material Design")
clock = pygame.time.Clock()

# Helper function for drawing rounded rectangles
def draw_rounded_rect(surface, color, rect, radius=10, border=0, border_color=(0, 0, 0)):
    """Draw a rounded rectangle with optional border"""
    x, y, width, height = rect

    if border > 0:
        # Draw border first (as a slightly larger rectangle)
        pygame.draw.rect(surface, border_color,
                        (x-border, y-border, width+2*border, height+2*border),
                        0, radius+border)

    # Draw the main rectangle
    pygame.draw.rect(surface, color, (x, y, width, height), 0, radius)

# Helper function for drawing material shadows
def draw_shadow(surface, rect, radius=10, alpha=30, offset=(0, 4), blur=4):
    """Draw a soft shadow under a rectangle"""
    shadow_surf = pygame.Surface((rect[2] + blur * 2, rect[3] + blur * 2), pygame.SRCALPHA)
    pygame.draw.rect(shadow_surf, (0, 0, 0, alpha),
                    (blur, blur, rect[2], rect[3]), 0, radius)

    # Apply simple blur by scaling down and up
    scale_factor = 0.5
    small_surf = pygame.transform.smoothscale(shadow_surf,
                                            (int(shadow_surf.get_width() * scale_factor),
                                             int(shadow_surf.get_height() * scale_factor)))
    blurred = pygame.transform.smoothscale(small_surf, shadow_surf.get_size())

    # Blit the shadow
    surface.blit(blurred, (rect[0] - blur + offset[0], rect[1] - blur + offset[1]))

# Helper function for drawing material buttons
def draw_material_button(surface, rect, text, font, colors, is_hover=False, is_active=False):
    """Draw a material design button with hover and active states"""
    base_color = colors["primary"] if is_active else colors["surface"]
    text_color = colors["text_primary"] if not is_active else (255, 255, 255)

    # Draw shadow
    if not is_hover:
        draw_shadow(surface, rect, radius=4, offset=(0, 2), blur=4)
    else:
        draw_shadow(surface, rect, radius=4, offset=(0, 4), blur=6)

    # Draw button
    draw_rounded_rect(surface, base_color, rect, radius=4)

    # Draw ripple effect if hovering
    if is_hover and not is_active:
        hover_color = (*base_color[:3], 20)  # Semi-transparent overlay
        hover_surf = pygame.Surface((rect[2], rect[3]), pygame.SRCALPHA)
        pygame.draw.rect(hover_surf, hover_color, (0, 0, rect[2], rect[3]), 0, 4)
        surface.blit(hover_surf, (rect[0], rect[1]))

    # Draw text
    text_surf = font.render(text, True, text_color)
    text_rect = text_surf.get_rect(center=(rect[0] + rect[2]//2, rect[1] + rect[3]//2))
    surface.blit(text_surf, text_rect)

# Helper function for drawing material cards
def draw_material_card(surface, rect, colors, elevation=2):
    """Draw a material design card with elevation"""
    # Draw shadow
    draw_shadow(surface, rect, radius=8, offset=(0, elevation), blur=elevation*2)

    # Draw card
    draw_rounded_rect(surface, colors["surface"], rect, radius=8)

class Disk:
    def __init__(self, size, color, theme_colors):
        self.size = size
        self.base_color = color
        self.theme_colors = theme_colors
        self.x = 0
        self.y = 0
        self.target_x = 0
        self.target_y = 0
        self.moving = False
        self.dragging = False
        self.drag_offset_x = 0
        self.drag_offset_y = 0
        self.width = (size + 1) * 30
        self.height = 24
        self.animation_progress = 0
        self.animation_duration = 0.3  # seconds
        self.animation_start_time = 0
        self.animation_start_pos = (0, 0)

    def draw(self, surface):
        # Calculate disk rectangle
        rect = (self.x - self.width // 2, self.y - self.height // 2, self.width, self.height)

        # Draw shadow
        if not self.dragging:
            shadow_offset = 2
            draw_shadow(surface, rect, radius=self.height//2, offset=(0, shadow_offset), blur=4)
        else:
            # Larger shadow when dragging
            shadow_offset = 6
            draw_shadow(surface, rect, radius=self.height//2, offset=(0, shadow_offset), blur=8, alpha=40)

        # Draw disk with rounded corners
        draw_rounded_rect(surface, self.base_color, rect, radius=self.height//2)

        # Add a subtle highlight on top for 3D effect
        highlight_rect = (self.x - self.width // 2, self.y - self.height // 2, self.width, self.height // 3)
        highlight_color = tuple(min(c + 30, 255) for c in self.base_color[:3])
        draw_rounded_rect(surface, highlight_color, highlight_rect, radius=self.height//2)

    def contains_point(self, point):
        """Check if the disk contains the given point"""
        return (abs(point[0] - self.x) <= self.width // 2 and
                abs(point[1] - self.y) <= self.height // 2)

    def start_drag(self, mouse_pos):
        """Start dragging the disk"""
        self.dragging = True
        self.drag_offset_x = self.x - mouse_pos[0]
        self.drag_offset_y = self.y - mouse_pos[1]

    def update_drag(self, mouse_pos):
        """Update the disk position while dragging"""
        if self.dragging:
            self.x = mouse_pos[0] + self.drag_offset_x
            self.y = mouse_pos[1] + self.drag_offset_y

    def end_drag(self):
        """End dragging the disk"""
        self.dragging = False

    def start_animation(self, target_x, target_y):
        """Start animating the disk to a new position"""
        self.moving = True
        self.animation_start_time = time.time()
        self.animation_progress = 0
        self.animation_start_pos = (self.x, self.y)
        self.target_x = target_x
        self.target_y = target_y

    def update_animation(self):
        """Update the disk animation"""
        if not self.moving:
            return False

        current_time = time.time()
        elapsed = current_time - self.animation_start_time
        self.animation_progress = min(elapsed / self.animation_duration, 1.0)

        # Use easeOutCubic easing function for smooth animation
        progress = 1 - (1 - self.animation_progress) ** 3

        # Update position
        self.x = self.animation_start_pos[0] + (self.target_x - self.animation_start_pos[0]) * progress
        self.y = self.animation_start_pos[1] + (self.target_y - self.animation_start_pos[1]) * progress

        # Check if animation is complete
        if self.animation_progress >= 1.0:
            self.x = self.target_x
            self.y = self.target_y
            self.moving = False
            return True

        return False

class Tower:
    def __init__(self, x, y, theme_colors):
        self.x = x
        self.y = y
        self.theme_colors = theme_colors
        self.disks = []
        self.base_width = 120
        self.pole_height = 220
        self.pole_width = 8
        self.highlight = False
        self.highlight_alpha = 0
        self.highlight_direction = 1

    def draw(self, surface):
        tower_color = self.theme_colors["tower"]

        # Draw base shadow
        base_rect = (self.x - self.base_width//2, self.y, self.base_width, 10)
        draw_shadow(surface, base_rect, radius=5, offset=(0, 2), blur=4)

        # Draw base
        draw_rounded_rect(surface, tower_color, base_rect, radius=5)

        # Draw tower pole with subtle gradient
        for i in range(self.pole_height):
            # Create subtle gradient effect
            shade = max(0, min(20, i // 10))
            color = tuple(max(0, min(255, c + shade)) for c in tower_color[:3])
            pygame.draw.rect(surface, color,
                            (self.x - self.pole_width//2, self.y - self.pole_height + i,
                             self.pole_width, 1))

        # Draw highlight if this tower is a valid drop target
        if self.highlight:
            self.highlight_alpha += self.highlight_direction * 5
            if self.highlight_alpha >= 60:
                self.highlight_alpha = 60
                self.highlight_direction = -1
            elif self.highlight_alpha <= 20:
                self.highlight_alpha = 20
                self.highlight_direction = 1

            highlight_color = (*self.theme_colors["primary"][:3], self.highlight_alpha)
            highlight_surf = pygame.Surface((self.base_width + 20, self.pole_height + 10), pygame.SRCALPHA)
            pygame.draw.rect(highlight_surf, highlight_color,
                            (0, 0, self.base_width + 20, self.pole_height + 10), 0, 10)
            surface.blit(highlight_surf,
                        (self.x - (self.base_width + 20)//2, self.y - self.pole_height - 5))

    def add_disk(self, disk):
        """Add a disk to this tower"""
        disk_y = self.y - len(self.disks) * disk.height - disk.height // 2
        disk.x = self.x
        disk.y = disk_y
        self.disks.append(disk)

    def remove_top_disk(self):
        """Remove and return the top disk from this tower"""
        if self.disks:
            return self.disks.pop()
        return None

    def can_add_disk(self, disk):
        """Check if a disk can be added to this tower"""
        if not self.disks:
            return True
        return disk.size < self.disks[-1].size

    def get_top_disk(self):
        """Get the top disk without removing it"""
        if self.disks:
            return self.disks[-1]
        return None

    def contains_point(self, point):
        """Check if the tower contains the given point for dropping"""
        return abs(point[0] - self.x) < self.base_width // 2

    def get_top_position(self):
        """Get the position for a new disk at the top of the tower"""
        disk_y = self.y - len(self.disks) * 24 - 24 // 2
        return self.x, disk_y

    def set_highlight(self, highlight):
        """Set whether this tower should be highlighted as a valid drop target"""
        self.highlight = highlight

class Button:
    def __init__(self, x, y, width, height, text, theme_colors, action=None):
        self.rect = (x, y, width, height)
        self.text = text
        self.theme_colors = theme_colors
        self.action = action
        self.hover = False
        self.active = False

        # Load font
        try:
            self.font = pygame.font.SysFont('Roboto', 18)
        except:
            self.font = pygame.font.SysFont(None, 18)

    def draw(self, surface):
        draw_material_button(surface, self.rect, self.text, self.font,
                            self.theme_colors, self.hover, self.active)

    def contains_point(self, point):
        x, y, width, height = self.rect
        return (x <= point[0] <= x + width and y <= point[1] <= y + height)

    def set_hover(self, hover):
        self.hover = hover

    def set_active(self, active):
        self.active = active

    def click(self):
        if self.action:
            self.action()

class Game:
    def __init__(self, num_disks=3, dark_mode=False):
        self.num_disks = min(max(num_disks, MIN_DISKS), MAX_DISKS)
        self.dark_mode = dark_mode
        self.theme = DARK_THEME if dark_mode else LIGHT_THEME
        self.disk_colors = DISK_COLORS_DARK if dark_mode else DISK_COLORS_LIGHT

        # Game state
        self.moves = 0
        self.start_time = time.time()
        self.elapsed_time = 0
        self.game_won = False
        self.show_win_animation = False
        self.win_animation_start = 0
        self.win_particles = []

        # Tower setup
        tower_y = HEIGHT - 100
        self.towers = [
            Tower(WIDTH // 4, tower_y, self.theme),
            Tower(WIDTH // 2, tower_y, self.theme),
            Tower(3 * WIDTH // 4, tower_y, self.theme)
        ]

        # Disk interaction state
        self.selected_disk = None
        self.source_tower = None
        self.last_valid_position = (0, 0)

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, self.disk_colors[(i - 1) % len(self.disk_colors)], self.theme)
            self.towers[0].add_disk(disk)

        # Load fonts
        try:
            self.title_font = pygame.font.SysFont('Roboto', 36)
            self.main_font = pygame.font.SysFont('Roboto', 24)
            self.button_font = pygame.font.SysFont('Roboto', 18)
        except:
            self.title_font = pygame.font.SysFont(None, 36)
            self.main_font = pygame.font.SysFont(None, 24)
            self.button_font = pygame.font.SysFont(None, 18)

        # Create UI buttons
        self.create_buttons()

        # Feedback message
        self.feedback_message = ""
        self.feedback_color = self.theme["text_primary"]
        self.feedback_timer = 0

    def create_buttons(self):
        """Create all UI buttons"""
        self.buttons = []

        # Reset button
        self.buttons.append(Button(20, HEIGHT - 60, 100, 40, "Reset", self.theme,
                                  action=lambda: self.reset()))

        # Theme toggle button
        theme_text = "Light Mode" if self.dark_mode else "Dark Mode"
        self.buttons.append(Button(140, HEIGHT - 60, 120, 40, theme_text, self.theme,
                                  action=lambda: self.toggle_theme()))

        # Disk count buttons
        for i in range(MIN_DISKS, MAX_DISKS + 1):
            x_pos = 280 + (i - MIN_DISKS) * 70
            self.buttons.append(Button(x_pos, HEIGHT - 60, 60, 40, str(i), self.theme,
                                      action=lambda i=i: self.reset(i)))

    def toggle_theme(self):
        """Toggle between light and dark themes"""
        self.dark_mode = not self.dark_mode
        self.theme = DARK_THEME if self.dark_mode else LIGHT_THEME
        self.disk_colors = DISK_COLORS_DARK if self.dark_mode else DISK_COLORS_LIGHT

        # Update tower colors
        for tower in self.towers:
            tower.theme_colors = self.theme

        # Update disk colors
        for tower in self.towers:
            for i, disk in enumerate(tower.disks):
                disk.theme_colors = self.theme
                disk.base_color = self.disk_colors[(disk.size) % len(self.disk_colors)]

        if self.selected_disk:
            self.selected_disk.theme_colors = self.theme
            self.selected_disk.base_color = self.disk_colors[(self.selected_disk.size) % len(self.disk_colors)]

        # Recreate buttons with new theme
        self.create_buttons()

    def draw(self, surface):
        # Fill background
        surface.fill(self.theme["background"])

        # Draw app bar
        pygame.draw.rect(surface, self.theme["primary"], (0, 0, WIDTH, 60))
        title_text = self.title_font.render("Tower of Hanoi", True, (255, 255, 255))
        surface.blit(title_text, (20, 15))

        # Draw towers
        for tower in self.towers:
            tower.draw(surface)

        # Draw disks on towers
        for tower in self.towers:
            for disk in tower.disks:
                disk.draw(surface)

        # Draw selected disk (being dragged) on top
        if self.selected_disk:
            self.selected_disk.draw(surface)

        # Draw move counter and timer
        self.draw_stats(surface)

        # Draw buttons
        for button in self.buttons:
            button.draw(surface)

        # Draw disk count indicator
        self.draw_disk_count_indicator(surface)

        # Draw feedback message
        if self.feedback_message and time.time() < self.feedback_timer:
            self.draw_feedback(surface)

        # Draw win animation
        if self.show_win_animation:
            self.draw_win_animation(surface)

        # Draw win message
        if self.game_won:
            self.draw_win_message(surface)

    def draw_stats(self, surface):
        # Create a card for stats
        stats_rect = (WIDTH - 200, 70, 180, 80)
        draw_material_card(surface, stats_rect, self.theme)

        # Draw move counter
        moves_text = self.main_font.render(f"Moves: {self.moves}", True, self.theme["text_primary"])
        surface.blit(moves_text, (WIDTH - 180, 85))

        # Draw timer
        minutes = int(self.elapsed_time) // 60
        seconds = int(self.elapsed_time) % 60
        time_text = self.main_font.render(f"Time: {minutes:02d}:{seconds:02d}", True,
                                         self.theme["text_primary"])
        surface.blit(time_text, (WIDTH - 180, 115))

    def draw_disk_count_indicator(self, surface):
        # Draw text above disk count buttons
        text = self.button_font.render("Number of Disks:", True, self.theme["text_primary"])
        surface.blit(text, (280, HEIGHT - 90))

        # Highlight the current disk count button
        for button in self.buttons[2:]:  # Skip Reset and Theme buttons
            button.set_active(button.text == str(self.num_disks))

    def draw_feedback(self, surface):
        text = self.main_font.render(self.feedback_message, True, self.feedback_color)
        text_rect = text.get_rect(center=(WIDTH // 2, HEIGHT - 100))
        surface.blit(text, text_rect)

    def draw_win_message(self, surface):
        # Create a card for the win message
        win_rect = (WIDTH // 2 - 150, 70, 300, 60)
        draw_material_card(surface, win_rect, self.theme, elevation=4)

        # Draw win text
        win_text = self.title_font.render("You Win!", True, self.theme["success"])
        text_rect = win_text.get_rect(center=(WIDTH // 2, 100))
        surface.blit(win_text, text_rect)

    def draw_win_animation(self, surface):
        # Generate particles
        if time.time() - self.win_animation_start < 2.0:
            if len(self.win_particles) < 100 and random.random() < 0.3:
                self.win_particles.append({
                    'x': random.randint(0, WIDTH),
                    'y': random.randint(0, HEIGHT // 2),
                    'size': random.randint(5, 15),
                    'color': random.choice([self.theme["primary"], self.theme["secondary"],
                                           self.theme["success"]]),
                    'speed': random.uniform(1, 3),
                    'angle': random.uniform(0, 2 * math.pi)
                })

        # Update and draw particles
        particles_to_keep = []
        for particle in self.win_particles:
            # Update position
            particle['y'] += particle['speed']
            particle['x'] += math.sin(particle['angle']) * 0.5

            # Draw particle
            pygame.draw.circle(surface, particle['color'],
                              (int(particle['x']), int(particle['y'])),
                              particle['size'])

            # Keep particles that are still on screen
            if particle['y'] < HEIGHT:
                particles_to_keep.append(particle)

        self.win_particles = particles_to_keep

        # End animation after a while
        if time.time() - self.win_animation_start > 3.0:
            self.show_win_animation = False

    def update(self):
        # Update elapsed time if game is not won
        if not self.game_won:
            self.elapsed_time = time.time() - self.start_time

        # Update disk animations
        for tower in self.towers:
            for disk in tower.disks:
                disk.update_animation()

    def handle_click(self, pos):
        # Check if a button was clicked
        for button in self.buttons:
            if button.contains_point(pos):
                button.click()
                return

        # Don't allow moves if game is won
        if self.game_won:
            return

        # Check if a tower was clicked
        if not self.selected_disk:
            # Try to select a disk
            for i, tower in enumerate(self.towers):
                if tower.disks and tower.get_top_disk().contains_point(pos):
                    self.source_tower = i
                    self.selected_disk = tower.remove_top_disk()
                    self.selected_disk.start_drag(pos)
                    self.last_valid_position = (self.selected_disk.x, self.selected_disk.y)
                    return
        else:
            # Try to place the disk
            self.end_disk_drag()

    def handle_mouse_motion(self, pos):
        # Update button hover states
        for button in self.buttons:
            button.set_hover(button.contains_point(pos))

        # Update selected disk position
        if self.selected_disk:
            self.selected_disk.update_drag(pos)

            # Highlight valid drop towers
            for i, tower in enumerate(self.towers):
                can_drop = tower.contains_point(pos) and tower.can_add_disk(self.selected_disk)
                tower.set_highlight(can_drop)

    def handle_mouse_up(self, pos):
        if self.selected_disk:
            self.end_disk_drag()

    def end_disk_drag(self):
        """End dragging the selected disk and place it on a tower if valid"""
        if not self.selected_disk:
            return

        valid_drop = False

        # Check if the disk is over a valid tower
        for i, tower in enumerate(self.towers):
            if tower.contains_point((self.selected_disk.x, self.selected_disk.y)):
                if tower.can_add_disk(self.selected_disk):
                    # Valid move
                    target_x, target_y = tower.get_top_position()
                    self.selected_disk.start_animation(target_x, target_y)
                    tower.add_disk(self.selected_disk)
                    self.selected_disk = None

                    # Update move counter
                    if i != self.source_tower:
                        self.moves += 1

                    # Check if the game is won
                    if len(self.towers[2].disks) == self.num_disks:
                        self.game_won = True
                        self.show_win_animation = True
                        self.win_animation_start = time.time()
                        import random
                        self.win_particles = []

                    valid_drop = True
                    break
                else:
                    # Invalid move - show feedback
                    self.show_feedback("Invalid move: Can't place larger disk on smaller disk",
                                      self.theme["error"])

        # If no valid drop, return the disk to its original tower
        if not valid_drop:
            self.towers[self.source_tower].add_disk(self.selected_disk)
            target_x, target_y = self.last_valid_position
            self.selected_disk.start_animation(target_x, target_y)
            self.selected_disk = None

        # Clear tower highlights
        for tower in self.towers:
            tower.set_highlight(False)

    def show_feedback(self, message, color=None):
        """Show a feedback message for a short time"""
        self.feedback_message = message
        self.feedback_color = color if color else self.theme["text_primary"]
        self.feedback_timer = time.time() + 2.0  # Show for 2 seconds

    def reset(self, num_disks=None):
        """Reset the game with the specified number of disks"""
        if num_disks is not None:
            self.num_disks = min(max(num_disks, MIN_DISKS), MAX_DISKS)

        # Reset game state
        self.moves = 0
        self.start_time = time.time()
        self.elapsed_time = 0
        self.game_won = False
        self.show_win_animation = False
        self.selected_disk = None
        self.source_tower = None

        # Reset towers
        self.towers = [
            Tower(WIDTH // 4, HEIGHT - 100, self.theme),
            Tower(WIDTH // 2, HEIGHT - 100, self.theme),
            Tower(3 * WIDTH // 4, HEIGHT - 100, self.theme)
        ]

        # Initialize the first tower with disks
        for i in range(self.num_disks, 0, -1):
            disk = Disk(i - 1, self.disk_colors[(i - 1) % len(self.disk_colors)], self.theme)
            self.towers[0].add_disk(disk)

        # Update disk count indicator
        for button in self.buttons[2:]:  # Skip Reset and Theme buttons
            button.set_active(button.text == str(self.num_disks))



def main():
    # Import needed only for win animation
    import random

    # Initialize game
    game = Game(3)

    # Main game loop
    running = True
    while running:
        # Handle events
        for event in pygame.event.get():
            if event.type == pygame.QUIT:
                running = False
            elif event.type == pygame.MOUSEBUTTONDOWN:
                if event.button == 1:  # Left mouse button
                    game.handle_click(event.pos)
            elif event.type == pygame.MOUSEMOTION:
                game.handle_mouse_motion(event.pos)
            elif event.type == pygame.MOUSEBUTTONUP:
                if event.button == 1:  # Left mouse button
                    game.handle_mouse_up(event.pos)

        # Update game state
        game.update()

        # Draw everything
        game.draw(screen)

        # Update display
        pygame.display.flip()

        # Cap the frame rate
        clock.tick(FPS)

    # Clean up
    pygame.quit()
    sys.exit()

if __name__ == "__main__":
    main()

How to burn money on VPC Endpoints

Martin Nanchev — Tue, 20 May 2025 19:21:25 +0000

Martin Nanchev for AWS Community Builders

May 20 '25

How (not) to Burn Money on VPC Endpoints (So You Don't Have To)

#aws #vpc #networking #costs

9 min read

[Boost]

Martin Nanchev — Tue, 20 May 2025 17:19:13 +0000