Amazon Bedrock агентите се како паметни асистенти за вашата AWS инфраструктура – тие можат да размислуваат, да одлучат што да прават следно и да покренат акции со користење на функциите на Lambda. Во оваа статија, ќе ви покажам како изградив супервизорски агент кој оркестрира повеќе AWS Lambdas на: 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. Ајде да провериме на дијаграмот и со друг пример за тоа што е агентот, за подобра визуелизација и разбирање: Корисникот го повикува агентот Bedrock (1) со некоја задача, да речеме: „Колку телевизори имате на залиха?“ Агентот знае од дефиниран повик дека ако прашањето е поврзано со проверка на состојбата на залихата, тие треба да ја повикаат (2) акциската група „база на податоци“ (3, AG). Во базата на податоци AG, дефиниравме ламбда функција за употреба (4), а оваа ламбда ќе го провери статусот во табелата DynamoDB (5), ќе го добие одговорот (6,7) и ќе го врати одговорот на корисникот (8). Ајде да провериме уште еден пример: Секој агент може да има повеќе групи на дејства, на пример, сакаме да добиеме информации за некои ресурси на AWS, како што се сите задачи на ECS, логиката е иста како и за претходниот. И уште еден пример: Додадовме уште еден AG со EKS групи за акција. Како што можете да видите тука, секоја група за акција може да има повеќе од една ламбда функција за поставување барања. Акционата група и функцијата lambda може да имаат било која функционалност, дури и ако треба да добиете податоци од API на трета страна за да ги добиете податоците за времето или достапноста на авионските билети. Се надевам дека сега е малку јасно, и да се вратиме на нашата поставеност на супервизорски агент: Во AWS конзолата, отворете Bedrock → Агенти → Креирај агент Дајте му име и создадете Откако ќе се создаде, можете да го промените моделот ако сакате или да го задржите Клод по дефолт. Додадете опис и инструкции за агентот. групата за акција што ќе ја креираме во следниот чекор Вие сте главниот агент за надзор на AWS. Цел: Помогнете да се анализира AWS инфраструктурата. Активни групи : ec2: list_instances → враќа инстанца листа + instanceIds Правилата се: Никогаш не ги повикувајте AWS APIs директно. For EC2: Call ec2__list_instances Секогаш користете „мислење“ пред дејствување. Вие сте главниот агент за надзор на AWS. Цел: Помогнете да се анализира AWS инфраструктурата. Активни групи : ec2: list_instances → враќа инстанца листа + instanceIds Правилата се: Никогаш не ги повикувајте AWS APIs директно. For EC2: Call ec2__list_instances Секогаш користете „мислење“ пред дејствување. Забелешка : ec2 - име на групата за акција list_instances - име на функција, како што споменав порано - можете да имате повеќе функции за секоја група на акции Кликнете на “Save” И копчињата "Подготви" на врвот.Подготви ќе бидат активни откако ќе зачувате. Превртете се надолу до групата акции → додадете Активна група. повикување - креирање на нова ламбда функција, каде мора да биде иста како што е дефинирано во инструкциите на агентот list_instances Додадете име и опис на групата за акција, кликнете на Креирај и повторно "Зачувај" и "Приготви". Одете на функцијата lambda, Bedrock ја создаде функцијата со EC2 префикс во името и додадете го овој код: 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)}' } ЗАБЕЛЕШКА: одговорот на функцијата мора да биде во Bedrock-специфичен формат, детали може да се најдат во документацијата: https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html Откако ќе го ажурирате кодот на функцијата - одете на функцијата Конфигурација → овластувања → име на улогата креирајте нова политика во линија: Како на JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Сега можеме да се вратиме на нашиот агент и кликнете на "Тест", внесете текст за да проверите дали навистина функционира: Првата група за акција функционира како што се очекува, овозможувајќи додавање на уште една група за акција за да ги наведете метриките на cloudwatch: Името на акционата група - cloudwatch Име на функцијата е getMetrics, додадете опис и параметри, бидејќи оваа lambda мора да ја знае инстанцата или инстанците за да ги провери метриките на Ажурирајте го повикот за агент за да објасните како сакаме да ја користиме новата група за акција, и повторно кликнете на "Зачувај" и "Приготви" Вие сте главниот агент за надзор на AWS. Цел: Помогнете да се анализира AWS инфраструктурата. Активни групи : ec2: describeInstances → враќа листа на инстанци + instanceIds cloudwatch: getMetrics → бара instance_ids Правилата се: Никогаш не ги повикувајте AWS APIs директно. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Комбинирајте ги резултатите. Секогаш користете „мислење“ пред дејствување. Вие сте главниот агент за надзор на AWS. Цел: Помогнете да се анализира AWS инфраструктурата. Активни групи : ec2: describeInstances → враќа листа на инстанци + instanceIds cloudwatch: getMetrics → бара instance_ids Правилата се: Никогаш не ги повикувајте AWS APIs директно. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Комбинирајте ги резултатите. Секогаш користете „мислење“ пред дејствување. Сега овозможува да го ажурирате нашиот 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)}" } И ажурирајте ги дозволите на cloudwatch lambda како што направивме за ec2 lambda: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } И повторно да го тестирате Имаме EC2 и CloudWatch групи за акција, и тие можат да бидат повикани од агентот за да добијат листа на EC2 инстанции и нивните CPU метрики. Наместо агентот да ги повикува EC2 и CloudWatch одделно, супервизорот се грижи за таа логика. Прво ја повикува функцијата EC2 за да ги добие сите инстанци, потоа ги пренесува оние ИД на инстанци на функцијата CloudWatch за да ги добие метриките, и на крајот ги комбинира сите во еден јасен резултат. На овој начин, агентот треба да повика само една акција - супервизорот - додека супервизорот ги координира сите чекори во позадина. Дајте му име и опис Give the function name and description И ажурирајте ги инструкциите на агентот за да избегнете директен повик до акциските групи ec2 и CloudWatch: Потоа кликнете на “Save” и “Prepare”. ажурирајте го контролорот ламбда функција код, 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}" И повторно, додадете надзорни lambda дозволи за да ги повикате нашите EC2 и Cloudwatch функции, на пример: { "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>" ] } ] } Да ја тестираме функцијата повторно, и изненадувачки не успева Јас ги провери мојот супервизор надзорник функција дневници и да го видите ова Еден тоа се чини дека тоа не покажува ништо корисно, но не - на навестување тоа 3000.00ms. неговата дефолт ламбда функција временски период, ви овозможува да го прилагодите. Оди на функцијата супервизор - конфигурација - општо и уредување на временски период параметар , го променив на 10 секунди И тоа помогна! Можете да продолжите да ја проширувате оваа функционалност со додавање на АWS анализа за фактурирање за да ги пронајдете најскапите ресурси или најмногу , и така натаму и не мора да бидете ограничени само од AWS ресурси. скапи ec2 инстанции што ги користите
