yike5460/basic_code_interpreter.py

## basic_code_interpreter.py
#!/usr/bin/env python3
"""
Amazon Bedrock AgentCore Code Interpreter Demo

A fully functional demonstration of AgentCore's Code Interpreter capabilities,
showcasing secure code execution, file operations, and agent integration.
"""

import json
import boto3
import time
import os
import sys
import asyncio
from typing import Dict, Any, List, Optional
from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter
from bedrock_agentcore._utils import endpoints

class CodeInterpreterDemo:
    """
    Comprehensive demo of AgentCore Code Interpreter functionality
    """

    def __init__(self, region: str = "us-west-2"):
        self.region = region
        self.code_client = None
        self.session_id = None
        self.interpreter_id = None
        self.setup_clients()

    def setup_clients(self):
        """Initialize AWS clients for Code Interpreter"""
        try:
            # Set up endpoints
            data_plane_endpoint = endpoints.get_data_plane_endpoint(self.region)
            control_plane_endpoint = endpoints.get_control_plane_endpoint(self.region)

            # Create clients
            self.cp_client = boto3.client(
                "bedrock-agentcore-control",
                region_name=self.region,
                endpoint_url=control_plane_endpoint
            )

            self.dp_client = boto3.client(
                "bedrock-agentcore",
                region_name=self.region,
                endpoint_url=data_plane_endpoint
            )

            print("✅ AWS clients initialized successfully")

        except Exception as e:
            print(f"❌ Error initializing clients: {e}")
            sys.exit(1)

    def create_interpreter(self, execution_role_arn: str) -> str:
        """Create a new Code Interpreter instance"""
        try:
            unique_name = f"demo_interpreter_{int(time.time())}"

            response = self.cp_client.create_code_interpreter(
                name=unique_name,
                description="Demo Code Interpreter for AgentCore tutorial",
                executionRoleArn=execution_role_arn,
                networkConfiguration={'networkMode': 'PUBLIC'}
            )

            self.interpreter_id = response["codeInterpreterId"]
            print(f"✅ Code Interpreter created: {self.interpreter_id}")
            return self.interpreter_id

        except Exception as e:
            print(f"❌ Error creating interpreter: {e}")
            return None

    def start_session(self, timeout_seconds: int = 1800) -> str:
        """Start a new Code Interpreter session"""
        try:
            response = self.dp_client.start_code_interpreter_session(
                codeInterpreterIdentifier=self.interpreter_id,
                name="DemoSession",
                sessionTimeoutSeconds=timeout_seconds
            )

            self.session_id = response["sessionId"]
            print(f"✅ Session started: {self.session_id}")
            return self.session_id

        except Exception as e:
            print(f"❌ Error starting session: {e}")
            return None

    def execute_code(self, code: str, language: str = "python") -> Dict[str, Any]:
        """Execute code in the interpreter"""
        try:
            response = self.dp_client.invoke_code_interpreter(
                codeInterpreterIdentifier=self.interpreter_id,
                sessionId=self.session_id,
                # Valid Values: executeCode | executeCommand | readFiles | listFiles | removeFiles | writeFiles | startCommandExecution | getTask | stopTask
                name="executeCode",
                arguments={
                    "language": language,
                    "code": code
                }
            )

            # Process response stream
            results = []
            for event in response['stream']:
                if 'result' in event:
                    results.append(event['result'])

            return {"success": True, "results": results}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def write_files(self, files: List[Dict[str, str]]) -> Dict[str, Any]:
        """Write files to the interpreter sandbox"""
        try:
            response = self.dp_client.invoke_code_interpreter(
                codeInterpreterIdentifier=self.interpreter_id,
                sessionId=self.session_id,
                # Valid Values: executeCode | executeCommand | readFiles | listFiles | removeFiles | writeFiles | startCommandExecution | getTask | stopTask
                name="writeFiles",
                arguments={"content": files}
            )

            results = []
            for event in response['stream']:
                if 'result' in event:
                    results.append(event['result'])

            return {"success": True, "results": results}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def list_files(self, directory_path: str = "") -> Dict[str, Any]:
        """List files in the interpreter sandbox"""
        try:
            response = self.dp_client.invoke_code_interpreter(
                codeInterpreterIdentifier=self.interpreter_id,
                sessionId=self.session_id,
                name="listFiles",
                arguments={"directoryPath": directory_path}
            )

            results = []
            for event in response['stream']:
                if 'result' in event:
                    results.append(event['result'])

            return {"success": True, "results": results}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def read_files(self, file_paths: List[str]) -> Dict[str, Any]:
        """Read files from the interpreter sandbox"""
        try:
            response = self.dp_client.invoke_code_interpreter(
                codeInterpreterIdentifier=self.interpreter_id,
                sessionId=self.session_id,
                name="readFiles",
                arguments={"paths": file_paths}
            )

            results = []
            for event in response['stream']:
                if 'result' in event:
                    results.append(event['result'])

            return {"success": True, "results": results}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def download_file(self, remote_path: str, local_path: str, debug: bool = False) -> bool:
        """Download a file from the interpreter sandbox to local environment"""
        try:
            # Read the file from sandbox
            read_result = self.read_files([remote_path])

            if debug:
                print(f"🔍 Debug - Read result for {remote_path}:")
                # Handle bytes in response by converting to string representation
                try:
                    print(json.dumps(read_result, indent=2))
                except TypeError as e:
                    print(f"Cannot serialize response as JSON: {e}")
                    print(f"Response type: {type(read_result)}")
                    print(f"Response: {str(read_result)}")

            if not read_result["success"]:
                print(f"❌ Failed to read file {remote_path}: {read_result.get('error', 'Unknown error')}")
                return False

            # Extract file content from results
            for result in read_result["results"]:
                if debug:
                    print(f"🔍 Debug - Processing result...")

                if "content" in result:
                    for content_item in result["content"]:
                        if debug:
                            print(f"🔍 Debug - Content item type: {content_item.get('type')}")

                        # Handle the actual response structure: type = "resource" with nested resource object
                        if content_item.get("type") == "resource":
                            resource = content_item.get("resource", {})
                            mime_type = resource.get("mimeType", "")

                            if debug:
                                print(f"🔍 Debug - Resource MIME type: {mime_type}")
                                print(f"🔍 Debug - Resource keys: {list(resource.keys())}")

                            # Create local directory if it doesn't exist
                            os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True)

                            # Handle text files
                            if "text" in resource:
                                with open(local_path, 'w', encoding='utf-8') as f:
                                    f.write(resource["text"])
                                print(f"✅ Downloaded text file {remote_path} to {local_path}")
                                return True

                            # Handle binary files (like images) - blob format
                            elif "blob" in resource:
                                # Blob data is already in bytes format
                                with open(local_path, 'wb') as f:
                                    f.write(resource["blob"])
                                print(f"✅ Downloaded binary file {remote_path} to {local_path}")
                                return True

                            # Handle binary files (like images) - data format
                            elif "data" in resource:
                                # Data is already in bytes format
                                with open(local_path, 'wb') as f:
                                    f.write(resource["data"])
                                print(f"✅ Downloaded binary file {remote_path} to {local_path}")
                                return True

                            # Handle base64 encoded data
                            elif "base64" in resource:
                                import base64
                                with open(local_path, 'wb') as f:
                                    f.write(base64.b64decode(resource["base64"]))
                                print(f"✅ Downloaded base64 file {remote_path} to {local_path}")
                                return True

                        # Handle legacy content types for backward compatibility
                        elif content_item.get("type") == "text":
                            text_content = content_item.get("text", "")
                            if text_content.strip():
                                os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True)
                                with open(local_path, 'w', encoding='utf-8') as f:
                                    f.write(text_content)
                                print(f"✅ Downloaded {remote_path} to {local_path}")
                                return True

                        elif content_item.get("type") == "file":
                            os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True)
                            if "data" in content_item:
                                import base64
                                with open(local_path, 'wb') as f:
                                    f.write(base64.b64decode(content_item["data"]))
                            elif "text" in content_item:
                                with open(local_path, 'w', encoding='utf-8') as f:
                                    f.write(content_item["text"])
                            print(f"✅ Downloaded {remote_path} to {local_path}")
                            return True

            print(f"❌ No file content found for {remote_path}")
            return False

        except Exception as e:
            print(f"❌ Error downloading file {remote_path}: {e}")
            import traceback
            if debug:
                traceback.print_exc()
            return False

    def download_all_files(self, download_dir: str = "./downloads") -> List[str]:
        """Download all files from the interpreter sandbox"""
        downloaded_files = []

        # List all files in sandbox
        list_result = self.list_files()
        if not list_result["success"]:
            print(f"❌ Failed to list files: {list_result.get('error', 'Unknown error')}")
            return downloaded_files

        # Extract file names from results based on the actual response structure
        files_to_download = []
        for result in list_result["results"]:
            if "content" in result:
                for content_item in result["content"]:
                    # Handle the new response format with resource_link
                    if content_item.get("type") == "resource_link":
                        name = content_item.get("name", "")
                        description = content_item.get("description", "")
                        mime_type = content_item.get("mimeType", "")

                        # Only download files (not directories) and skip hidden files
                        if description == "File" and not name.startswith('.'):
                            files_to_download.append(name)

                    # Handle legacy text-based file listing
                    elif content_item.get("type") == "text":
                        # Parse file listing output
                        lines = content_item["text"].strip().split('\n')
                        for line in lines:
                            if line.strip() and not line.startswith('total'):
                                # Extract filename from ls output
                                parts = line.split()
                                if len(parts) >= 1:
                                    filename = parts[-1]
                                    if not filename.startswith('.'):  # Skip hidden files
                                        files_to_download.append(filename)

        # Download each file
        print(f"📥 Downloading {len(files_to_download)} files: {files_to_download}")
        for filename in files_to_download:
            local_path = os.path.join(download_dir, filename)
            if self.download_file(filename, local_path):
                downloaded_files.append(local_path)

        return downloaded_files

    def execute_command(self, command: str) -> Dict[str, Any]:
        """Execute shell command in the interpreter"""
        try:
            response = self.dp_client.invoke_code_interpreter(
                codeInterpreterIdentifier=self.interpreter_id,
                sessionId=self.session_id,
                #
                name="executeCommand",
                arguments={"command": command}
            )

            results = []
            for event in response['stream']:
                if 'result' in event:
                    results.append(event['result'])

            return {"success": True, "results": results}

        except Exception as e:
            return {"success": False, "error": str(e)}

    def demo_basic_python(self):
        """Demonstrate basic Python code execution"""
        print("\n🐍 === Basic Python Execution Demo ===")

        # Simple calculation
        result = self.execute_code('print("Hello from AgentCore Code Interpreter!")')
        self.print_result("Hello World", result)

        # Mathematical operations
        result = self.execute_code("""
import math

# Basic calculations
result = 15 * 23 + 7
print(f"15 * 23 + 7 = {result}")

# Advanced math
pi_approx = math.pi
print(f"π ≈ {pi_approx:.6f}")

# List comprehension
squares = [x**2 for x in range(1, 6)]
print(f"Squares 1-5: {squares}")
""")
        self.print_result("Mathematical Operations", result)

    def demo_data_analysis(self):
        """Demonstrate data analysis capabilities"""
        print("\n📊 === Data Analysis Demo ===")

        # Create sample data
        sample_data = {
            "path": "sales_data.csv",
            "text": "month,sales,profit\nJan,10000,2000\nFeb,12000,2400\nMar,15000,3000\nApr,18000,3600\nMay,20000,4000\nJun,22000,4400"
        }

        # Write data file
        write_result = self.write_files([sample_data])
        print("📄 Sample data file created")

        # Analyze data
        analysis_code = """
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np

# Load and analyze data
df = pd.read_csv('sales_data.csv')
print("📈 Sales Data Analysis")
print("=" * 40)
print(f"Total months: {len(df)}")
print(f"Average sales: ${df['sales'].mean():,.2f}")
print(f"Total profit: ${df['profit'].sum():,.2f}")
print(f"Profit margin: {(df['profit'].sum() / df['sales'].sum() * 100):.1f}%")

# Growth analysis
df['sales_growth'] = df['sales'].pct_change() * 100
print(f"\\nAverage monthly growth: {df['sales_growth'].mean():.1f}%")

# Create visualization
plt.figure(figsize=(10, 6))
plt.subplot(1, 2, 1)
plt.plot(df['month'], df['sales'], marker='o', linewidth=2, markersize=8)
plt.title('Monthly Sales Trend')
plt.xlabel('Month')
plt.ylabel('Sales ($)')
plt.xticks(rotation=45)
plt.grid(True, alpha=0.3)

plt.subplot(1, 2, 2)
plt.bar(df['month'], df['profit'], color='green', alpha=0.7)
plt.title('Monthly Profit')
plt.xlabel('Month')
plt.ylabel('Profit ($)')
plt.xticks(rotation=45)
plt.grid(True, alpha=0.3)

plt.tight_layout()
plt.savefig('sales_analysis.png', dpi=150, bbox_inches='tight')
print("\\n📊 Chart saved as 'sales_analysis.png'")

# Statistical summary
print("\\n📋 Statistical Summary:")
print(df.describe())
"""

        result = self.execute_code(analysis_code)
        self.print_result("Data Analysis", result)

        # Verify files exist after creating the plot
        print("\n🔍 === Verifying Files After Plot Creation ===")
        verify_result = self.execute_code("import os; print('Files in current directory:'); print(os.listdir('.'))")
        self.print_result("File Verification", verify_result)

        # Download the generated chart to local environment
        print("\n📥 === Downloading Generated Files ===")
        if self.download_file("sales_analysis.png", "./downloads/sales_analysis.png"):
            print("✅ Chart downloaded successfully!")
        else:
            print("❌ Failed to download chart")

    def demo_file_operations(self):
        """Demonstrate file operations"""
        print("\n📁 === File Operations Demo ===")

        # Create multiple files
        files_to_create = [
            {
                "path": "config.json",
                "text": json.dumps({
                    "app_name": "AgentCore Demo",
                    "version": "1.0.0",
                    "features": ["code_execution", "file_ops", "data_analysis"]
                }, indent=2)
            },
            {
                "path": "demo_script.py",
                "text": """
def fibonacci(n):
    \"\"\"Generate Fibonacci sequence up to n terms\"\"\"
    if n <= 0:
        return []
    elif n == 1:
        return [0]
    elif n == 2:
        return [0, 1]

    fib = [0, 1]
    for i in range(2, n):
        fib.append(fib[i-1] + fib[i-2])
    return fib

if __name__ == "__main__":
    print("Fibonacci sequence (first 10 terms):")
    print(fibonacci(10))
"""
            }
        ]

        # Write files
        write_result = self.write_files(files_to_create)
        print("📄 Created demo files")

        # List files
        list_result = self.list_files()
        print("📋 Files in sandbox:")
        if list_result["success"]:
            for result in list_result["results"]:
                if "content" in result:
                    for item in result["content"]:
                        if item.get("type") == "resource_link" and item.get("description") == "File":
                            print(f"  {item['name']} ({item.get('mimeType', 'unknown')})")
                        elif item.get("type") == "text":
                            print(f"  {item['text']}")

        # Execute the demo script
        result = self.execute_code("exec(open('demo_script.py').read())")
        self.print_result("Script Execution", result)

        # Download config file as example
        print("\n📥 === Downloading Config File ===")
        if self.download_file("config.json", "./downloads/config.json"):
            print("✅ Config file downloaded successfully!")
        else:
            print("❌ Failed to download config file")

    def demo_shell_commands(self):
        """Demonstrate shell command execution"""
        print("\n🐚 === Shell Commands Demo ===")

        # System information
        result = self.execute_command("uname -a && python --version")
        self.print_result("System Info", result)

        # Install package
        result = self.execute_command("pip install requests")
        self.print_result("Package Installation", result)

        # Use installed package
        web_code = """
import requests
import json

try:
    # Get some public data
    response = requests.get('https://jsonplaceholder.typicode.com/posts/1')
    if response.status_code == 200:
        data = response.json()
        print("📡 Successfully fetched data from API:")
        print(f"Title: {data['title']}")
        print(f"Body: {data['body'][:100]}...")
    else:
        print(f"❌ API request failed: {response.status_code}")
except Exception as e:
    print(f"❌ Error: {e}")
"""

        result = self.execute_code(web_code)
        self.print_result("Web Request", result)

    def demo_machine_learning(self):
        """Demonstrate machine learning capabilities"""
        print("\n🤖 === Machine Learning Demo ===")

        # Install scikit-learn
        install_result = self.execute_command("pip install scikit-learn")

        ml_code = """
import numpy as np
from sklearn.linear_model import LinearRegression
from sklearn.metrics import r2_score
import matplotlib.pyplot as plt

# Generate sample data
np.random.seed(42)
X = np.random.randn(100, 1) * 10
y = 2 * X.flatten() + 3 + np.random.randn(100) * 5

# Create and train model
model = LinearRegression()
model.fit(X, y)

# Make predictions
y_pred = model.predict(X)

# Calculate metrics
r2 = r2_score(y, y_pred)
print(f"🎯 Model Performance:")
print(f"R² Score: {r2:.4f}")
print(f"Coefficient: {model.coef_[0]:.4f}")
print(f"Intercept: {model.intercept_:.4f}")

# Create visualization
plt.figure(figsize=(10, 6))
plt.scatter(X, y, alpha=0.6, label='Data points')
plt.plot(X, y_pred, color='red', linewidth=2, label='Regression line')
plt.xlabel('X')
plt.ylabel('y')
plt.title('Linear Regression Demo')
plt.legend()
plt.grid(True, alpha=0.3)
plt.savefig('ml_demo.png', dpi=150, bbox_inches='tight')
print("\\n📊 ML visualization saved as 'ml_demo.png'")
"""

        result = self.execute_code(ml_code)
        self.print_result("Machine Learning", result)

        # Download the ML visualization
        print("\n📥 === Downloading ML Visualization ===")
        if self.download_file("ml_demo.png", "./downloads/ml_demo.png"):
            print("✅ ML visualization downloaded successfully!")
        else:
            print("❌ Failed to download ML visualization")

    def print_result(self, title: str, result: Dict[str, Any]):
        """Pretty print execution results"""
        print(f"\n--- {title} ---")
        if result["success"]:
            for res in result["results"]:
                if "content" in res:
                    for item in res["content"]:
                        if item["type"] == "text":
                            print(item["text"])
        else:
            print(f"❌ Error: {result['error']}")

    def cleanup(self):
        """Clean up resources"""
        try:
            if self.session_id and self.interpreter_id:
                self.dp_client.stop_code_interpreter_session(
                    codeInterpreterIdentifier=self.interpreter_id,
                    sessionId=self.session_id
                )
                print("✅ Session stopped")
            # Such sequential order is required since deleteing an interpreter while a session is still active could cause errors or leave resources in an inconsistent state
            if self.interpreter_id:
                self.cp_client.delete_code_interpreter(
                    codeInterpreterId=self.interpreter_id
                )
                print("✅ Interpreter deleted")

        except Exception as e:
            print(f"⚠️ Cleanup warning: {e}")

