Amazon Bedrock Agents የእርስዎ AWS ኢንቨስትራክሽን ለወንዶች እንደወንዶች ናቸው - እነርሱ መውሰድ ይችላሉ, ቀጣይ እንዴት እንደሚቻል, እና Lambda ባህሪያት በመጠቀም እንቅስቃሴዎችን ለመጀመር. በዚህ ጽሑፍ ውስጥ, እኔ ብዙ AWS Lambdas ለመርዳት አንድ Supervisor Agent እንዴት ለመፍጠር ይመልከቱ: 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) የ “database” ተግባር ቡድን (3, AG) ይደውሉ. በ database AG, እኛ አንድ lambda ተግባር ይጠቀማል (4), እና ይህ lambda በ DynamoDB ቱቦ ውስጥ የአካል ብስክሌት (5), መልስ (6,7) ያግኙ እና መልስ ወደ ተጠቃሚን ይመልከቱ (8). አንድ ተጨማሪ ለምሳሌ ይመልከቱ: እያንዳንዱ አግኝተዋል በርካታ የሥራ ቡድን ሊሆን ይችላል, ለምሳሌ, እኛ አንዳንድ AWS መሳሪያዎች ስለ መረጃ ለማግኘት ይፈልጋሉ, እንደ ሁሉም ECS መሳሪያዎች ዝርዝር እንደ, ትክክለኛነት የቀድሞው አግኝተዋል ጋር ተመሳሳይ ነው. ወደ አግኝተዋል አግኝተዋል አግኝተዋል አግኝተዋል አግኝተዋል ያውቃሉ - እኛ የሚፈልጉን ይጠቀማል, ከዚያ በኋላ ያውቃል. ተጨማሪ ለምሳሌ: በ EKS መተግበሪያ ቡድን ጋር አንድ ተጨማሪ AG ያካትታል. እያንዳንዱ መተግበሪያ ቡድን አንድ በላይ Lambda ተግባር ሊኖረው ይችላል. በዚህ ምሳሌ ውስጥ, አንዳንድ የ K8S ግምገማዎች ከ መተግበሪያዎች መለያ እና መቁረጫ ነው. የ Action Group እና Lambda ተግባር ማንኛውም ተግባራዊነት ሊሆን ይችላል, አንተ የአየር ውሂብ ወይም የበረራ ኬብል አጠቃቀም ለማግኘት የቴክኖሎጂ መተግበሪያ መተግበሪያ መተግበሪያ (API) ውሂብ ለማግኘት አለብዎት. አሁን ይህ ትንሽ ቀላል ይሆናል, እና የእኛ supervisor agent መተግበሪያ ላይ ይመልከቱ: በ AWS ኮንሰል በ Bedrock → Agents → Create agent ያግኙ አንድ ስም ያግኙ እና ለመፍጠር አንድ ጊዜ የተመሰረተ, እርስዎ የሚፈልጉ ከሆነ ሞዴል መተግበሪያ ይችላሉ, ወይም Claude በ default. I will change to Nova Premier 1.0 አግኝቷል.አንድ አግኝቷል.አንድ አግኝቷል.አንድ አግኝቷል. እርስዎ ዋና AWS Supervisor Agent ነው. መተግበሪያ: የ AWS Infrastructure አጠቃቀም ይረዳል. ክወና ቡድን: ec2: list_instances → instance list + instanceIds ያውቃል ግዴታዎች : የ AWS APIs ማንኛውም ጊዜ በቀጥታ ይደውሉ. For EC2: Call ec2__list_instances ሁልጊዜ ከባድነት በፊት "ወደ" ይጠቀማል. እርስዎ ዋና AWS Supervisor Agent ነው. መተግበሪያ: የ AWS Infrastructure አጠቃቀም ይረዳል. ክወና ቡድን: ec2: list_instances → instance list + instanceIds ያውቃል ግዴታዎች : የ AWS APIs ማንኛውም ጊዜ በቀጥታ ይደውሉ. For EC2: Call ec2__list_instances ሁልጊዜ ከባድነት በፊት "ወደ" ይጠቀማል. ዝርዝር: ec2 - እንቅስቃሴ ቡድን ስም list_instances - ተግባር ስም, እንደ እኔ በፊት ታውቃለሁ - እያንዳንዱ ተግባር ቡድን ላይ በርካታ ተግባሮች ሊሆን ይችላል እና “Save” ጠቅ ያድርጉ እና ከላይ ላይ የ "የመግቢያ" ግምገማዎች. የ "የመግቢያ" ማከማቻዎ በኋላ ይሰራል. መተግበሪያዎች ቡድን ወደ ታች ያግኙ → Add መተግበሪያ ቡድን. invocation - አንድ አዲስ lambda ተግባር ለመፍጠር, ይህም በ Agent instructions ውስጥ የተመሠረተ ተመሳሳይ መሆን አለበት list_instances አጠቃቀም ቡድን ስም እና ዝርዝር ያካትታል, ለመፍጠር ጠቅ ያድርጉ, ከዚያም “Save” እና “Prepare” ይጫኑ. ወደ 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 የክፍያ ኮድ ለማሻሻል በኋላ - የክፍያ → permissions → role name አንድ አዲስ inline ፖሊሲ ለመፍጠር ይጎብኙ: እንደ JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Now we can go back to our agent and click “Test”, text enter to check if it actually works: አሁን እኛ የእኛን መሣሪያን ይጎብኙ እና “Test” ጠቅ ያድርጉ. የ First Action Group እንደ የሚፈልጉትን ይሰራል, Cloudwatch መለያዎች ለማሳየት አንድ ተጨማሪ Action Group ያካትታል: የፕሮጀክሽን ቡድን ስም - cloudwatch የክፍያ ስም getMetrics ነው, ዝርዝር እና ፓርሜትር ያካትታል, ይህ lambda የክፍያ ወይም intances ማረጋገጥ ለማወቅ አለብዎት መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ መተግበሪያ እርስዎ ዋና AWS Supervisor Agent ነው. መተግበሪያ: የ AWS Infrastructure አጠቃቀም ይረዳል. ክወና ቡድን: ec2: describeInstances → ምልክት ዝርዝር + instanceIds ይውላል cloudwatch: getMetrics → needs instance_ids ግዴታዎች : Never call AWS APIs directly. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics ውጤቶችን ያካትታል. ሁልጊዜ ከባድነት በፊት "ወደ" ይጠቀማል. እርስዎ ዋና AWS Supervisor Agent ነው. መተግበሪያ: የ AWS Infrastructure አጠቃቀም ይረዳል. ክወና ቡድን: ec2: describeInstances → ምልክት ዝርዝር + instanceIds ይውላል cloudwatch: getMetrics → needs 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)}" } እና ለ ec2 lambda እንደ የ cloudwatch lambda ችሎታዎች ማሻሻል: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } ከዚያም ይሞክሩ እኛ EC2 እና CloudWatch ተግባራዊ ቡድን አላቸው, እና እነርሱ EC2 እይታዎች እና የ CPU ሜትሪኮች ዝርዝር ለማግኘት ከባድ ይደውሉ. አሁን, ይህ ሂደት ይበልጥ ጥንካሬ እና ውጤታማ ያደርጋል የ Supervisor ተግባር ያካትታል. በ EC2 እና በ CloudWatch መተግበሪያዎችን በመስመር ላይ በመጀመሪያ, EC2 ተግባር ለመውሰድ, ከዚያም እነዚህ instance IDs ወደ CloudWatch ተግባር ለመውሰድ እና በመጨረሻም ሁሉንም ነገር አንድ ቀላል ውጤት ውስጥ ያካትታል. በዚህ መንገድ, የ Agent ብቻ አንድ እንቅስቃሴ - የ Supervisor - ይደውሉ, ነገር ግን የ Supervisor በፊል ውስጥ ሁሉንም ደረጃዎችን ያካትታል. ይህ ቀላል, ፈጣን, እና ለመጠበቅ ቀላል ነው. አንድ ስም እና ትዕዛዝ ያግኙ ተግባር ስም እና ዝርዝር ያግኙ እና በ ec2 እና CloudWatch እንቅስቃሴ ቡድን ወደ ቀጣይ መዳረሻ ለመጠበቅ መተግበሪያዎች ማሻሻል: ከዚያም “Save” እና “Prepare” ጠቅ ያድርጉ. የ Supervisor Lambda Function Code ያግኙን. 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}" በተጨማሪም, የእኛ 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>" ] } ] } መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው መተግበሪያው እኔ የ Supervisor Supervisor ተግባር መለያዎች ይመልከቱ እና ይህን ይመልከቱ አንድ እርስዎ ምንም ምቹ ነገር ማየት አይችልም, ነገር ግን አይችልም - ይህ 3000.00ms ይሰጣል. የእርስዎ መደበኛ lambda ተግባር timeout, መተግበሪያን ይቀየዳል. supervisor ተግባር - configuration - general እና Timeout ፓራሚተር መተግበሪያ , እኔ ወደ 10 ሰከንዶች ይቀየዳል እርሱም እርዳታ ነበር! የ AWS የክፍያ ልኬት ልኬት ልኬት ልኬት ልኬት ልኬት ልኬት ልኬት ልኬት ልኬት ልኬት እናንተ ብቻ AWS መሳሪያዎች ላይ የተጠበቀ አይችልም. አንዳንድ ውስጣዊ አፈጻጸም ለማግኘት ነጻ ይሆናል. የ EC2 ኮምፒውተር እርስዎ ይጠቀማል
