Cómo Construir un Agente Autónomo de Despliegue con Amazon Bedrock, MCP y AWS CDK 🛠️☁️

[ Desarrollador / CLI ]
│
│ Requerimiento ("Stack web SSR con base de datos gestionada")
▼
┌──────────────────────────────────────────────────────────────┐
│ Agente Orquestador (Amazon Bedrock - Converse API) │
│ • Modelo: Anthropic Claude 3.5 Sonnet / Amazon Nova Pro │
│ • Tarea: Planificación de topología y selección de herramientas│
└──────────────────────────────┬───────────────────────────────┘
│
┌──────────────────┴──────────────────┐
│ Invocación de Herramientas (Tool Use)│
▼ ▼
┌──────────────────────────────┐ ┌──────────────────────────────┐
│ Servidor MCP: CDK Core │ │ Servidor MCP: FinOps Guard │
│ • generate_cdk_construct() │ │ • estimate_monthly_cost() │
│ • cdk_synth() │ │ • validate_budget_limit() │
└──────────────┬───────────────┘ └──────────────┬───────────────┘
│ │
│ Template Validado │ Costo < Límite
└──────────────────┬───────────────────┘
▼
┌──────────────────────────────────────┐
│ Motor de Despliegue (AWS CDK CLI) │
│ • cdk deploy --require-approval never│
└──────────────────┬───────────────────┘
▼
┌──────────────────────────────────────┐
│ AWS CloudFormation │
│ • ECS / Fargate / RDS / S3 / ACM │
└──────────────────────────────────────┘

Entrada de Usuario
└── 1. Extracción de Entidades (Framework, Tráfico, Persistencia)
└── 2. Consulta al Servidor FinOps
├── ¿Costo Proyectado > Límite ($30 USD/mes)?
│ └── [ABORTAR] Reportar alternativas serverless al usuario
└── ¿Costo Proyectado <= Límite?
└── 3. Generación y Síntesis de CDK
├── Error de Sintaxis / Tipado
│ └── Realimentar error a Bedrock para autocorrección
└── Síntesis Exitosa (cdk.out validado)
└── 4. Despliegue en CloudFormation
└── 5. Retorno de Endpoints y Secretos al Usuario

TOOLS_SCHEMA = [
{
"toolSpec": {
"name": "estimate_monthly_cost",
"description": "Calcula el costo proyectado en USD/mes según los tipos de instancia y servicios solicitados.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"compute_type": {
"type": "string",
"enum": ["FARGATE_SPOT", "FARGATE_ON_DEMAND", "LAMBDA", "EC2_T3_MICRO"],
"description": "El tipo de cómputo seleccionado."
},
"has_relational_db": {
"type": "boolean",
"description": "Indica si el stack requiere Aurora Serverless o RDS."
}
},
"required": ["compute_type", "has_relational_db"]
}
}
}
},
{
"toolSpec": {
"name": "deploy_cdk_stack",
"description": "Compila y despliega un stack de CDK predefinido pasando los parámetros validados.",
"inputSchema": {
"json": {
"type": "object",
"properties": {
"stack_name": {"type": "string"},
"environment": {"type": "string", "enum": ["dev", "prod"]},
"compute_type": {"type": "string"}
},
"required": ["stack_name", "environment", "compute_type"]
}
}
}
}
]

import subprocess
import json

BUDGET_HARD_LIMIT_USD = 50.0 # Límite estricto de FinOps

def handle_cost_estimation(compute_type: str, has_relational_db: bool) -> dict:
costs = {
"LAMBDA": 2.0,
"FARGATE_SPOT": 12.0,
"FARGATE_ON_DEMAND": 28.0,
"EC2_T3_MICRO": 9.5
}
base_cost = costs.get(compute_type, 30.0)
db_cost = 25.0 if has_relational_db else 0.0
total = base_cost + db_cost

approved = total <= BUDGET_HARD_LIMIT_USD
return {
"projected_cost_usd": total,
"budget_limit_usd": BUDGET_HARD_LIMIT_USD,
"approved": approved,
"policy_violation": None if approved else "El costo proyectado excede el límite máximo de FinOps."
}

