Amazon Bedrock Agents seperti asisten cerdas untuk infrastruktur AWS Anda – mereka dapat berpendapat, memutuskan apa yang harus dilakukan selanjutnya, dan memicu tindakan menggunakan fungsi Lambda. Dalam artikel ini, saya akan menunjukkan bagaimana saya membangun Supervisor Agent yang mengorchestrasikan beberapa AWS Lambdas untuk: 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. Mari kita periksa pada diagram dan dengan contoh lain dari apa agen, untuk visibilitas dan pemahaman yang lebih baik: Pengguna membuat panggilan ke agen Bedrock (1) dengan beberapa tugas, katakanlah, “berapa banyak TV yang Anda miliki?” Agen tahu dengan petunjuk yang didefinisikan bahwa jika pertanyaan terkait dengan memeriksa status stok, mereka perlu memanggil (2) kelompok tindakan “database” (3, AG). Dalam database AG, kami mendefinisikan fungsi lambda untuk digunakan (4), dan lambda ini akan memeriksa status di tabel DynamoDB (5), mendapatkan jawaban (6,7) dan mengembalikan jawaban kepada pengguna (8). Mari kita periksa satu contoh lagi: Setiap agen dapat memiliki beberapa kelompok tindakan, misalnya, kita ingin mendapatkan informasi tentang beberapa sumber daya AWS, seperti daftar semua tugas ECS, logika sama seperti untuk yang sebelumnya. Dan lebih banyak contoh: Kami menambahkan satu AG lagi dengan kelompok tindakan EKS. Seperti yang Anda lihat di sini, setiap kelompok tindakan dapat memiliki lebih dari satu fungsi lambda untuk membuat permintaan. Kelompok tindakan dan fungsi lambda dapat memiliki fungsi apa pun, bahkan jika Anda perlu mendapatkan data dari API pihak ketiga untuk mengambil data cuaca atau ketersediaan tiket penerbangan. Saya berharap itu sedikit jelas sekarang, dan mari kita kembali ke pengaturan agen supervisor kami: Di konsol AWS, buka Bedrock → Agen → Membuat agen Berikan nama dan buatlah Setelah dibuat, Anda dapat mengubah model jika Anda ingin atau menjaga Claude sebagai default. Tambahkan deskripsi dan instruksi untuk agen. kelompok tindakan yang akan kita buat pada langkah berikutnya Anda adalah Agen Supervisor AWS utama. Tujuan: Membantu menganalisis infrastruktur AWS. Kelompok aksi : ec2: list_instances → mengembalikan daftar instance + instanceIds Peraturan : Jangan pernah menghubungi AWS APIs secara langsung. For EC2: Call ec2__list_instances Selalu gunakan ‘berpikir’ sebelum bertindak. Anda adalah Agen Supervisor AWS utama. Tujuan: Membantu menganalisis infrastruktur AWS. Kelompok aksi : ec2: list_instances → mengembalikan daftar instance + instanceIds Peraturan : Jangan pernah menghubungi AWS APIs secara langsung. For EC2: Call ec2__list_instances Selalu gunakan ‘berpikir’ sebelum bertindak. Catatan : ec2 - nama kelompok aksi list_instances - nama fungsi, seperti yang saya sebutkan sebelumnya - Anda dapat memiliki beberapa fungsi per kelompok tindakan Klik pada “Save” Dan tombol “Prepare” di bagian atas.Prepare akan aktif setelah Anda menyimpan. Gulir ke bawah ke grup tindakan → tambahkan Nama - EC2. kelompok tindakan. invocation - membuat fungsi lambda baru, di mana harus sama seperti yang kami definisikan dalam instruksi agen list_instances Tambahkan nama grup tindakan dan deskripsi, klik Create dan lagi “Save” dan “Prepare”. Pergi ke fungsi lambda Anda, Bedrock membuat fungsi dengan prefix EC2 dalam nama dan menambahkan kode ini: 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)}' } Catatan: tanggapan fungsi harus dalam format Bedrock-spesifik, rincian dapat ditemukan dalam dokumentasi: https://docs.aws.amazon.com/bedrock/latest/userguide/agents-lambda.html Setelah Anda memperbarui kode fungsi Anda - pergi ke fungsi Konfigurasi → izin → nama peran membuat kebijakan inline baru: Untuk JSON: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "ec2:DescribeInstances" ], "Resource": [ "*" ] } ] } Sekarang kita bisa kembali ke agen kami dan klik “Test”, masukkan teks untuk memeriksa apakah itu benar-benar bekerja: Cool! kelompok tindakan pertama bekerja seperti yang diharapkan, memungkinkan untuk menambahkan satu lagi kelompok tindakan untuk daftar metrik cloudwatch: Nama kelompok aksi - cloudwatch Nama fungsi adalah getMetrics, tambahkan deskripsi dan parameter, karena lambda ini harus tahu instance atau intances untuk memeriksa metrik Memperbarui petunjuk agen untuk menjelaskan bagaimana kami ingin menggunakan kelompok tindakan baru, dan klik “Save” dan “Prepare” lagi Anda adalah Agen Supervisor AWS utama. Tujuan: Membantu menganalisis infrastruktur AWS. Kelompok aksi : ec2: describeInstances → mengembalikan daftar instance + instanceIds cloudwatch: getMetrics → membutuhkan instance_ids Peraturan : Jangan pernah menghubungi AWS APIs secara langsung. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Combine results. Selalu gunakan ‘berpikir’ sebelum bertindak. Anda adalah Agen Supervisor AWS utama. Tujuan: Membantu menganalisis infrastruktur AWS. Kelompok aksi : ec2: describeInstances → mengembalikan daftar instance + instanceIds cloudwatch: getMetrics → membutuhkan instance_ids Peraturan : Jangan pernah menghubungi AWS APIs secara langsung. For EC2 + CPU: Call ec2__describeInstances Extract instanceIds Call cloudwatch__getMetrics Menggabungkan hasil Selalu gunakan ‘berpikir’ sebelum bertindak. Sekarang memungkinkan untuk memperbarui kode fungsi cloudwatch kami: 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)}" } Dan memperbarui izin lambda cloudwatch seperti yang kami lakukan untuk ec2 lambda: { "Version": "2012-10-17", "Statement": [ { "Sid": "Statement1", "Effect": "Allow", "Action": [ "cloudwatch:GetMetricStatistics" ], "Resource": [ "*" ] } ] } dan coba lagi Kami memiliki kelompok tindakan EC2 dan CloudWatch, dan mereka dapat dipanggil dari agen untuk mendapatkan daftar instansi EC2 dan metrik CPU mereka. Alih-alih agen memanggil EC2 dan CloudWatch secara terpisah, Supervisor mengurus logika itu. ia pertama memanggil fungsi EC2 untuk mendapatkan semua instansi, kemudian mentransfer ID instansi tersebut ke fungsi CloudWatch untuk mengumpulkan metrik, dan akhirnya menggabungkan semuanya menjadi satu hasil yang jelas. Dengan cara ini, agen hanya perlu memanggil satu tindakan - Supervisor - sementara Supervisor mengoordinasikan semua langkah di latar belakang. Berikan nama dan deskripsi Masukkan nama dan deskripsi fungsi Dan memperbarui instruksi agen untuk menghindari panggilan langsung ke kelompok tindakan ec2 dan CloudWatch: Kemudian klik “Save” dan “Prepare”. Perbarui kode fungsi 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}" Dan sekali lagi, tambahkan izin lambda supervisor untuk memanggil fungsi EC2 dan Cloudwatch kami, misalnya: { "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>" ] } ] } Biarkan kami menguji fungsi lagi, dan secara mengejutkan gagal Saya memeriksa log fungsi supervisor saya dan melihat ini Satu tampaknya tidak menunjukkan apa-apa yang berguna, tetapi tidak - petunjuk itu 3000.00ms. fungsi lambda default timeout, memungkinkan untuk menyesuaikan. pergi ke fungsi supervisor - konfigurasi - umum dan mengedit parameter Timeout , saya berubah menjadi 10 detik Dan itu membantu! Anda dapat terus memperluas fungsionalitas ini dengan menambahkan analisis tagihan AWS untuk menemukan sumber daya yang paling mahal atau paling banyak. , dan sebagainya dan Anda tidak harus dibatasi hanya oleh sumber daya AWS. instansi ec2 yang mahal yang Anda jalankan