def main():
    """Main demo function"""
    print("🚀 Amazon Bedrock AgentCore Code Interpreter Demo")
    print("=" * 60)

    # Check for required environment variable
    execution_role_arn = os.getenv("AGENTCORE_EXECUTION_ROLE_ARN")
    if not execution_role_arn:
        print("❌ Error: AGENTCORE_EXECUTION_ROLE_ARN environment variable not set")
        print("   Please set it to your AgentCore execution role ARN")
        print("   Example: export AGENTCORE_EXECUTION_ROLE_ARN='arn:aws:iam::123456789012:role/AgentCoreRole'")
        return

    # Initialize demo
    demo = CodeInterpreterDemo()

    try:
        # Set up interpreter
        print("\n🔧 Setting up Code Interpreter...")
        interpreter_id = demo.create_interpreter(execution_role_arn)
        if not interpreter_id:
            return

        # Start session
        session_id = demo.start_session()
        if not session_id:
            return

        print(f"\n✅ Code Interpreter ready!")
        print(f"   Interpreter ID: {interpreter_id}")
        print(f"   Session ID: {session_id}")

        # Run demonstrations
        demo.demo_basic_python()
        demo.demo_data_analysis()
        demo.demo_file_operations()
        demo.demo_shell_commands()
        demo.demo_machine_learning()

        print("\n🎉 All demonstrations completed successfully!")
        print("\nKey capabilities demonstrated:")
        print("✓ Basic Python code execution")
        print("✓ Data analysis with pandas and matplotlib")
        print("✓ File operations (create, read, list)")
        print("✓ Shell command execution")
        print("✓ Package installation")
        print("✓ Machine learning with scikit-learn")
        print("✓ Web requests and API calls")
        print("✓ Data visualization")
        print("✓ File download from sandbox to local environment")

        # Download all remaining files
        print("\n📥 === Downloading All Files from Sandbox ===")
        downloaded_files = demo.download_all_files("./downloads")

        # Show downloaded files
        downloads_dir = "./downloads"
        if os.path.exists(downloads_dir):
            all_files = os.listdir(downloads_dir)
            if all_files:
                print(f"\n📁 Downloaded files in {downloads_dir}:")
                for file in all_files:
                    file_path = os.path.join(downloads_dir, file)
                    file_size = os.path.getsize(file_path)
                    print(f"  • {file} ({file_size:,} bytes)")
        else:
            print(f"\n📁 No files downloaded to {downloads_dir}")

    except KeyboardInterrupt:
        print("\n\n⏹️ Demo interrupted by user")
    except Exception as e:
        print(f"\n❌ Demo error: {e}")
    finally:
        print("\n🧹 Cleaning up resources...")
        demo.cleanup()
        print("✅ Demo completed")