def handle_cdk_deployment(stack_name: str, environment: str, compute_type: str) -> dict:
# Comando determinista de despliegue
cmd = [
"cdk", "deploy", stack_name,
"--context", f"env={environment}",
"--context", f"compute={compute_type}",
"--require-approval", "never"
]

try:
# Simulamos la llamada CLI para fines de auditoría controlada
process = subprocess.run(
["echo", f"Simulando: {' '.join(cmd)}"],
capture_output=True,
text=True,
check=True
)
return {
"status": "SUCCESS",
"output": process.stdout.strip(),
"stack_id": f"arn:aws:cloudformation:us-east-1:123456789012:stack/{stack_name}"
}
except subprocess.CalledProcessError as e:
return {"status": "FAILED", "error": e.stderr}

import boto3
import json
from tool_definitions import TOOLS_SCHEMA
from tool_handlers import handle_cost_estimation, handle_cdk_deployment

bedrock_client = boto3.client(service_name="bedrock-runtime", region_name="us-east-1")
MODEL_ID = "anthropic.claude-3-5-sonnet-20240620-v1:0"

SYSTEM_PROMPT = """Eres un Arquitecto de Infraestructura Autónomo en AWS.
Tu labor es aprovisionar stacks en la nube siguiendo estrictamente estas reglas:
1. NUNCA despliegues infraestructura sin antes estimar su costo mensual con 'estimate_monthly_cost'.
2. Si el costo excede el límite permitido, debes detenerte y sugerir una arquitectura más económica (ej. Serverless Lambda en lugar de Fargate).
3. Solo si la estimación es aprobada, procede con 'deploy_cdk_stack'.
"""

def execute_agent_loop(user_prompt: str):
messages = [{"role": "user", "content": [{"text": user_prompt}]}]
print(f"\n[USER]: {user_prompt}\n")

while True:
response = bedrock_client.converse(
modelId=MODEL_ID,
messages=messages,
system=[{"text": SYSTEM_PROMPT}],
toolConfig={"tools": TOOLS_SCHEMA}
)

message_output = response["output"]["message"]
messages.append(message_output)
stop_reason = response["stopReason"]

# Si el modelo solo responde con texto, terminamos
if stop_reason == "end_turn":
for block in message_output["content"]:
if "text" in block:
print(f"[AGENTE]: {block['text']}")
break

# Si el modelo solicita ejecutar una herramienta (Tool Use)
elif stop_reason == "tool_use":
tool_results = []
for block in message_output["content"]:
if "toolUse" in block:
tool_call = block["toolUse"]
tool_name = tool_call["name"]
tool_args = tool_call["input"]
tool_id = tool_call["toolUseId"]

print(f"-> [EJECUTANDO TOOL]: {tool_name} con parámetros: {tool_args}")

if tool_name == "estimate_monthly_cost":
result_data = handle_cost_estimation(**tool_args)
elif tool_name == "deploy_cdk_stack":
result_data = handle_cdk_deployment(**tool_args)
else:
result_data = {"error": "Herramienta desconocida"}

tool_results.append({
"toolResult": {
"toolUseId": tool_id,
"content": [{"json": result_data}]
}
})

messages.append({"role": "user", "content": tool_results})

if __name__ == "__main__":
prompt = "Necesito montar una API en producción usando Fargate On-Demand con una base de datos relacional."
execute_agent_loop(prompt)

[USER]: Necesito montar una API en producción usando Fargate On-Demand con una base de datos relacional.

-> [EJECUTANDO TOOL]: estimate_monthly_cost con parámetros: {'compute_type': 'FARGATE_ON_DEMAND', 'has_relational_db': True}

[AGENTE]: He calculado el costo proyectado para tu solicitud:
- Cómputo (Fargate On-Demand): $28.00 USD/mes
- Base de Datos Relacional: $25.00 USD/mes
- Costo Total Estimado: $53.00 USD/mes

El presupuesto máximo establecido por las políticas de FinOps es de $50.00 USD/mes. La operación ha sido bloqueada para prevenir sobrecostos no autorizados.

¿Deseas que reconfiguremos la arquitectura usando **FARGATE_SPOT** ($12.00/mes) o **AWS Lambda** ($2.00/mes) para mantenernos dentro del límite?