Els agents d'Amazon Bedrock són com els assistents intel·ligents per a la vostra infraestructura AWS: poden raonar, decidir què fer a continuació i desencadenar accions utilitzant les funcions de Lambda. En aquest article, mostraré com he construït un agent supervisor que orquestra múltiples Lambdas d'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. Anem a comprovar en el diagrama i amb un altre exemple del que l'agent és, per a una millor visibilitat i comprensió: L'usuari fa una trucada a l'agent Bedrock (1) amb alguna tasca, diguem, "quants televisors teniu en estoc?" L'agent sap per una invitació definida que si la pregunta està relacionada amb la comprovació de l'estat d'estoc, han de trucar (2) al grup d'acció "base de dades" (3, AG). En la base de dades AG, hem definit una funció lambda per utilitzar (4), i aquesta lambda comprovarà l'estat en la taula DynamoDB (5), obtindrà la resposta (6,7) i retornarà la resposta a l'usuari (8). Vegem un altre exemple: Cada agent pot tenir múltiples grups d'acció, per exemple, volem obtenir informació sobre alguns recursos d'AWS, com ara una llista de totes les tasques ECS, la lògica és la mateixa que per a l'agent anterior. I més exemples: Hem afegit un AG més amb grups d'acció EKS. Com veieu aquí, cada grup d'acció pot tenir més d'una funció lambda per fer sol·licituds. El grup d'acció i la funció lambda poden tenir qualsevol funcionalitat, fins i tot si necessiteu obtenir dades d'una API de tercers per recollir dades meteorològiques o disponibilitat de bitllets de vol. Espero que ara estigui una mica clar, i tornem a la nostra configuració d'agent supervisor: A la consola AWS, obri Bedrock → Agents → Crear agent Dóna-li un nom i crea Una vegada creat, podeu canviar el model si voleu o mantenir Claude per defecte. Afegir descripció i instruccions per a l'Agent. El grup d'acció que crearem en el següent pas Vostè és el principal agent de supervisors de AWS. Objectiu: Ajuda a analitzar la infraestructura d'AWS. Grups d’Acció: ec2: list_instances → retorna la llista d'instàncies + instanceIds Les regles: Mai trucar directament a les APIs d'AWS. For EC2: Call ec2__list_instances Sempre utilitzeu “pensar” abans d’actuar. Vostè és el principal agent de supervisors de AWS. Objectiu: Ajuda a analitzar la infraestructura d'AWS. Grups d’Acció: ec2: list_instances → retorna la llista d'instàncies + instanceIds Les regles: Mai trucar directament a les APIs d'AWS. For EC2: Call ec2__list_instances Sempre utilitzeu “pensar” abans d’actuar. Notació : ec2 - nom del grup d'acció list_instances - nom de funció, com he esmentat anteriorment - podeu tenir múltiples funcions per cada grup d'acció Feu clic a “Salvar” I els botons “Preparar” a la part superior. Prepare estarà actiu una vegada que salvi. Desplaçar-se cap avall al grup d'accions → afegir Nom - EC2. grup d'acció. invocació - crear una nova funció lambda, on ha de ser el mateix que hem definit en les instruccions de l'agent list_instances Afegir el nom i la descripció del grup d'acció, fer clic a Crear i tornar a "Salvar" i "Preparar". Anar a la funció lambda, Bedrock va crear la funció amb el prefix EC2 en el nom i afegir aquest codi: 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 resposta de la funció ha de ser en format específic de Bedrock, detalls es poden trobar a la documentació: https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html Després d'actualitzar el codi de la funció, aneu a la funció Configuració → permisos → nom de rol crear una nova política en línia: Com a JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Ara podem tornar al nostre agent i fer clic a "Test", introduir text per comprovar si realment funciona: El primer grup d'acció funciona com s'esperava, permet afegir un altre grup d'acció a la llista de les mètriques de cloudwatch: Nom del grup d'acció - cloudwatch El nom de la funció és getMetrics, afegiu descripció i paràmetres, ja que aquesta lambda ha de conèixer l'instància o intances per comprovar les mètriques de Actualitzar l'agent prompt per explicar com volem utilitzar el nou grup d'acció, i tornar a fer clic a "Salvar" i "Preparar" Vostè és el principal agent de supervisors de AWS. Objectiu: Ajuda a analitzar la infraestructura d'AWS. Grups d’Acció: ec2: describeInstances → retorna la llista d'instàncies + instanceIds cloudwatch: getMetrics → necessita instance_ids Les regles: Mai trucar directament a les APIs d'AWS. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Combina els resultats. Sempre utilitzeu “pensar” abans d’actuar. Vostè és el principal agent de supervisors de AWS. Objectiu: Ajuda a analitzar la infraestructura d'AWS. Grups d’Acció: ec2: describeInstances → retorna la llista d'instàncies + instanceIds cloudwatch: getMetrics → necessita instance_ids Les regles: Mai trucar directament a les APIs d'AWS. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Combina els resultats. Sempre utilitzeu “pensar” abans d’actuar. Ara permet actualitzar el nostre codi de funció 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)}" } I actualitzar els permisos lambda de cloudwatch com ho vam fer per ec2 lambda: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } Provar-ho de nou Disposem de grups d'acció EC2 i CloudWatch, i els podem trucar des de l'agent per obtenir una llista d'instàncies EC2 i les seves mètriques de CPU. En lloc que l'agent truqui tant a EC2 com a CloudWatch per separat, el Supervisor s'encarrega d'aquesta lògica. primer truca a la funció EC2 per obtenir totes les instàncies, després passa aquests ID d'instàncies a la funció CloudWatch per recollir mètriques, i finalment combina tot en un resultat clar. D'aquesta manera, l'agent només necessita trucar a una acció - el Supervisor - mentre que el Supervisor coordina tots els passos en el fons. Donar-li un nom i una descripció Escriu el nom i la descripció de la funció I actualitzar les instruccions de l'agent per evitar una trucada directa als grups d'acció ec2 i CloudWatch: Feu clic a “Salvar” i “Preparar”. actualitzar el codi de funció lambda del supervisor, 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}" I de nou, afegiu permisos de supervisor lambda per invocar les nostres funcions EC2 i Cloudwatch, per exemple: { "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>" ] } ] } Deixem que torni a provar la funció, i sorprenentment fracassa Vaig comprovar els registres de funcions del supervisor del meu supervisor i vaig veure això Un sembla que no mostra res útil, però no - la pista que 3000.00ms. el seu temporal de la funció lambda predeterminada, permet ajustar-lo. Anar a la funció supervisor - configuració - general i editar el paràmetre de temps, vaig canviar a 10 segons I això va ajudar! Podeu continuar ampliant aquesta funcionalitat afegint l'anàlisi de facturació d'AWS per trobar els recursos més cars o els més utilitzats. , i així successivament i no has de ser limitat només pels recursos d'AWS. Instàncies ec2 cares que s'executen
