mirror of
https://github.com/awslabs/amazon-bedrock-agentcore-samples.git
synced 2025-09-08 20:50:46 +00:00
9f2f9f470f
- Complete serverless AI-powered AWS operations platform - Multi-Lambda architecture with Function URL deployment - Bedrock AgentCore Gateway integration with MCP protocol - 20 AWS service tools for comprehensive operations - Dual authentication: AWS SigV4 + Okta JWT - Natural language interface with streaming responses - DynamoDB conversation persistence - Docker-based MCP Tool Lambda with Strands framework - Production-ready with enterprise security patterns - Comprehensive documentation and setup guides - Read-only operations by default with write enablement guide - Interactive client with CLI interface - Complete Okta OAuth2 PKCE setup - Management scripts for gateway and target operations - Sanitized configuration with dummy data for public sharing
229 lines
8.3 KiB
Python
Executable File
229 lines
8.3 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""
|
|
Get Bedrock AgentCore Gateway Details
|
|
Pulls live data from Bedrock AgentCore Gateway endpoint (AWS is source of truth)
|
|
"""
|
|
import json
|
|
import boto3
|
|
import logging
|
|
import argparse
|
|
import sys
|
|
from datetime import datetime
|
|
from pathlib import Path
|
|
|
|
# Add configs directory to path
|
|
sys.path.append(str(Path(__file__).parent.parent / "configs"))
|
|
from config_manager import BedrockAgentCoreConfigManager
|
|
|
|
# Configure logging
|
|
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
|
|
logger = logging.getLogger(__name__)
|
|
|
|
def parse_arguments():
|
|
"""Parse command line arguments"""
|
|
parser = argparse.ArgumentParser(description='Get Bedrock AgentCore Gateway Details')
|
|
parser.add_argument('--gateway-id', required=True, help='Gateway ID to retrieve')
|
|
parser.add_argument("--endpoint", type=str, choices=["beta", "gamma", "production"], help="Endpoint to use (beta, gamma, production)")
|
|
parser.add_argument("--environment", type=str, default=None, help="Environment to use (dev, gamma, prod)")
|
|
return parser.parse_args()
|
|
|
|
def print_request(title, request_data):
|
|
"""Print formatted request"""
|
|
print(f"\n{title}")
|
|
print("=" * 60)
|
|
print(json.dumps(request_data, indent=2, default=str))
|
|
print("=" * 60)
|
|
|
|
def print_response(title, response_data):
|
|
"""Print formatted response"""
|
|
print(f"\n{title}")
|
|
print("=" * 60)
|
|
print(json.dumps(response_data, indent=2, default=str))
|
|
print("=" * 60)
|
|
|
|
def get_gateway_from_aws(config_manager, environment, gateway_id):
|
|
"""Get gateway details from AWS"""
|
|
|
|
# Get configuration for the specific environment
|
|
aws_config = config_manager.get_aws_config(environment)
|
|
endpoints = config_manager.get_bedrock_agentcore_endpoints()
|
|
|
|
print(f"Using Configuration:")
|
|
print(f" Environment: {environment}")
|
|
print(f" AWS Profile: {aws_config['profile']}")
|
|
print(f" AWS Region: {aws_config['region']}")
|
|
print(f" AWS Account: {aws_config['account']}")
|
|
print(f" Bedrock AgentCore Endpoint: {endpoints['control_plane']}")
|
|
|
|
# Create AWS session with environment-specific profile and region
|
|
session = boto3.Session(
|
|
profile_name=aws_config['profile'],
|
|
region_name=aws_config['region']
|
|
)
|
|
|
|
bedrock_agentcore_client = session.client(
|
|
'bedrock-agentcore-control',
|
|
endpoint_url=endpoints['control_plane']
|
|
)
|
|
|
|
# Prepare request
|
|
request_data = {
|
|
'gatewayIdentifier': gateway_id
|
|
}
|
|
|
|
print_request("GET GATEWAY REQUEST", request_data)
|
|
|
|
try:
|
|
# Get gateway details
|
|
response = bedrock_agentcore_client.get_gateway(**request_data)
|
|
|
|
print_response("GET GATEWAY RESPONSE", response)
|
|
|
|
return response, bedrock_agentcore_client
|
|
|
|
except Exception as e:
|
|
logger.error(f"Failed to get gateway from AWS: {str(e)}")
|
|
print(f"\nFailed to get gateway from AWS: {str(e)}")
|
|
print(f" Check AWS profile '{aws_config['profile']}' and region '{aws_config['region']}'")
|
|
return None, None
|
|
|
|
def get_gateway_targets(bedrock_agentcore_client, gateway_id):
|
|
"""Get targets for the gateway"""
|
|
|
|
try:
|
|
response = bedrock_agentcore_client.list_gateway_targets(gatewayIdentifier=gateway_id)
|
|
return response.get('items', [])
|
|
except Exception as e:
|
|
logger.error(f"Failed to get targets for gateway {gateway_id}: {str(e)}")
|
|
return []
|
|
|
|
def display_gateway_details(config_manager, gateway_data, bedrock_agentcore_client):
|
|
"""Display formatted gateway details"""
|
|
|
|
print(f"\nGateway Details")
|
|
print("=" * 50)
|
|
|
|
# Basic information
|
|
gateway_id = gateway_data.get('gatewayId', 'Unknown')
|
|
print(f"Gateway ID: {gateway_id}")
|
|
print(f"Name: {gateway_data.get('name', 'Unknown')}")
|
|
print(f"Status: {gateway_data.get('status', 'Unknown')}")
|
|
print(f"Protocol Type: {gateway_data.get('protocolType', 'Unknown')}")
|
|
print(f"Authorizer Type: {gateway_data.get('authorizerType', 'Unknown')}")
|
|
print(f"Description: {gateway_data.get('description', 'None')}")
|
|
print(f"Role ARN: {gateway_data.get('roleArn', 'Unknown')}")
|
|
|
|
# Timestamps
|
|
if 'createdAt' in gateway_data:
|
|
print(f"Created At: {gateway_data['createdAt']}")
|
|
if 'updatedAt' in gateway_data:
|
|
print(f"Updated At: {gateway_data['updatedAt']}")
|
|
|
|
# Gateway URL/Endpoint
|
|
if 'gatewayUrl' in gateway_data:
|
|
print(f"Gateway URL: {gateway_data['gatewayUrl']}")
|
|
elif 'endpoint' in gateway_data:
|
|
print(f"Gateway Endpoint: {gateway_data['endpoint']}")
|
|
|
|
# MCP endpoint (calculated)
|
|
if gateway_id:
|
|
mcp_endpoint = config_manager.get_mcp_gateway_url(gateway_id)
|
|
print(f"MCP Endpoint: {mcp_endpoint}")
|
|
|
|
# Authorizer configuration
|
|
if 'authorizerConfiguration' in gateway_data:
|
|
print(f"\nAuthorizer Configuration:")
|
|
auth_config = gateway_data['authorizerConfiguration']
|
|
print(json.dumps(auth_config, indent=2))
|
|
|
|
# Protocol configuration
|
|
if 'protocolConfiguration' in gateway_data:
|
|
print(f"\nProtocol Configuration:")
|
|
protocol_config = gateway_data['protocolConfiguration']
|
|
print(json.dumps(protocol_config, indent=2))
|
|
|
|
# Encryption key
|
|
if 'kmsKeyArn' in gateway_data:
|
|
print(f"\nEncryption Key ARN: {gateway_data['kmsKeyArn']}")
|
|
|
|
# Get and display targets
|
|
print(f"\nAssociated Targets:")
|
|
targets = get_gateway_targets(bedrock_agentcore_client, gateway_id)
|
|
|
|
if targets:
|
|
total_tools = 0
|
|
print(f" Found {len(targets)} targets:")
|
|
|
|
for i, target in enumerate(targets, 1):
|
|
target_id = target.get('targetId', 'Unknown')
|
|
target_name = target.get('name', 'Unknown')
|
|
target_status = target.get('status', 'Unknown')
|
|
|
|
print(f" {i}. Target ID: {target_id}")
|
|
print(f" Name: {target_name}")
|
|
print(f" Status: {target_status}")
|
|
|
|
# Extract tool information
|
|
target_config = target.get('targetConfiguration', {})
|
|
if 'mcp' in target_config and 'lambda' in target_config['mcp']:
|
|
lambda_config = target_config['mcp']['lambda']
|
|
lambda_arn = lambda_config.get('lambdaArn', 'Unknown')
|
|
print(f" Lambda ARN: {lambda_arn}")
|
|
|
|
tool_schema = lambda_config.get('toolSchema', {})
|
|
if 'inlinePayload' in tool_schema:
|
|
tools = tool_schema['inlinePayload']
|
|
tool_count = len(tools)
|
|
total_tools += tool_count
|
|
tool_names = [tool.get('name', 'unknown') for tool in tools[:3]]
|
|
if len(tools) > 3:
|
|
tool_names.append('...')
|
|
print(f" Tools: {tool_count} ({', '.join(tool_names)})")
|
|
|
|
print(f"\n Total Tools Across All Targets: {total_tools}")
|
|
else:
|
|
print(f" No targets found")
|
|
|
|
def main():
|
|
"""Main function"""
|
|
args = parse_arguments()
|
|
|
|
# Initialize configuration manager
|
|
config_manager = BedrockAgentCoreConfigManager()
|
|
|
|
# Use default environment from config
|
|
environment = config_manager.get_default_environment()
|
|
|
|
print("Get Bedrock AgentCore Gateway Details")
|
|
print("=" * 40)
|
|
print(f"Environment: {environment}")
|
|
print(f"Gateway ID: {args.gateway_id}")
|
|
print(f"Timestamp: {datetime.now().isoformat()}")
|
|
print(f"Data Source: AWS Bedrock AgentCore API (Live)")
|
|
|
|
try:
|
|
# Validate configuration
|
|
if not config_manager.validate_config():
|
|
print("Configuration validation failed")
|
|
sys.exit(1)
|
|
|
|
# Get gateway details from AWS
|
|
aws_data, bedrock_agentcore_client = get_gateway_from_aws(config_manager, environment, args.gateway_id)
|
|
|
|
if not aws_data:
|
|
print(f"Gateway {args.gateway_id} not found in AWS")
|
|
sys.exit(1)
|
|
|
|
# Display detailed information
|
|
display_gateway_details(config_manager, aws_data, bedrock_agentcore_client)
|
|
|
|
print(f"\nRetrieved gateway details from live AWS data")
|
|
|
|
except Exception as e:
|
|
logger.error(f"Operation failed: {str(e)}")
|
|
print(f"\nOperation failed: {str(e)}")
|
|
sys.exit(1)
|
|
|
|
if __name__ == "__main__":
|
|
main()
|