if __name__ == "__main__":
    main()
	#!/usr/bin/env python3
	"""
	Amazon Bedrock AgentCore Code Interpreter Demo

	A fully functional demonstration of AgentCore's Code Interpreter capabilities,
	showcasing secure code execution, file operations, and agent integration.
	"""

	import json
	import boto3
	import time
	import os
	import sys
	import asyncio
	from typing import Dict, Any, List, Optional
	from bedrock_agentcore.tools.code_interpreter_client import CodeInterpreter
	from bedrock_agentcore._utils import endpoints

	class CodeInterpreterDemo:
	"""
	Comprehensive demo of AgentCore Code Interpreter functionality
	"""

	def __init__(self, region: str = "us-west-2"):
	self.region = region
	self.code_client = None
	self.session_id = None
	self.interpreter_id = None
	self.setup_clients()

	def setup_clients(self):
	"""Initialize AWS clients for Code Interpreter"""
	try:
	# Set up endpoints
	data_plane_endpoint = endpoints.get_data_plane_endpoint(self.region)
	control_plane_endpoint = endpoints.get_control_plane_endpoint(self.region)

	# Create clients
	self.cp_client = boto3.client(
	"bedrock-agentcore-control",
	region_name=self.region,
	endpoint_url=control_plane_endpoint
	)

	self.dp_client = boto3.client(
	"bedrock-agentcore",
	region_name=self.region,
	endpoint_url=data_plane_endpoint
	)

	print("✅ AWS clients initialized successfully")

	except Exception as e:
	print(f"❌ Error initializing clients: {e}")
	sys.exit(1)

	def create_interpreter(self, execution_role_arn: str) -> str:
	"""Create a new Code Interpreter instance"""
	try:
	unique_name = f"demo_interpreter_{int(time.time())}"

	response = self.cp_client.create_code_interpreter(
	name=unique_name,
	description="Demo Code Interpreter for AgentCore tutorial",
	executionRoleArn=execution_role_arn,
	networkConfiguration={'networkMode': 'PUBLIC'}
	)

	self.interpreter_id = response["codeInterpreterId"]
	print(f"✅ Code Interpreter created: {self.interpreter_id}")
	return self.interpreter_id

	except Exception as e:
	print(f"❌ Error creating interpreter: {e}")
	return None

	def start_session(self, timeout_seconds: int = 1800) -> str:
	"""Start a new Code Interpreter session"""
	try:
	response = self.dp_client.start_code_interpreter_session(
	codeInterpreterIdentifier=self.interpreter_id,
	name="DemoSession",
	sessionTimeoutSeconds=timeout_seconds
	)

	self.session_id = response["sessionId"]
	print(f"✅ Session started: {self.session_id}")
	return self.session_id

	except Exception as e:
	print(f"❌ Error starting session: {e}")
	return None

	def execute_code(self, code: str, language: str = "python") -> Dict[str, Any]:
	"""Execute code in the interpreter"""
	try:
	response = self.dp_client.invoke_code_interpreter(
	codeInterpreterIdentifier=self.interpreter_id,
	sessionId=self.session_id,
	# Valid Values: executeCode \| executeCommand \| readFiles \| listFiles \| removeFiles \| writeFiles \| startCommandExecution \| getTask \| stopTask
	name="executeCode",
	arguments={
	"language": language,
	"code": code
	}
	)

	# Process response stream
	results = []
	for event in response['stream']:
	if 'result' in event:
	results.append(event['result'])

	return {"success": True, "results": results}

	except Exception as e:
	return {"success": False, "error": str(e)}

	def write_files(self, files: List[Dict[str, str]]) -> Dict[str, Any]:
	"""Write files to the interpreter sandbox"""
	try:
	response = self.dp_client.invoke_code_interpreter(
	codeInterpreterIdentifier=self.interpreter_id,
	sessionId=self.session_id,
	# Valid Values: executeCode \| executeCommand \| readFiles \| listFiles \| removeFiles \| writeFiles \| startCommandExecution \| getTask \| stopTask
	name="writeFiles",
	arguments={"content": files}
	)

	results = []
	for event in response['stream']:
	if 'result' in event:
	results.append(event['result'])

	return {"success": True, "results": results}

	except Exception as e:
	return {"success": False, "error": str(e)}

	def list_files(self, directory_path: str = "") -> Dict[str, Any]:
	"""List files in the interpreter sandbox"""
	try:
	response = self.dp_client.invoke_code_interpreter(
	codeInterpreterIdentifier=self.interpreter_id,
	sessionId=self.session_id,
	name="listFiles",
	arguments={"directoryPath": directory_path}
	)

	results = []
	for event in response['stream']:
	if 'result' in event:
	results.append(event['result'])

	return {"success": True, "results": results}

	except Exception as e:
	return {"success": False, "error": str(e)}

	def read_files(self, file_paths: List[str]) -> Dict[str, Any]:
	"""Read files from the interpreter sandbox"""
	try:
	response = self.dp_client.invoke_code_interpreter(
	codeInterpreterIdentifier=self.interpreter_id,
	sessionId=self.session_id,
	name="readFiles",
	arguments={"paths": file_paths}
	)

	results = []
	for event in response['stream']:
	if 'result' in event:
	results.append(event['result'])

	return {"success": True, "results": results}

	except Exception as e:
	return {"success": False, "error": str(e)}

	def download_file(self, remote_path: str, local_path: str, debug: bool = False) -> bool:
	"""Download a file from the interpreter sandbox to local environment"""
	try:
	# Read the file from sandbox
	read_result = self.read_files([remote_path])

	if debug:
	print(f"🔍 Debug - Read result for {remote_path}:")
	# Handle bytes in response by converting to string representation
	try:
	print(json.dumps(read_result, indent=2))
	except TypeError as e:
	print(f"Cannot serialize response as JSON: {e}")
	print(f"Response type: {type(read_result)}")
	print(f"Response: {str(read_result)}")

	if not read_result["success"]:
	print(f"❌ Failed to read file {remote_path}: {read_result.get('error', 'Unknown error')}")
	return False

	# Extract file content from results
	for result in read_result["results"]:
	if debug:
	print(f"🔍 Debug - Processing result...")

	if "content" in result:
	for content_item in result["content"]:
	if debug:
	print(f"🔍 Debug - Content item type: {content_item.get('type')}")

	# Handle the actual response structure: type = "resource" with nested resource object
	if content_item.get("type") == "resource":
	resource = content_item.get("resource", {})
	mime_type = resource.get("mimeType", "")

	if debug:
	print(f"🔍 Debug - Resource MIME type: {mime_type}")
	print(f"🔍 Debug - Resource keys: {list(resource.keys())}")

	# Create local directory if it doesn't exist
	os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True)

	# Handle text files
	if "text" in resource:
	with open(local_path, 'w', encoding='utf-8') as f:
	f.write(resource["text"])
	print(f"✅ Downloaded text file {remote_path} to {local_path}")
	return True

	# Handle binary files (like images) - blob format
	elif "blob" in resource:
	# Blob data is already in bytes format
	with open(local_path, 'wb') as f:
	f.write(resource["blob"])
	print(f"✅ Downloaded binary file {remote_path} to {local_path}")
	return True

	# Handle binary files (like images) - data format
	elif "data" in resource:
	# Data is already in bytes format
	with open(local_path, 'wb') as f:
	f.write(resource["data"])
	print(f"✅ Downloaded binary file {remote_path} to {local_path}")
	return True

	# Handle base64 encoded data
	elif "base64" in resource:
	import base64
	with open(local_path, 'wb') as f:
	f.write(base64.b64decode(resource["base64"]))
	print(f"✅ Downloaded base64 file {remote_path} to {local_path}")
	return True

	# Handle legacy content types for backward compatibility
	elif content_item.get("type") == "text":
	text_content = content_item.get("text", "")
	if text_content.strip():
	os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True)
	with open(local_path, 'w', encoding='utf-8') as f:
	f.write(text_content)
	print(f"✅ Downloaded {remote_path} to {local_path}")
	return True

	elif content_item.get("type") == "file":
	os.makedirs(os.path.dirname(local_path) if os.path.dirname(local_path) else '.', exist_ok=True)
	if "data" in content_item:
	import base64
	with open(local_path, 'wb') as f:
	f.write(base64.b64decode(content_item["data"]))
	elif "text" in content_item:
	with open(local_path, 'w', encoding='utf-8') as f:
	f.write(content_item["text"])
	print(f"✅ Downloaded {remote_path} to {local_path}")
	return True

	print(f"❌ No file content found for {remote_path}")
	return False

	except Exception as e:
	print(f"❌ Error downloading file {remote_path}: {e}")
	import traceback
	if debug:
	traceback.print_exc()
	return False

	def download_all_files(self, download_dir: str = "./downloads") -> List[str]:
	"""Download all files from the interpreter sandbox"""
	downloaded_files = []

	# List all files in sandbox
	list_result = self.list_files()
	if not list_result["success"]:
	print(f"❌ Failed to list files: {list_result.get('error', 'Unknown error')}")
	return downloaded_files

	# Extract file names from results based on the actual response structure
	files_to_download = []
	for result in list_result["results"]:
	if "content" in result:
	for content_item in result["content"]:
	# Handle the new response format with resource_link
	if content_item.get("type") == "resource_link":
	name = content_item.get("name", "")
	description = content_item.get("description", "")
	mime_type = content_item.get("mimeType", "")

	# Only download files (not directories) and skip hidden files
	if description == "File" and not name.startswith('.'):
	files_to_download.append(name)

	# Handle legacy text-based file listing
	elif content_item.get("type") == "text":
	# Parse file listing output
	lines = content_item["text"].strip().split('\n')
	for line in lines:
	if line.strip() and not line.startswith('total'):
	# Extract filename from ls output
	parts = line.split()
	if len(parts) >= 1:
	filename = parts[-1]
	if not filename.startswith('.'): # Skip hidden files
	files_to_download.append(filename)

	# Download each file
	print(f"📥 Downloading {len(files_to_download)} files: {files_to_download}")
	for filename in files_to_download:
	local_path = os.path.join(download_dir, filename)
	if self.download_file(filename, local_path):
	downloaded_files.append(local_path)

	return downloaded_files

	def execute_command(self, command: str) -> Dict[str, Any]:
	"""Execute shell command in the interpreter"""
	try:
	response = self.dp_client.invoke_code_interpreter(
	codeInterpreterIdentifier=self.interpreter_id,
	sessionId=self.session_id,
	#
	name="executeCommand",
	arguments={"command": command}
	)

	results = []
	for event in response['stream']:
	if 'result' in event:
	results.append(event['result'])

	return {"success": True, "results": results}

	except Exception as e:
	return {"success": False, "error": str(e)}

	def demo_basic_python(self):
	"""Demonstrate basic Python code execution"""
	print("\n🐍 === Basic Python Execution Demo ===")

	# Simple calculation
	result = self.execute_code('print("Hello from AgentCore Code Interpreter!")')
	self.print_result("Hello World", result)

	# Mathematical operations
	result = self.execute_code("""
	import math

	# Basic calculations
	result = 15 * 23 + 7
	print(f"15 * 23 + 7 = {result}")

	# Advanced math
	pi_approx = math.pi
	print(f"π ≈ {pi_approx:.6f}")

	# List comprehension
	squares = [x**2 for x in range(1, 6)]
	print(f"Squares 1-5: {squares}")
	""")
	self.print_result("Mathematical Operations", result)

	def demo_data_analysis(self):
	"""Demonstrate data analysis capabilities"""
	print("\n📊 === Data Analysis Demo ===")

	# Create sample data
	sample_data = {
	"path": "sales_data.csv",
	"text": "month,sales,profit\nJan,10000,2000\nFeb,12000,2400\nMar,15000,3000\nApr,18000,3600\nMay,20000,4000\nJun,22000,4400"
	}

	# Write data file
	write_result = self.write_files([sample_data])
	print("📄 Sample data file created")

	# Analyze data
	analysis_code = """
	import pandas as pd
	import matplotlib.pyplot as plt
	import numpy as np

	# Load and analyze data
	df = pd.read_csv('sales_data.csv')
	print("📈 Sales Data Analysis")
	print("=" * 40)
	print(f"Total months: {len(df)}")
	print(f"Average sales: ${df['sales'].mean():,.2f}")
	print(f"Total profit: ${df['profit'].sum():,.2f}")
	print(f"Profit margin: {(df['profit'].sum() / df['sales'].sum() * 100):.1f}%")

	# Growth analysis
	df['sales_growth'] = df['sales'].pct_change() * 100
	print(f"\\nAverage monthly growth: {df['sales_growth'].mean():.1f}%")

	# Create visualization
	plt.figure(figsize=(10, 6))
	plt.subplot(1, 2, 1)
	plt.plot(df['month'], df['sales'], marker='o', linewidth=2, markersize=8)
	plt.title('Monthly Sales Trend')
	plt.xlabel('Month')
	plt.ylabel('Sales ($)')
	plt.xticks(rotation=45)
	plt.grid(True, alpha=0.3)

	plt.subplot(1, 2, 2)
	plt.bar(df['month'], df['profit'], color='green', alpha=0.7)
	plt.title('Monthly Profit')
	plt.xlabel('Month')
	plt.ylabel('Profit ($)')
	plt.xticks(rotation=45)
	plt.grid(True, alpha=0.3)

	plt.tight_layout()
	plt.savefig('sales_analysis.png', dpi=150, bbox_inches='tight')
	print("\\n📊 Chart saved as 'sales_analysis.png'")

	# Statistical summary
	print("\\n📋 Statistical Summary:")
	print(df.describe())
	"""

	result = self.execute_code(analysis_code)
	self.print_result("Data Analysis", result)

	# Verify files exist after creating the plot
	print("\n🔍 === Verifying Files After Plot Creation ===")
	verify_result = self.execute_code("import os; print('Files in current directory:'); print(os.listdir('.'))")
	self.print_result("File Verification", verify_result)

	# Download the generated chart to local environment
	print("\n📥 === Downloading Generated Files ===")
	if self.download_file("sales_analysis.png", "./downloads/sales_analysis.png"):
	print("✅ Chart downloaded successfully!")
	else:
	print("❌ Failed to download chart")

	def demo_file_operations(self):
	"""Demonstrate file operations"""
	print("\n📁 === File Operations Demo ===")

	# Create multiple files
	files_to_create = [
	{
	"path": "config.json",
	"text": json.dumps({
	"app_name": "AgentCore Demo",
	"version": "1.0.0",
	"features": ["code_execution", "file_ops", "data_analysis"]
	}, indent=2)
	},
	{
	"path": "demo_script.py",
	"text": """
	def fibonacci(n):
	\"\"\"Generate Fibonacci sequence up to n terms\"\"\"
	if n <= 0:
	return []
	elif n == 1:
	return [0]
	elif n == 2:
	return [0, 1]

	fib = [0, 1]
	for i in range(2, n):
	fib.append(fib[i-1] + fib[i-2])
	return fib

	if __name__ == "__main__":
	print("Fibonacci sequence (first 10 terms):")
	print(fibonacci(10))
	"""
	}
	]

	# Write files
	write_result = self.write_files(files_to_create)
	print("📄 Created demo files")

	# List files
	list_result = self.list_files()
	print("📋 Files in sandbox:")
	if list_result["success"]:
	for result in list_result["results"]:
	if "content" in result:
	for item in result["content"]:
	if item.get("type") == "resource_link" and item.get("description") == "File":
	print(f" {item['name']} ({item.get('mimeType', 'unknown')})")
	elif item.get("type") == "text":
	print(f" {item['text']}")

	# Execute the demo script
	result = self.execute_code("exec(open('demo_script.py').read())")
	self.print_result("Script Execution", result)

	# Download config file as example
	print("\n📥 === Downloading Config File ===")
	if self.download_file("config.json", "./downloads/config.json"):
	print("✅ Config file downloaded successfully!")
	else:
	print("❌ Failed to download config file")

	def demo_shell_commands(self):
	"""Demonstrate shell command execution"""
	print("\n🐚 === Shell Commands Demo ===")

	# System information
	result = self.execute_command("uname -a && python --version")
	self.print_result("System Info", result)

	# Install package
	result = self.execute_command("pip install requests")
	self.print_result("Package Installation", result)

	# Use installed package
	web_code = """
	import requests
	import json

	try:
	# Get some public data
	response = requests.get('https://jsonplaceholder.typicode.com/posts/1')
	if response.status_code == 200:
	data = response.json()
	print("📡 Successfully fetched data from API:")
	print(f"Title: {data['title']}")
	print(f"Body: {data['body'][:100]}...")
	else:
	print(f"❌ API request failed: {response.status_code}")
	except Exception as e:
	print(f"❌ Error: {e}")
	"""

	result = self.execute_code(web_code)
	self.print_result("Web Request", result)

	def demo_machine_learning(self):
	"""Demonstrate machine learning capabilities"""
	print("\n🤖 === Machine Learning Demo ===")

	# Install scikit-learn
	install_result = self.execute_command("pip install scikit-learn")

	ml_code = """
	import numpy as np
	from sklearn.linear_model import LinearRegression
	from sklearn.metrics import r2_score
	import matplotlib.pyplot as plt

	# Generate sample data
	np.random.seed(42)
	X = np.random.randn(100, 1) * 10
	y = 2 * X.flatten() + 3 + np.random.randn(100) * 5

	# Create and train model
	model = LinearRegression()
	model.fit(X, y)

	# Make predictions
	y_pred = model.predict(X)

	# Calculate metrics
	r2 = r2_score(y, y_pred)
	print(f"🎯 Model Performance:")
	print(f"R² Score: {r2:.4f}")
	print(f"Coefficient: {model.coef_[0]:.4f}")
	print(f"Intercept: {model.intercept_:.4f}")

	# Create visualization
	plt.figure(figsize=(10, 6))
	plt.scatter(X, y, alpha=0.6, label='Data points')
	plt.plot(X, y_pred, color='red', linewidth=2, label='Regression line')
	plt.xlabel('X')
	plt.ylabel('y')
	plt.title('Linear Regression Demo')
	plt.legend()
	plt.grid(True, alpha=0.3)
	plt.savefig('ml_demo.png', dpi=150, bbox_inches='tight')
	print("\\n📊 ML visualization saved as 'ml_demo.png'")
	"""

	result = self.execute_code(ml_code)
	self.print_result("Machine Learning", result)

	# Download the ML visualization
	print("\n📥 === Downloading ML Visualization ===")
	if self.download_file("ml_demo.png", "./downloads/ml_demo.png"):
	print("✅ ML visualization downloaded successfully!")
	else:
	print("❌ Failed to download ML visualization")

	def print_result(self, title: str, result: Dict[str, Any]):
	"""Pretty print execution results"""
	print(f"\n--- {title} ---")
	if result["success"]:
	for res in result["results"]:
	if "content" in res:
	for item in res["content"]:
	if item["type"] == "text":
	print(item["text"])
	else:
	print(f"❌ Error: {result['error']}")

	def cleanup(self):
	"""Clean up resources"""
	try:
	if self.session_id and self.interpreter_id:
	self.dp_client.stop_code_interpreter_session(
	codeInterpreterIdentifier=self.interpreter_id,
	sessionId=self.session_id
	)
	print("✅ Session stopped")
	# Such sequential order is required since deleteing an interpreter while a session is still active could cause errors or leave resources in an inconsistent state
	if self.interpreter_id:
	self.cp_client.delete_code_interpreter(
	codeInterpreterId=self.interpreter_id
	)
	print("✅ Interpreter deleted")

	except Exception as e:
	print(f"⚠️ Cleanup warning: {e}")

	def main():
	"""Main demo function"""
	print("🚀 Amazon Bedrock AgentCore Code Interpreter Demo")
	print("=" * 60)

	# Check for required environment variable
	execution_role_arn = os.getenv("AGENTCORE_EXECUTION_ROLE_ARN")
	if not execution_role_arn:
	print("❌ Error: AGENTCORE_EXECUTION_ROLE_ARN environment variable not set")
	print(" Please set it to your AgentCore execution role ARN")
	print(" Example: export AGENTCORE_EXECUTION_ROLE_ARN='arn:aws:iam::123456789012:role/AgentCoreRole'")
	return

	# Initialize demo
	demo = CodeInterpreterDemo()

	try:
	# Set up interpreter
	print("\n🔧 Setting up Code Interpreter...")
	interpreter_id = demo.create_interpreter(execution_role_arn)
	if not interpreter_id:
	return

	# Start session
	session_id = demo.start_session()
	if not session_id:
	return

	print(f"\n✅ Code Interpreter ready!")
	print(f" Interpreter ID: {interpreter_id}")
	print(f" Session ID: {session_id}")

	# Run demonstrations
	demo.demo_basic_python()
	demo.demo_data_analysis()
	demo.demo_file_operations()
	demo.demo_shell_commands()
	demo.demo_machine_learning()

	print("\n🎉 All demonstrations completed successfully!")
	print("\nKey capabilities demonstrated:")
	print("✓ Basic Python code execution")
	print("✓ Data analysis with pandas and matplotlib")
	print("✓ File operations (create, read, list)")
	print("✓ Shell command execution")
	print("✓ Package installation")
	print("✓ Machine learning with scikit-learn")
	print("✓ Web requests and API calls")
	print("✓ Data visualization")
	print("✓ File download from sandbox to local environment")

	# Download all remaining files
	print("\n📥 === Downloading All Files from Sandbox ===")
	downloaded_files = demo.download_all_files("./downloads")

	# Show downloaded files
	downloads_dir = "./downloads"
	if os.path.exists(downloads_dir):
	all_files = os.listdir(downloads_dir)
	if all_files:
	print(f"\n📁 Downloaded files in {downloads_dir}:")
	for file in all_files:
	file_path = os.path.join(downloads_dir, file)
	file_size = os.path.getsize(file_path)
	print(f" • {file} ({file_size:,} bytes)")
	else:
	print(f"\n📁 No files downloaded to {downloads_dir}")

	except KeyboardInterrupt:
	print("\n\n⏹️ Demo interrupted by user")
	except Exception as e:
	print(f"\n❌ Demo error: {e}")
	finally:
	print("\n🧹 Cleaning up resources...")
	demo.cleanup()
	print("✅ Demo completed")

	if __name__ == "__main__":
	main()
No results found