Los agentes de Amazon Bedrock son como asistentes inteligentes para su infraestructura de AWS: pueden razonar, decidir qué hacer a continuación y desencadenar acciones utilizando las funciones de Lambda. En este artículo, mostraré cómo construí un Supervisor Agent que orquesta múltiples Lambdas de AWS a: List EC2 instances, Fetch their CPU metrics from CloudWatch, Combine both results intelligently — all without the agent ever calling AWS APIs directly. By the end, you’ll understand how Bedrock Agents work, how to use action groups, and how to chain Lambdas through a supervisor function — a clean, scalable pattern for multi-step automation. Compruebe el diagrama y con otro ejemplo de lo que es el agente, para una mejor visualidad y comprensión: El usuario llama al agente Bedrock (1) con alguna tarea, digamos, “¿cuántos televisores tienes en stock?”.El agente sabe por una solicitud definida que si la pregunta está relacionada con la verificación del estado del stock, necesitan llamar (2) al grupo de acción “database” (3, AG). En la base de datos AG, definimos una función lambda para usar (4), y esta lambda comprobará el estado en la tabla DynamoDB (5), obtendrá la respuesta (6,7) y devolverá la respuesta al usuario (8). Veamos un ejemplo más: Cada agente puede tener varios grupos de acción, por ejemplo, queremos obtener información sobre algunos recursos de AWS, como listar todas las tareas de ECS, la lógica es la misma que para la anterior. Y más ejemplo: Hemos añadido otro AG con grupos de acción EKS. Como se puede ver aquí, cada grupo de acción puede tener más de una función lambda para realizar solicitudes. El grupo de acción y la función lambda pueden tener cualquier funcionalidad, incluso si necesita obtener datos de una API de terceros para obtener datos meteorológicos o disponibilidad de billetes de vuelo. Espero que sea un poco claro ahora, y volvamos a nuestra configuración de agente de supervisión: En la consola de AWS, abra Bedrock → Agentes → Crear agente Dale un nombre y crea Una vez creado, puede cambiar el modelo si desea o mantener a Claude por defecto. Agregar descripción e instrucciones para el agente.El grupo de acción que crearemos en el siguiente paso Usted es el principal agente de supervisores de AWS. Objetivo: Ayudar a analizar la infraestructura de AWS. Grupos de Acción: ec2: list_instances → devuelve lista de instancias + instanceIds Las reglas: Never call AWS APIs directly. For EC2: Call ec2__list_instances Siempre use “pensar” antes de actuar. Usted es el principal agente de supervisores de AWS. Objetivo: Ayudar a analizar la infraestructura de AWS. Grupos de Acción: ec2: list_instances → devuelve lista de instancias + instanceIds Las reglas: Nunca llame las APIs de AWS directamente. For EC2: Call ec2__list_instances Siempre use “pensar” antes de actuar. Notas : ec2 - nombre del grupo de acción list_instances - nombre de función, como mencioné anteriormente - puede tener múltiples funciones por cada grupo de acción Haga clic en “Salvar” Y los botones “Preparar” en la parte superior.Preparar estará activo una vez que salves. Desplácese hacia abajo hasta el grupo de acciones → añadir Nombre - EC2. grupo de acción. invocación - crear una nueva función lambda, donde debe ser el mismo que definimos en las instrucciones del agente list_instances Añade el nombre y la descripción del grupo de acción, haga clic en Crear y de nuevo en “Guardar” y “Preparar”. Vaya a su función lambda, Bedrock creó la función con el prefixo EC2 en el nombre y agregó este código: import logging from typing import Dict, Any from http import HTTPStatus import boto3 logger = logging.getLogger() logger.setLevel(logging.INFO) ec2_client = boto3.client('ec2') def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: """ AWS Lambda handler for processing Bedrock agent requests related to EC2 instances. Supports: - Listing all EC2 instances - Describing a specific instance by ID """ try: action_group = event['actionGroup'] function = event['function'] message_version = event.get('messageVersion', 1) parameters = event.get('parameters', []) response_text = "" if function == "list_instances": # List all EC2 instances instances = ec2_client.describe_instances() instance_list = [] for reservation in instances['Reservations']: for instance in reservation['Instances']: instance_list.append({ 'InstanceId': instance.get('InstanceId'), 'State': instance.get('State', {}).get('Name'), 'InstanceType': instance.get('InstanceType'), 'PrivateIpAddress': instance.get('PrivateIpAddress', 'N/A'), 'PublicIpAddress': instance.get('PublicIpAddress', 'N/A') }) response_text = f"Found {len(instance_list)} EC2 instance(s): {instance_list}" elif function == "describe_instance": # Expect a parameter with the instance ID instance_id_param = next((p for p in parameters if p['name'] == 'instanceId'), None) if not instance_id_param: raise KeyError("Missing required parameter: instanceId") instance_id = instance_id_param['value'] result = ec2_client.describe_instances(InstanceIds=[instance_id]) instance = result['Reservations'][0]['Instances'][0] response_text = ( f"Instance {instance_id} details: " f"State={instance['State']['Name']}, " f"Type={instance['InstanceType']}, " f"Private IP={instance.get('PrivateIpAddress', 'N/A')}, " f"Public IP={instance.get('PublicIpAddress', 'N/A')}" ) else: response_text = f"Unknown function '{function}' requested." # Format Bedrock agent response response_body = { 'TEXT': { 'body': response_text } } action_response = { 'actionGroup': action_group, 'function': function, 'functionResponse': { 'responseBody': response_body } } response = { 'response': action_response, 'messageVersion': message_version } logger.info('Response: %s', response) return response except KeyError as e: logger.error('Missing required field: %s', str(e)) return { 'statusCode': HTTPStatus.BAD_REQUEST, 'body': f'Error: {str(e)}' } except Exception as e: logger.error('Unexpected error: %s', str(e)) return { 'statusCode': HTTPStatus.INTERNAL_SERVER_ERROR, 'body': f'Internal server error: {str(e)}' } NOTA: la respuesta de la función debe estar en formato específico de Bedrock, los detalles se pueden encontrar en la documentación: https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html Después de actualizar el código de la función, vaya a la función Configuración → permisos → nombre de rol crear una nueva política en línea: Como en JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Ahora podemos volver a nuestro agente y hacer clic en “Test”, introducir texto para comprobar si realmente funciona: El primer grupo de acción funciona como se esperaba, añadiendo otro grupo de acción para listar las métricas de cloudwatch: Nombre del grupo de acción - Cloudwatch Nombre de la función es getMetrics, añade descripción y parámetros, ya que esta lambda debe conocer la instancia o intances para comprobar las métricas de Actualizar el prompt de agente para explicar cómo queremos usar el nuevo grupo de acción, y hacer clic de nuevo en "Guardar" y "Preparar" Usted es el principal agente de supervisores de AWS. Objetivo: Ayudar a analizar la infraestructura de AWS. Grupos de Acción: ec2: describeInstancias → devuelve lista de instancias + instanceIds cloudwatch: getMetrics → necesita instance_ids Las reglas: Nunca llame las APIs de AWS directamente. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Combinar los resultados. Siempre use “pensar” antes de actuar. Usted es el principal agente de supervisores de AWS. Objetivo: Ayudar a analizar la infraestructura de AWS. Grupos de Acción: ec2: describeInstancias → devuelve lista de instancias + instanceIds cloudwatch: getMetrics → necesita instance_ids Las reglas: Nunca llame las APIs de AWS directamente. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Combinar los resultados. Siempre use “pensar” antes de actuar. Ahora permite actualizar nuestro código de función de cloudwatch: import boto3 import datetime import logging import json from typing import Dict, Any from http import HTTPStatus logger = logging.getLogger() logger.setLevel(logging.INFO) def lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]: try: action_group = event["actionGroup"] function = event["function"] message_version = event.get("messageVersion", 1) parameters = event.get("parameters", []) region = "us-east-1" instance_ids = [] # --- Parse parameters --- for param in parameters: if param.get("name") == "region": region = param.get("value") elif param.get("name") == "instance_ids": raw_value = param.get("value") if isinstance(raw_value, str): # Clean up stringified list from Bedrock agent raw_value = raw_value.strip().replace("[", "").replace("]", "").replace("'", "") instance_ids = [x.strip() for x in raw_value.split(",") if x.strip()] elif isinstance(raw_value, list): instance_ids = raw_value logger.info(f"Parsed instance IDs: {instance_ids}") if not instance_ids: response_text = f"No instance IDs provided for CloudWatch metrics in {region}." else: cloudwatch = boto3.client("cloudwatch", region_name=region) now = datetime.datetime.utcnow() start_time = now - datetime.timedelta(hours=1) metrics_output = [] for instance_id in instance_ids: try: metric = cloudwatch.get_metric_statistics( Namespace="AWS/EC2", MetricName="CPUUtilization", Dimensions=[{"Name": "InstanceId", "Value": instance_id}], StartTime=start_time, EndTime=now, Period=300, Statistics=["Average"] ) datapoints = metric.get("Datapoints", []) if datapoints: datapoints.sort(key=lambda x: x["Timestamp"]) avg_cpu = round(datapoints[-1]["Average"], 2) metrics_output.append(f"{instance_id}: {avg_cpu}% CPU (avg last hour)") else: metrics_output.append(f"{instance_id}: No recent CPU data") except Exception as e: logger.error(f"Error fetching metrics for {instance_id}: {e}") metrics_output.append(f"{instance_id}: Error fetching metrics") response_text = ( f"CPU Utilization (last hour) in {region}:\n" + "\n".join(metrics_output) ) # --- Bedrock Agent response format --- response_body = { "TEXT": { "body": response_text } } action_response = { "actionGroup": action_group, "function": function, "functionResponse": { "responseBody": response_body } } response = { "response": action_response, "messageVersion": message_version } logger.info("Response: %s", response) return response except Exception as e: logger.error(f"Unexpected error: {e}") return { "statusCode": HTTPStatus.INTERNAL_SERVER_ERROR, "body": f"Internal server error: {str(e)}" } Y actualice los permisos de lambda de cloudwatch como lo hicimos para ec2 lambda: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } y lo pruebe de nuevo Tenemos grupos de acción de EC2 y CloudWatch, y pueden ser llamados desde el agente para obtener una lista de instancias de EC2 y sus métricas de CPU. En lugar de que el agente llame tanto a EC2 como a CloudWatch por separado, el Supervisor se ocupa de esa lógica. Primero llama la función EC2 para obtener todas las instancias, luego pasa esos ID de instancia a la función CloudWatch para obtener métricas, y finalmente combina todo en un resultado claro. De esta manera, el agente solo necesita llamar una acción, el Supervisor, mientras que el Supervisor coordina todos los pasos en el fondo. Dale un nombre y una descripción Nombre y descripción de la función Y actualice las instrucciones del agente para evitar una llamada directa a los grupos de acción de ec2 y CloudWatch: Haga clic en “Salvar” y “Preparar”. Actualiza el código de función de supervisor lambda, NOTE: need to update your EC2 and Cloudwatch functions name in the code below: import boto3 import json import logging import re import ast logger = logging.getLogger() logger.setLevel(logging.INFO) lambda_client = boto3.client("lambda") def lambda_handler(event, context): try: action_group = event["actionGroup"] function = event["function"] parameters = event.get("parameters", []) message_version = event.get("messageVersion", "1.0") # Parse parameters region = "us-east-1" for param in parameters: if param.get("name") == "region": region = param.get("value") # Decide routing if function == "analyzeInfrastructure": logger.info("Supervisor: calling EC2 and CloudWatch") # Step 1: call EC2 Lambda ec2_payload = { "actionGroup": "ec2", "function": "list_instances", "parameters": [{"name": "region", "value": region}], "messageVersion": "1.0" } ec2_response = invoke_lambda("ec2-yeikw", ec2_payload) #### CHANGE TO YOUR EC2 FUNCTION NAME instances = extract_instance_ids(ec2_response) # Step 2: call CloudWatch Lambda (if instances found) if instances: cw_payload = { "actionGroup": "cloudwatch", "function": "getMetrics", "parameters": [ {"name": "region", "value": region}, {"name": "instance_ids", "value": instances} ], "messageVersion": "1.0" } cw_response = invoke_lambda("cloudwatch-ef6ty", cw_payload) #### CHANGE TO YOUR CLOUDWATCH FUNCTION NAME final_text = merge_responses(ec2_response, cw_response) else: final_text = "No instances found to analyze." else: final_text = f"Unknown function: {function}" # Construct Bedrock-style response response = { "messageVersion": message_version, "response": { "actionGroup": action_group, "function": function, "functionResponse": { "responseBody": { "TEXT": {"body": final_text} } } } } logger.info("Supervisor response: %s", response) return response except Exception as e: logger.exception("Error in supervisor") return { "statusCode": 500, "body": f"Supervisor error: {str(e)}" } def invoke_lambda(name, payload): """Helper to call another Lambda and parse response""" response = lambda_client.invoke( FunctionName=name, InvocationType="RequestResponse", Payload=json.dumps(payload), ) result = json.loads(response["Payload"].read()) return result def extract_instance_ids(ec2_response): """Extract instance IDs from EC2 Lambda response""" try: body = ec2_response["response"]["functionResponse"]["responseBody"]["TEXT"]["body"] # Try to extract JSON-like data after "Found X EC2 instance(s):" if "Found" in body and "[" in body and "]" in body: data_part = body.split(":", 1)[1].strip() try: instances = ast.literal_eval(data_part) # safely parse the list return [i["InstanceId"] for i in instances if "InstanceId" in i] except Exception: pass # fallback regex in case of plain text return re.findall(r"i-[0-9a-f]+", body) except Exception as e: logger.error("extract_instance_ids error: %s", e) return [] def merge_responses(ec2_resp, cw_resp): """Combine EC2 and CloudWatch outputs""" ec2_text = ec2_resp["response"]["functionResponse"]["responseBody"]["TEXT"]["body"] cw_text = cw_resp["response"]["functionResponse"]["responseBody"]["TEXT"]["body"] return f"{ec2_text}\n\n{cw_text}" Y otra vez, añade permisos de lambda supervisor para invocar nuestras funciones de EC2 y Cloudwatch, por ejemplo: { "Version": "2012-10-17", "Statement": [ { "Sid": "VisualEditor0", "Effect": "Allow", "Action": "lambda:InvokeFunction", "Resource": [ "arn:aws:lambda:us-east-1:<account_id>:function:ec2-<id>", "arn:aws:lambda:us-east-1:<account_id>:function:cloudwatch-<id>" ] } ] } Hemos probado la función de nuevo, y sorprendentemente fracasó He comprobado mis registros de funciones de supervisor de supervisor y he visto esto Uno parece que no muestra nada útil, pero no - el indicio que 3000.00ms. su función lambda temporal predeterminado, permite ajustarlo. Vaya a la función supervisor - configuración - general y editar el parámetro de Timeout , cambié a 10 segundos ¡Y eso ayudó! Puede continuar ampliando esta funcionalidad añadiendo análisis de facturación de AWS para encontrar los recursos más caros o más , y así sucesivamente y no tiene que limitarse únicamente a los recursos de AWS. Siéntese libre de tener alguna funcionalidad externa. costosas instancias ec2 que ejecuta
