From 8ae138d959bb631dc2c2c12112b104bfbc1b1587 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Tue, 14 Oct 2025 18:53:56 -0700 Subject: [PATCH 1/6] feat: Add automatic session management to Code Interpreter tool - Make session_name parameter optional in all action models - Implement auto-session mode with _ensure_session() method - Enable automatic session creation when sessions don't exist - Update tool description to prioritize common use cases - Improve developer experience by reducing manual session management --- .../code_interpreter/__init__.py | 7 +- .../agent_core_code_interpreter.py | 195 ++++----- .../code_interpreter/code_interpreter.py | 372 +++++++----------- src/strands_tools/code_interpreter/models.py | 106 ++--- .../test_agent_core_code_interpreter.py | 92 +++++ .../code_interpreter/test_code_interpreter.py | 5 +- tests/test_editor.py | 28 +- .../test_agent_core_code_interpreter.py | 34 +- ...core_code_interpreter_custom_identifier.py | 111 +++++- 9 files changed, 539 insertions(+), 411 deletions(-) diff --git a/src/strands_tools/code_interpreter/__init__.py b/src/strands_tools/code_interpreter/__init__.py index 0591edde..7302f528 100644 --- a/src/strands_tools/code_interpreter/__init__.py +++ b/src/strands_tools/code_interpreter/__init__.py @@ -31,9 +31,4 @@ from .agent_core_code_interpreter import AgentCoreCodeInterpreter from .code_interpreter import CodeInterpreter -__all__ = [ - # Base classes - "CodeInterpreter", - # Platform implementations - "AgentCoreCodeInterpreter", -] +__all__ = ["CodeInterpreter", "AgentCoreCodeInterpreter"] diff --git a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py index 19d9cbe2..d3d14654 100644 --- a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py +++ b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py @@ -47,81 +47,51 @@ class AgentCoreCodeInterpreter(CodeInterpreter): This class provides a code interpreter interface using AWS Bedrock AgentCore services. It supports executing Python, JavaScript, and TypeScript code in isolated sandbox - environments with custom code interpreter identifiers. + environments with automatic session management. The class maintains session state and provides methods for code execution, file operations, and session management. It supports both default AWS code interpreter environments and custom environments specified by identifier. - Examples: - Basic usage with default identifier: - - >>> interpreter = AgentCoreCodeInterpreter(region="us-west-2") - >>> # Uses default identifier: "aws.codeinterpreter.v1" - - Using a custom code interpreter identifier: - - >>> custom_id = "my-custom-interpreter-abc123" - >>> interpreter = AgentCoreCodeInterpreter( - ... region="us-west-2", - ... identifier=custom_id - ... ) - - Environment-specific usage: - - >>> # For testing environments - >>> test_interpreter = AgentCoreCodeInterpreter( - ... region="us-east-1", - ... identifier="test-interpreter-xyz789" - ... ) - - >>> # For production environments - >>> prod_interpreter = AgentCoreCodeInterpreter( - ... region="us-west-2", - ... identifier="prod-interpreter-def456" - ... ) + Args: + region (Optional[str]): AWS region for the sandbox service. If not provided, + the region will be resolved from AWS configuration. + identifier (Optional[str]): Custom code interpreter identifier to use + for code execution sessions. If not provided, defaults to the AWS-managed + identifier "aws.codeinterpreter.v1". + auto_session (bool): Enable automatic session creation when a session doesn't exist. + Defaults to True. + default_session (str): Default session name to use when session_name is not specified + in operations. Defaults to "default". Attributes: region (str): The AWS region where the code interpreter service is hosted. identifier (str): The code interpreter identifier being used for sessions. + auto_session (bool): Whether automatic session creation is enabled. + default_session (str): The default session name. """ - def __init__(self, region: Optional[str] = None, identifier: Optional[str] = None) -> None: + def __init__( + self, + region: Optional[str] = None, + identifier: Optional[str] = None, + auto_session: bool = True, + default_session: str = "default", + ) -> None: """ Initialize the Bedrock AgentCore code interpreter. Args: - region (Optional[str]): AWS region for the sandbox service. If not provided, - the region will be resolved from AWS configuration (environment variables, - AWS config files, or instance metadata). Defaults to None. - identifier (Optional[str]): Custom code interpreter identifier to use - for code execution sessions. This allows you to specify custom code - interpreter environments instead of the default AWS-provided one. - - Valid formats include: - - Default identifier: "aws.codeinterpreter.v1" (used when None) - - Custom identifier: "my-custom-interpreter-abc123" - - Environment-specific: "test-interpreter-xyz789" - - Note: Use the code interpreter ID, not the full ARN. The AWS service - expects the identifier portion only (e.g., "my-interpreter-123" rather - than "arn:aws:bedrock-agentcore:region:account:code-interpreter-custom/my-interpreter-123"). - - If not provided, defaults to "aws.codeinterpreter.v1" for backward - compatibility. Defaults to None. - - Note: - This constructor maintains full backward compatibility. Existing code - that doesn't specify the identifier parameter will continue to work - unchanged with the default AWS code interpreter environment. - - Raises: - Exception: If there are issues with AWS region resolution or client - initialization during session creation. + region (Optional[str]): AWS region for the sandbox service. + identifier (Optional[str]): Custom code interpreter identifier. + auto_session (bool): Enable automatic session creation. Defaults to True. + default_session (str): Default session name. Defaults to "default". """ super().__init__() self.region = resolve_region(region) self.identifier = identifier or "aws.codeinterpreter.v1" + self.auto_session = auto_session + self.default_session = default_session self._sessions: Dict[str, SessionInfo] = {} def start_platform(self) -> None: @@ -240,84 +210,121 @@ def list_local_sessions(self) -> Dict[str, Any]: "content": [{"json": {"sessions": sessions_info, "totalSessions": len(sessions_info)}}], } + def _ensure_session(self, session_name: Optional[str]) -> tuple[str, Optional[Dict[str, Any]]]: + """ + Ensure a session exists, creating it automatically if needed. + + This method checks if the specified session exists. If auto_session is enabled + and the session doesn't exist, it will be created automatically. + + Args: + session_name (Optional[str]): The session name to ensure exists. If None, + uses the default_session. + + Returns: + tuple[str, Optional[Dict[str, Any]]]: A tuple containing: + - The session name (either provided or default) + - An error dictionary if session creation failed, None otherwise + """ + if not session_name: + session_name = self.default_session + + if session_name in self._sessions: + return session_name, None + + if self.auto_session: + logger.info(f"Auto-creating session: {session_name}") + init_action = InitSessionAction( + type="initSession", + session_name=session_name, + description=f"Auto-initialized session for {session_name}", + ) + result = self.init_session(init_action) + + if result.get("status") != "success": + return session_name, result + + logger.info(f"Successfully auto-created session: {session_name}") + return session_name, None + + return session_name, {"status": "error", "content": [{"text": f"Session '{session_name}' not found"}]} + def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: - """Execute code in a Bedrock AgentCoresession.""" - if action.session_name not in self._sessions: - return {"status": "error", "content": [{"text": f"Session '{action.session_name}' not found"}]} + """Execute code in a Bedrock AgentCore session with automatic session management.""" + session_name, error = self._ensure_session(action.session_name) + if error: + return error - logger.debug(f"Executing {action.language} code in session '{action.session_name}'") + logger.debug(f"Executing {action.language} code in session '{session_name}'") - # Use the invoke method with proper parameters as shown in the example params = {"code": action.code, "language": action.language.value, "clearContext": action.clear_context} - response = self._sessions[action.session_name].client.invoke("executeCode", params) + response = self._sessions[session_name].client.invoke("executeCode", params) return self._create_tool_result(response) def execute_command(self, action: ExecuteCommandAction) -> Dict[str, Any]: - """Execute a command in a Bedrock AgentCoresession.""" - if action.session_name not in self._sessions: - return {"status": "error", "content": [{"text": f"Session '{action.session_name}' not found"}]} + """Execute a command in a Bedrock AgentCore session with automatic session management.""" + session_name, error = self._ensure_session(action.session_name) + if error: + return error - logger.debug(f"Executing command in session '{action.session_name}': {action.command}") + logger.debug(f"Executing command in session '{session_name}': {action.command}") - # Use the invoke method with proper parameters as shown in the example params = {"command": action.command} - response = self._sessions[action.session_name].client.invoke("executeCommand", params) + response = self._sessions[session_name].client.invoke("executeCommand", params) return self._create_tool_result(response) def read_files(self, action: ReadFilesAction) -> Dict[str, Any]: - """Read files from a Bedrock AgentCoresession.""" - if action.session_name not in self._sessions: - return {"status": "error", "content": [{"text": f"Session '{action.session_name}' not found"}]} + """Read files from a Bedrock AgentCore session with automatic session management.""" + session_name, error = self._ensure_session(action.session_name) + if error: + return error - logger.debug(f"Reading files from session '{action.session_name}': {action.paths}") + logger.debug(f"Reading files from session '{session_name}': {action.paths}") - # Use the invoke method with proper parameters as shown in the example params = {"paths": action.paths} - response = self._sessions[action.session_name].client.invoke("readFiles", params) + response = self._sessions[session_name].client.invoke("readFiles", params) return self._create_tool_result(response) def list_files(self, action: ListFilesAction) -> Dict[str, Any]: - """List files in a Bedrock AgentCoresession directory.""" - if action.session_name not in self._sessions: - return {"status": "error", "content": [{"text": f"Session '{action.session_name}' not found"}]} + """List files in a Bedrock AgentCore session directory with automatic session management.""" + session_name, error = self._ensure_session(action.session_name) + if error: + return error - logger.debug(f"Listing files in session '{action.session_name}' at path: {action.path}") + logger.debug(f"Listing files in session '{session_name}' at path: {action.path}") - # Use the invoke method with proper parameters as shown in the example params = {"path": action.path} - response = self._sessions[action.session_name].client.invoke("listFiles", params) + response = self._sessions[session_name].client.invoke("listFiles", params) return self._create_tool_result(response) def remove_files(self, action: RemoveFilesAction) -> Dict[str, Any]: - """Remove files from a Bedrock AgentCoresession.""" - if action.session_name not in self._sessions: - return {"status": "error", "content": [{"text": f"Session '{action.session_name}' not found"}]} + """Remove files from a Bedrock AgentCore session with automatic session management.""" + session_name, error = self._ensure_session(action.session_name) + if error: + return error - logger.debug(f"Removing files from session '{action.session_name}': {action.paths}") + logger.debug(f"Removing files from session '{session_name}': {action.paths}") - # Use the invoke method with proper parameters as shown in the example params = {"paths": action.paths} - response = self._sessions[action.session_name].client.invoke("removeFiles", params) + response = self._sessions[session_name].client.invoke("removeFiles", params) return self._create_tool_result(response) def write_files(self, action: WriteFilesAction) -> Dict[str, Any]: - """Write files to a Bedrock AgentCoresession.""" - if action.session_name not in self._sessions: - return {"status": "error", "content": [{"text": f"Session '{action.session_name}' not found"}]} + """Write files to a Bedrock AgentCore session with automatic session management.""" + session_name, error = self._ensure_session(action.session_name) + if error: + return error - logger.debug(f"Writing {len(action.content)} files to session '{action.session_name}'") + logger.debug(f"Writing {len(action.content)} files to session '{session_name}'") - # Convert FileContent objects to dictionaries for the API content_dicts = [{"path": fc.path, "text": fc.text} for fc in action.content] - - # Use the invoke method with proper parameters as shown in the example params = {"content": content_dicts} - response = self._sessions[action.session_name].client.invoke("writeFiles", params) + response = self._sessions[session_name].client.invoke("writeFiles", params) return self._create_tool_result(response) diff --git a/src/strands_tools/code_interpreter/code_interpreter.py b/src/strands_tools/code_interpreter/code_interpreter.py index 7e29e65e..87cdf974 100644 --- a/src/strands_tools/code_interpreter/code_interpreter.py +++ b/src/strands_tools/code_interpreter/code_interpreter.py @@ -1,9 +1,4 @@ -""" -Code Interpreter Tool implementation using Strands @tool decorator. - -This module contains the base tool class that provides lifecycle management -and can be extended by specific platform implementations. -""" +"""Code Interpreter base class - IMPROVED VERSION""" import logging from abc import ABC, abstractmethod @@ -30,197 +25,151 @@ class CodeInterpreter(ABC): def __init__(self): self._started = False - # Dynamically override the ToolSpec description using the implementation-defined supported languages - self.code_interpreter.tool_spec["description"] = """ - Code Interpreter tool for executing code in isolated sandbox environments. - - This tool provides a comprehensive code execution platform that supports multiple programming - languages with persistent session management, file operations, and shell command execution. - Built on the Bedrock AgentCore Code Sandbox platform, it offers secure, isolated environments - for code execution with full lifecycle management. - - Key Features: - 1. Multi-Language Support: - The tool supports the following programming languages: {supported_languages_list} - • Full standard library access for each supported language - • Runtime environment appropriate for each language - • Shell command execution for system operations - - 2. Session Management: - • Create named, persistent sessions for stateful code execution - • List and manage multiple concurrent sessions - • Automatic session cleanup and resource management - • Session isolation for security and resource separation - - 3. File System Operations: - • Read files from the sandbox environment - • Write multiple files with custom content - • List directory contents and navigate file structures - • Remove files and manage sandbox storage - - 4. Advanced Execution Features: - • Context preservation across code executions within sessions - • Optional context clearing for fresh execution environments - • Real-time output capture and error handling - • Support for long-running processes and interactive code - - How It Works: - ------------ - 1. The tool accepts structured action inputs defining the operation type - 2. Sessions are created on-demand with isolated sandbox environments - 3. Code is executed within the Bedrock AgentCore platform with full runtime support - 4. Results, outputs, and errors are captured and returned in structured format - 5. File operations interact directly with the sandbox file system - 6. Platform lifecycle is managed automatically with cleanup on completion - - Operation Types: - -------------- - - initSession: Create a new isolated code execution session - - listLocalSessions: View all active sessions and their status - - executeCode: Run code in a specified programming language - - executeCommand: Execute shell commands in the sandbox - - readFiles: Read file contents from the sandbox file system - - writeFiles: Create or update files in the sandbox - - listFiles: Browse directory contents and file structures - - removeFiles: Delete files from the sandbox environment - - Common Usage Scenarios: - --------------------- - - Data analysis: Execute Python scripts for data processing and visualization - - Web development: Run JavaScript/TypeScript for frontend/backend development - - System administration: Execute shell commands for environment setup - - File processing: Read, transform, and write files programmatically - - Educational coding: Provide safe environments for learning and experimentation - - CI/CD workflows: Execute build scripts and deployment commands - - API testing: Run code to test external services and APIs - - Usage with Strands Agent: - ```python - from strands import Agent - from strands_tools.code_interpreter import AgentCoreCodeInterpreter - - # Create the code interpreter tool - bedrock_agent_core_code_interpreter = AgentCoreCodeInterpreter(region="us-west-2") - agent = Agent(tools=[bedrock_agent_core_code_interpreter.code_interpreter]) - - # Create a session - agent.tool.code_interpreter( - code_interpreter_input={{ - "action": {{ - "type": "initSession", - "description": "Data analysis session", - "session_name": "analysis-session" - }} - }} - ) - - # Execute Python code - agent.tool.code_interpreter( - code_interpreter_input={{ - "action": {{ - "type": "executeCode", - "session_name": "analysis-session", - "code": "import pandas as pd\\ndf = pd.read_csv('data.csv')\\nprint(df.head())", - "language": "python" - }} - }} - ) - - # Write files to the sandbox - agent.tool.code_interpreter( - code_interpreter_input={{ - "action": {{ - "type": "writeFiles", - "session_name": "analysis-session", - "content": [ - {{"path": "config.json", "text": '{{"debug": true}}'}}, - {{"path": "script.py", "text": "print('Hello, World!')"}} - ] - }} - }} - ) - # Execute shell commands - agent.tool.code_interpreter( - code_interpreter_input={{ - "action": {{ - "type": "executeCommand", - "session_name": "analysis-session", - "command": "ls -la && python script.py" - }} - }} - ) - ``` - - Args: - code_interpreter_input: Structured input containing the action to perform. - Must be a CodeInterpreterInput object with an 'action' field specifying - the operation type and required parameters. - - Action Types and Required Fields: - - InitSessionAction: type="initSession", description (required), session_name (optional) - - ExecuteCodeAction: type="executeCode", session_name, code, language, clear_context (optional) - * language must be one of: {{supported_languages_enum}} - - ExecuteCommandAction: type="executeCommand", session_name, command - - ReadFilesAction: type="readFiles", session_name, paths (list) - - WriteFilesAction: type="writeFiles", session_name, content (list of FileContent objects) - - ListFilesAction: type="listFiles", session_name, path - - RemoveFilesAction: type="removeFiles", session_name, paths (list) - - ListLocalSessionsAction: type="listLocalSessions" - - Returns: - Dict containing execution results in the format: - {{ - "status": "success|error", - "content": [{{"text": "...", "json": {{...}}}}] - }} - - Success responses include: - - Session information for session operations - - Code execution output and results - - File contents for read operations - - Operation confirmations for write/delete operations - - Error responses include: - - Session not found errors - - Code compilation/execution errors - - File system operation errors - - Platform connectivity issues + self.code_interpreter.tool_spec["description"] = """ +Execute code in isolated sandbox environments with automatic session management. + +COMMON USE CASE - Execute Code Directly: + +To execute code, provide the code to run. Sessions are created and managed automatically: + +{{ + "action": {{ + "type": "executeCode", + "code": "result = 25 * 48\\nprint(f'Result: {{result}}')", + "language": "python" + }} +}} + +The session_name parameter is optional. If not provided, a default session will be used +and created automatically if it doesn't exist. + +Supported Languages: {supported_languages_list} + +KEY FEATURES: + +1. Automatic Session Management + - Sessions are created on-demand when executing code + - No manual initialization required for simple use cases + - State persists within a session between executions + +2. File System Operations + - Read, write, list, and remove files in the sandbox + - Files persist within a session + +3. Shell Command Execution + - Execute system commands in the sandbox environment + +4. Multiple Programming Languages + - Python, JavaScript, and TypeScript supported + +OPERATION EXAMPLES: + +1. Execute Python Code: + {{ + "action": {{ + "type": "executeCode", + "code": "import pandas as pd\\ndf = pd.DataFrame({{'a': [1,2,3]}})\\nprint(df)", + "language": "python" + }} + }} + +2. Execute Shell Command: + {{ + "action": {{ + "type": "executeCommand", + "command": "ls -la" + }} + }} + +3. Write Files: + {{ + "action": {{ + "type": "writeFiles", + "content": [ + {{"path": "data.txt", "text": "Hello World"}}, + {{"path": "config.json", "text": '{{"debug": true}}'}} + ] + }} + }} + +4. Read Files: + {{ + "action": {{ + "type": "readFiles", + "paths": ["data.txt", "config.json"] + }} + }} + +5. List Files: + {{ + "action": {{ + "type": "listFiles", + "path": "." + }} + }} + +ADVANCED: Manual Session Management + +For organizing multiple execution contexts, you can explicitly create named sessions: + +{{ + "action": {{ + "type": "initSession", + "session_name": "data-analysis", + "description": "Session for data analysis tasks" + }} +}} + +Then reference the session in subsequent operations: + +{{ + "action": {{ + "type": "executeCode", + "session_name": "data-analysis", + "code": "print('Using named session')" + }} +}} + +RESPONSE FORMAT: + +Success: +{{ + "status": "success", + "content": [{{"text": "execution output"}}] +}} + +Error: +{{ + "status": "error", + "content": [{{"text": "error description"}}] +}} + +Args: + code_interpreter_input: Structured input containing the action to perform. + +Returns: + Dictionary containing execution results with status and content. """.format( supported_languages_list=", ".join([f"{lang.name}" for lang in self.get_supported_languages()]), ) @tool def code_interpreter(self, code_interpreter_input: CodeInterpreterInput) -> Dict[str, Any]: - """ - Execute code in isolated sandbox environments. - - Usage with Strands Agent: - ```python - code_interpreter = AgentCoreCodeInterpreter(region="us-west-2") - agent = Agent(tools=[code_interpreter.code_interpreter]) - ``` - - Args: - code_interpreter_input: Structured input containing the action to perform. - - Returns: - Dict containing execution results. - """ + """Execute code in isolated sandbox environments.""" - # Auto-start platform on first use if not self._started: self._start() if isinstance(code_interpreter_input, dict): - logger.debug("Action was passed as Dict, mapping to CodeInterpreterAction type action") + logger.debug("Mapping dict to CodeInterpreterInput") action = CodeInterpreterInput.model_validate(code_interpreter_input).action else: action = code_interpreter_input.action - logger.debug(f"Processing action {type(action)}") + logger.debug(f"Processing action: {type(action).__name__}") - # Delegate to platform-specific implementations + # Delegate to implementations if isinstance(action, InitSessionAction): return self.init_session(action) elif isinstance(action, ListLocalSessionsAction): @@ -238,84 +187,61 @@ def code_interpreter(self, code_interpreter_input: CodeInterpreterInput) -> Dict elif isinstance(action, WriteFilesAction): return self.write_files(action) else: - return {"status": "error", "content": [{"text": f"Unknown action type: {type(action)}"}]} + return {"status": "error", "content": [{"text": f"Unknown action: {type(action)}"}]} def _start(self) -> None: - """Start the platform and initialize any required connections.""" + """Start the platform.""" if not self._started: self.start_platform() self._started = True - logger.debug("Code Interpreter Tool started") + logger.debug("Code Interpreter started") def _cleanup(self) -> None: - """Clean up platform resources and connections.""" + """Clean up platform resources.""" if self._started: self.cleanup_platform() self._started = False - logger.debug("Code Interpreter Tool cleaned up") + logger.debug("Code Interpreter cleaned up") def __del__(self): - """Cleanup: Clear platform resources when tool is destroyed.""" + """Cleanup on destruction.""" try: if self._started: - logger.debug("Code Interpreter tool destructor called - cleaning up platform") + logger.debug("Cleaning up in destructor") self._cleanup() - logger.debug("Platform cleanup completed successfully") except Exception as e: - logger.debug("exception=<%s> | platform cleanup during destruction skipped", str(e)) + logger.debug(f"Cleanup during destruction skipped: {e}") - # Abstract methods that must be implemented by subclasses + # Abstract methods @abstractmethod - def start_platform(self) -> None: - """Initialize the platform connection and resources.""" - ... + def start_platform(self) -> None: ... @abstractmethod - def cleanup_platform(self) -> None: - """Clean up platform resources and connections.""" - ... + def cleanup_platform(self) -> None: ... @abstractmethod - def init_session(self, action: InitSessionAction) -> Dict[str, Any]: - """Initialize a new sandbox session.""" - ... + def init_session(self, action: InitSessionAction) -> Dict[str, Any]: ... @abstractmethod - def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: - """Execute code in a sandbox session.""" - ... + def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: ... @abstractmethod - def execute_command(self, action: ExecuteCommandAction) -> Dict[str, Any]: - """Execute a shell command in a sandbox session.""" - ... + def execute_command(self, action: ExecuteCommandAction) -> Dict[str, Any]: ... @abstractmethod - def read_files(self, action: ReadFilesAction) -> Dict[str, Any]: - """Read files from a sandbox session.""" - ... + def read_files(self, action: ReadFilesAction) -> Dict[str, Any]: ... @abstractmethod - def list_files(self, action: ListFilesAction) -> Dict[str, Any]: - """List files in a session directory.""" - ... + def list_files(self, action: ListFilesAction) -> Dict[str, Any]: ... @abstractmethod - def remove_files(self, action: RemoveFilesAction) -> Dict[str, Any]: - """Remove files from a sandbox session.""" - ... + def remove_files(self, action: RemoveFilesAction) -> Dict[str, Any]: ... @abstractmethod - def write_files(self, action: WriteFilesAction) -> Dict[str, Any]: - """Write files to a sandbox session.""" - ... + def write_files(self, action: WriteFilesAction) -> Dict[str, Any]: ... @abstractmethod - def list_local_sessions(self) -> Dict[str, Any]: - """List all sessions created by this platform instance.""" - ... + def list_local_sessions(self) -> Dict[str, Any]: ... @abstractmethod - def get_supported_languages(self) -> List[LanguageType]: - """list supported languages""" - ... + def get_supported_languages(self) -> List[LanguageType]: ... diff --git a/src/strands_tools/code_interpreter/models.py b/src/strands_tools/code_interpreter/models.py index 0dfbc515..88fca478 100644 --- a/src/strands_tools/code_interpreter/models.py +++ b/src/strands_tools/code_interpreter/models.py @@ -1,18 +1,15 @@ """ -Pydantic models for BedrockAgentCore Code Sandbox Strands tool. - -This module contains all the Pydantic models used for type-safe action definitions -with discriminated unions, ensuring required fields are present for each action type. +Pydantic models for Code Interpreter """ from enum import Enum -from typing import List, Literal, Union +from typing import List, Literal, Optional, Union from pydantic import BaseModel, Field class LanguageType(str, Enum): - """Supported programming languages for code execution.""" + """Supported programming languages.""" PYTHON = "python" JAVASCRIPT = "javascript" @@ -20,86 +17,99 @@ class LanguageType(str, Enum): class FileContent(BaseModel): - """Represents a file with its path and text content for writing to the sandbox file system. Used when creating or - updating files during code execution sessions.""" + """File content for writing to sandbox.""" - path: str = Field(description="The file path where content should be written") - text: str = Field(description="Text content for the file") + path: str = Field(description="File path") + text: str = Field(description="File content") -# Action-specific Pydantic models using discriminated unions class InitSessionAction(BaseModel): - """Create a new isolated code execution environment. Use this when starting a new coding task, data analysis - project, or when you need a fresh sandbox environment. Each session maintains its own state, variables, - and file system.""" + """Create a new session.""" - type: Literal["initSession"] = Field(description="Initialize a new code interpreter session") - description: str = Field(description="Required description of what this session will be used for") - session_name: str = Field(description="human-readable session name") + type: Literal["initSession"] = Field(description="Initialize session") + description: str = Field(description="Session purpose") + session_name: str = Field(description="Session name") class ListLocalSessionsAction(BaseModel): - """View all active code interpreter sessions managed by this tool instance. Use this to see what sessions are - available, check their status, or find the session name you need for other operations.""" + """List all sessions.""" - type: Literal["listLocalSessions"] = Field(description="List all local sessions managed by this tool instance") + type: Literal["listLocalSessions"] = Field(description="List sessions") class ExecuteCodeAction(BaseModel): - """Execute code in a specific programming language within an existing session. Use this for running Python - scripts, JavaScript/TypeScript code, data analysis, calculations, or any programming task. The session maintains - state between executions.""" + """Execute code in a specific programming language within an existing session.""" + + type: Literal["executeCode"] = Field(description="Execute code") - type: Literal["executeCode"] = Field(description="Execute code in the code interpreter") - session_name: str = Field(description="Required session name from a previous initSession call") - code: str = Field(description="Required code to execute") + session_name: Optional[str] = Field( + default=None, + description="Session name. If not provided, uses the " "default session which will be auto-created.", + ) + + code: str = Field(description="Code to execute") language: LanguageType = Field(default=LanguageType.PYTHON, description="Programming language for code execution") clear_context: bool = Field(default=False, description="Whether to clear the execution context before running code") class ExecuteCommandAction(BaseModel): - """Execute shell/terminal commands within the sandbox environment. Use this for system operations like installing - packages, running scripts, file management, or any command-line tasks that need to be performed in the session.""" + """Execute shell command within the sandbox environment.""" + + type: Literal["executeCommand"] = Field(description="Execute a shell command") - type: Literal["executeCommand"] = Field(description="Execute a shell command in the code interpreter") - session_name: str = Field(description="Required session name from a previous initSession call") - command: str = Field(description="Required shell command to execute") + session_name: Optional[str] = Field( + default=None, description="Session name. If not provided, uses the default session." + ) + + command: str = Field(description="Shell command to execute") class ReadFilesAction(BaseModel): - """Read the contents of one or more files from the sandbox file system. Use this to examine data files, - configuration files, code files, or any other files that have been created or uploaded to the session.""" + """Read the contents of one or more files from the sandbox file system.""" + + type: Literal["readFiles"] = Field(description="Read files") + + session_name: Optional[str] = Field( + default=None, description="Session name. If not provided, uses the default session." + ) - type: Literal["readFiles"] = Field(description="Read files from the code interpreter") - session_name: str = Field(description="Required session name from a previous initSession call") paths: List[str] = Field(description="List of file paths to read") class ListFilesAction(BaseModel): - """Browse and list files and directories within the sandbox file system. Use this to explore the directory - structure, find files, or understand what's available in the session before reading or manipulating files.""" + """Browse and list files and directories within the sandbox file system.""" type: Literal["listFiles"] = Field(description="List files in a directory") - session_name: str = Field(description="Required session name from a previous initSession call") + + session_name: Optional[str] = Field( + default=None, description="Session name. If not provided, uses the default session." + ) + path: str = Field(default=".", description="Directory path to list (defaults to current directory)") class RemoveFilesAction(BaseModel): - """Delete one or more files from the sandbox file system. Use this to clean up temporary files, remove outdated - data, or manage storage space within the session. Be careful as this permanently removes files.""" + """Delete one or more files from the sandbox file system.""" + + type: Literal["removeFiles"] = Field(description="Remove files") - type: Literal["removeFiles"] = Field(description="Remove files from the code interpreter") - session_name: str = Field(description="Required session name from a previous initSession call") - paths: List[str] = Field(description="Required list of file paths to remove") + session_name: Optional[str] = Field( + default=None, description="Session name. If not provided, uses the default session." + ) + + paths: List[str] = Field(description="List of file paths to remove") class WriteFilesAction(BaseModel): - """Create or update multiple files in the sandbox file system with specified content. Use this to save data, - create configuration files, write code files, or store any text-based content that your code execution will need.""" + """Create or update multiple files in the sandbox file system with specified content.""" + + type: Literal["writeFiles"] = Field(description="Write files") + + session_name: Optional[str] = Field( + default=None, description="Session name. If not provided, uses the default session." + ) - type: Literal["writeFiles"] = Field(description="Write files to the code interpreter") - session_name: str = Field(description="Required session name from a previous initSession call") - content: List[FileContent] = Field(description="Required list of file content to write") + content: List[FileContent] = Field(description="List of file content to write") class CodeInterpreterInput(BaseModel): diff --git a/tests/code_interpreter/test_agent_core_code_interpreter.py b/tests/code_interpreter/test_agent_core_code_interpreter.py index 7db5266b..e6feaefa 100644 --- a/tests/code_interpreter/test_agent_core_code_interpreter.py +++ b/tests/code_interpreter/test_agent_core_code_interpreter.py @@ -47,6 +47,64 @@ def test_initialization(interpreter): assert interpreter.identifier == "aws.codeinterpreter.v1" # Should use default identifier assert interpreter._sessions == {} assert not interpreter._started + assert interpreter.auto_session is True # Check default value + assert interpreter.default_session == "default" # Check default value + + +def test_ensure_session_existing_session(interpreter, mock_client): + """Test _ensure_session with existing session.""" + # Create a test session + session_info = SessionInfo(session_id="test-id", description="Test", client=mock_client) + interpreter._sessions["test-session"] = session_info + + # Test with existing session + session_name, error = interpreter._ensure_session("test-session") + + assert session_name == "test-session" + assert error is None + + +def test_ensure_session_new_session_with_auto_session(interpreter): + """Test _ensure_session with auto session creation.""" + with patch.object(interpreter, "init_session") as mock_init: + mock_init.return_value = {"status": "success", "content": [{"text": "Created"}]} + + # Test with non-existent session but auto_session enabled + session_name, error = interpreter._ensure_session("new-session") + + assert session_name == "new-session" + assert error is None + mock_init.assert_called_once() + # Verify init_session was called with correct parameters + call_args = mock_init.call_args[0][0] + assert call_args.session_name == "new-session" + assert "Auto-initialized" in call_args.description + + +def test_ensure_session_no_auto_session(interpreter): + """Test _ensure_session with auto session disabled.""" + # Disable auto session + interpreter.auto_session = False + + # Test with non-existent session + session_name, error = interpreter._ensure_session("non-existent") + + assert session_name == "non-existent" + assert error is not None + assert error["status"] == "error" + assert "not found" in error["content"][0]["text"] + + +def test_ensure_session_default_session_name(interpreter): + """Test _ensure_session uses default session name when none provided.""" + with patch.object(interpreter, "init_session") as mock_init: + mock_init.return_value = {"status": "success", "content": [{"text": "Created"}]} + + # Test with None session name + session_name, error = interpreter._ensure_session(None) + + assert session_name == "default" # Should use default session name + assert error is None def test_initialization_with_default_region(): @@ -500,6 +558,9 @@ def test_execute_code_success(interpreter, mock_client): def test_execute_code_session_not_found(interpreter): """Test code execution with non-existent session.""" + # Disable auto session creation + interpreter.auto_session = False + action = ExecuteCodeAction( type="executeCode", session_name="non-existent", code="print('Hello')", language=LanguageType.PYTHON ) @@ -525,6 +586,9 @@ def test_execute_command_success(interpreter, mock_client): def test_execute_command_session_not_found(interpreter): """Test command execution with non-existent session.""" + # Disable auto session creation + interpreter.auto_session = False + action = ExecuteCommandAction(type="executeCommand", session_name="non-existent", command="ls -la") result = interpreter.execute_command(action) @@ -667,3 +731,31 @@ def test_get_supported_languages(): assert LanguageType.JAVASCRIPT in languages assert LanguageType.TYPESCRIPT in languages assert len(languages) == 3 + + +@patch("strands_tools.code_interpreter.agent_core_code_interpreter.BedrockAgentCoreCodeInterpreterClient") +def test_execute_code_with_auto_session_creation(mock_client_class, interpreter): + """Test code execution with automatic session creation.""" + # Configure the mock client + mock_client = mock_client_class.return_value + mock_client.session_id = "auto-session-id" + mock_client.invoke.return_value = {"stream": [{"result": {"content": "Success"}}], "isError": False} + + # Execute code without creating session first + action = ExecuteCodeAction( + type="executeCode", + session_name=None, # Use default session + code="print('Auto session')", + language=LanguageType.PYTHON, + ) + + result = interpreter.execute_code(action) + + assert result["status"] == "success" + + # Verify session was created + assert "default" in interpreter._sessions + # Verify code was executed + mock_client.invoke.assert_called_with( + "executeCode", {"code": "print('Auto session')", "language": "python", "clearContext": False} + ) diff --git a/tests/code_interpreter/test_code_interpreter.py b/tests/code_interpreter/test_code_interpreter.py index bd13f72d..6c72c23f 100644 --- a/tests/code_interpreter/test_code_interpreter.py +++ b/tests/code_interpreter/test_code_interpreter.py @@ -223,7 +223,8 @@ def test_unknown_action_type(mock_interpreter): result = mock_interpreter.code_interpreter({"action": {"type": "unknownAction"}}) assert result["status"] == "error" - assert "Unknown action type" in result["content"][0]["text"] + # Update assertion to match the new error message format + assert "Unknown action:" in result["content"][0]["text"] def test_cleanup_method(mock_interpreter): @@ -291,6 +292,6 @@ def test_multiple_tool_calls_only_start_once(mock_interpreter): def test_dynamic_tool_spec(mock_interpreter): assert ( - "The tool supports the following programming languages: PYTHON, JAVASCRIPT, TYPESCRIPT" + "Supported Languages: PYTHON, JAVASCRIPT, TYPESCRIPT" in mock_interpreter.code_interpreter.tool_spec["description"] ) diff --git a/tests/test_editor.py b/tests/test_editor.py index 9d340d4b..cdc61595 100644 --- a/tests/test_editor.py +++ b/tests/test_editor.py @@ -259,12 +259,7 @@ def test_str_replace_no_backup(self, mock_user_input, temp_file, clean_content_h """Test str_replace without creating backup when EDITOR_DISABLE_BACKUP is set.""" mock_user_input.return_value = "y" - result = editor.editor( - command="str_replace", - path=temp_file, - old_str="Line 2", - new_str="Modified Line 2" - ) + result = editor.editor(command="str_replace", path=temp_file, old_str="Line 2", new_str="Modified Line 2") assert result["status"] == "success" @@ -278,12 +273,7 @@ def test_pattern_replace_no_backup(self, mock_user_input, temp_file, clean_conte """Test pattern_replace without creating backup.""" mock_user_input.return_value = "y" - result = editor.editor( - command="pattern_replace", - path=temp_file, - pattern="Line.*", - new_str="Updated Line" - ) + result = editor.editor(command="pattern_replace", path=temp_file, pattern="Line.*", new_str="Updated Line") assert result["status"] == "success" backup_path = f"{temp_file}.bak" @@ -295,12 +285,7 @@ def test_insert_no_backup(self, mock_user_input, temp_file, clean_content_histor """Test insert without creating backup.""" mock_user_input.return_value = "y" - result = editor.editor( - command="insert", - path=temp_file, - new_str="New line", - insert_line=2 - ) + result = editor.editor(command="insert", path=temp_file, new_str="New line", insert_line=2) assert result["status"] == "success" backup_path = f"{temp_file}.bak" @@ -315,12 +300,7 @@ def test_backup_created_by_default(self, mock_user_input, temp_file, clean_conte mock_user_input.return_value = "y" - result = editor.editor( - command="str_replace", - path=temp_file, - old_str="Line 2", - new_str="Modified Line 2" - ) + result = editor.editor(command="str_replace", path=temp_file, old_str="Line 2", new_str="Modified Line 2") assert result["status"] == "success" backup_path = f"{temp_file}.bak" diff --git a/tests_integ/code_interpreter/test_agent_core_code_interpreter.py b/tests_integ/code_interpreter/test_agent_core_code_interpreter.py index 442be3da..857df82c 100644 --- a/tests_integ/code_interpreter/test_agent_core_code_interpreter.py +++ b/tests_integ/code_interpreter/test_agent_core_code_interpreter.py @@ -74,4 +74,36 @@ def test_complex_natural_language_workflow(agent): 7. If and only if all steps succeed RESPOND WITH ONLY "PASS" IF NOT RETURN "FAIL" """) - assert "PASS" in result.message["content"][0]["text"] \ No newline at end of file + assert "PASS" in result.message["content"][0]["text"] + +@skip_if_github_action.mark +def test_auto_session_creation(bedrock_agent_core_code_interpreter): + """Test automatic session creation on first code execution.""" + # Execute code directly without initializing session + result = bedrock_agent_core_code_interpreter.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "code": "import platform\nprint(f'Running on Python {platform.python_version()}')", + "language": "python" + } + } + ) + + assert result['status'] == 'success' + + # Verify the default session exists + assert 'default' in bedrock_agent_core_code_interpreter._sessions + + # Execute a second command in the same auto-created session + result2 = bedrock_agent_core_code_interpreter.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "code": "print('Second execution in auto-created session')", + "language": "python" + } + } + ) + + assert result2['status'] == 'success' \ No newline at end of file diff --git a/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py b/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py index 6b200b08..df798ece 100644 --- a/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py +++ b/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py @@ -27,9 +27,9 @@ @pytest.fixture def custom_identifier_interpreter() -> AgentCoreCodeInterpreter: - """Create a real AgentCoreCodeInterpreter with custom identifier.""" - custom_id = "test-custom-interpreter-abc123" - return AgentCoreCodeInterpreter(region="us-west-2", identifier=custom_id) + """Create a real AgentCoreCodeInterpreter with custom parameters but default identifier.""" + # Use default identifier but customize other parameters + return AgentCoreCodeInterpreter(region="us-west-2", auto_session=True, default_session="custom-default") @pytest.fixture @@ -95,16 +95,16 @@ def test_complete_session_creation_flow_with_custom_identifier(self, custom_iden assert code_result['status'] == 'success' @skip_if_github_action.mark - def test_multiple_sessions_with_different_identifiers(self): - """Test creating multiple sessions with different custom identifiers.""" - # Create interpreters with different custom identifiers + def test_multiple_sessions_with_different_configurations(self): + """Test creating multiple sessions with different configurations.""" + # Create interpreters with different session configurations but same valid identifier interpreter1 = AgentCoreCodeInterpreter( - region="us-west-2", - identifier="test-interpreter-1-abc123" + region="us-west-2", + default_session="session1-default" ) interpreter2 = AgentCoreCodeInterpreter( region="us-west-2", - identifier="test-interpreter-2-def456" + default_session="session2-default" ) # Create sessions with each interpreter @@ -112,7 +112,7 @@ def test_multiple_sessions_with_different_identifiers(self): code_interpreter_input={ "action": { "type": "initSession", - "description": "First custom session", + "description": "First session", "session_name": "session-1" } } @@ -122,7 +122,7 @@ def test_multiple_sessions_with_different_identifiers(self): code_interpreter_input={ "action": { "type": "initSession", - "description": "Second custom session", + "description": "Second session", "session_name": "session-2" } } @@ -329,7 +329,7 @@ def test_logging_includes_identifier_context_on_errors(self): def test_session_not_found_error_with_custom_identifier(self): """Test that session not found errors work correctly with custom identifiers.""" custom_id = "session-not-found-test" - interpreter = AgentCoreCodeInterpreter(region="us-west-2", identifier=custom_id) + interpreter = AgentCoreCodeInterpreter(region="us-west-2", identifier=custom_id, auto_session=False) # Disable auto session # Try to execute code in non-existent session result = interpreter.code_interpreter( @@ -410,6 +410,27 @@ def test_whitespace_only_identifier_uses_default(self): # This test documents the current behavior assert interpreter.identifier == " " + @skip_if_github_action.mark + def test_complete_session_creation_flow_with_default_identifier(self, custom_identifier_interpreter): + """Test complete session creation flow with default identifier (custom ones aren't supported).""" + # Test direct tool call with default identifier + result = custom_identifier_interpreter.code_interpreter( + code_interpreter_input={ + "action": { + "type": "initSession", + "description": "Custom identifier test session", + "session_name": "custom-id-session" + } + } + ) + + # Verify session was created successfully + assert result['status'] == 'success' + assert 'sessionName' in result['content'][0]['json'] + assert result['content'][0]['json']['sessionName'] == 'custom-id-session' + assert result['content'][0]['json']['description'] == 'Custom identifier test session' + assert 'sessionId' in result['content'][0]['json'] + def test_very_long_identifier(self): """Test handling of very long identifier strings.""" long_id = "a" * 1000 # Very long identifier @@ -442,4 +463,68 @@ def test_complex_identifier_end_to_end(self): # but it tests that the identifier is passed through correctly # The specific error will depend on AWS validation # For now, we just verify the identifier was stored correctly - assert interpreter.identifier == complex_id \ No newline at end of file + assert interpreter.identifier == complex_id + + @skip_if_github_action.mark + def test_auto_session_creation_with_custom_identifier(self): # Fixed signature with self + """Test automatic session creation when executing code directly.""" + # Create interpreter with auto_session=True + interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_session=True) + + # Execute code without creating a session first + result = interpreter.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "code": "print('Auto-created session with custom identifier')", + "language": "python" + } + } + ) + + assert result['status'] == 'success' + + # Verify default session was created - FIX THIS LINE: + assert 'default' in interpreter._sessions # Use interpreter, not custom_identifier_interpreter + + # Verify we can list our auto-created session + list_result = interpreter.code_interpreter( # Use interpreter here too + code_interpreter_input={"action": {"type": "listLocalSessions"}} + ) + + assert list_result['status'] == 'success' + assert list_result['content'][0]['json']['totalSessions'] >= 1 + + # At least one session should be the default session + session_names = [s['sessionName'] for s in list_result['content'][0]['json']['sessions']] + assert 'default' in session_names + + @skip_if_github_action.mark + def test_auto_session_creation_with_default_identifier(self): + """Test automatic session creation when executing code directly.""" + # Create interpreter with auto_session=True + interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_session=True) + + # Execute code without creating a session first + result = interpreter.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "code": "print('Auto-created session with default identifier')", + "language": "python" + } + } + ) + + assert result['status'] == 'success' + + # Verify default session was created + assert 'default' in interpreter._sessions + + # Verify we can list our auto-created session + list_result = interpreter.code_interpreter( + code_interpreter_input={"action": {"type": "listLocalSessions"}} + ) + + assert list_result['status'] == 'success' + assert list_result['content'][0]['json']['totalSessions'] >= 1 \ No newline at end of file From df903b386532d22296d9128b1b748e0e9787ba45 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Tue, 14 Oct 2025 19:30:05 -0700 Subject: [PATCH 2/6] fix: docstring --- src/strands_tools/code_interpreter/code_interpreter.py | 7 ++++++- 1 file changed, 6 insertions(+), 1 deletion(-) diff --git a/src/strands_tools/code_interpreter/code_interpreter.py b/src/strands_tools/code_interpreter/code_interpreter.py index 87cdf974..9c4bd564 100644 --- a/src/strands_tools/code_interpreter/code_interpreter.py +++ b/src/strands_tools/code_interpreter/code_interpreter.py @@ -1,4 +1,9 @@ -"""Code Interpreter base class - IMPROVED VERSION""" +""" +Code Interpreter Tool implementation using Strands @tool decorator. + +This module contains the base tool class that provides lifecycle management +and can be extended by specific platform implementations. +""" import logging from abc import ABC, abstractmethod From 35d4587a8845fdb1e685d9b8c0d18ff8bf565d23 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Wed, 15 Oct 2025 18:05:59 -0700 Subject: [PATCH 3/6] fix: address PE feedback. Improved Isolation Multi-Case Support. Removed large tool description formatting changes from this PR --- .../agent_core_code_interpreter.py | 138 ++++---- .../code_interpreter/code_interpreter.py | 311 +++++++++++------- .../test_agent_core_code_interpreter.py | 69 ++-- .../code_interpreter/test_code_interpreter.py | 2 +- .../test_agent_core_code_interpreter.py | 43 +-- ...core_code_interpreter_custom_identifier.py | 132 +++++--- 6 files changed, 411 insertions(+), 284 deletions(-) diff --git a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py index d3d14654..c3af6006 100644 --- a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py +++ b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py @@ -42,58 +42,61 @@ class SessionInfo: class AgentCoreCodeInterpreter(CodeInterpreter): - """ - Bedrock AgentCore implementation of the CodeInterpreter. - - This class provides a code interpreter interface using AWS Bedrock AgentCore services. - It supports executing Python, JavaScript, and TypeScript code in isolated sandbox - environments with automatic session management. - - The class maintains session state and provides methods for code execution, file - operations, and session management. It supports both default AWS code interpreter - environments and custom environments specified by identifier. - - Args: - region (Optional[str]): AWS region for the sandbox service. If not provided, - the region will be resolved from AWS configuration. - identifier (Optional[str]): Custom code interpreter identifier to use - for code execution sessions. If not provided, defaults to the AWS-managed - identifier "aws.codeinterpreter.v1". - auto_session (bool): Enable automatic session creation when a session doesn't exist. - Defaults to True. - default_session (str): Default session name to use when session_name is not specified - in operations. Defaults to "default". - - Attributes: - region (str): The AWS region where the code interpreter service is hosted. - identifier (str): The code interpreter identifier being used for sessions. - auto_session (bool): Whether automatic session creation is enabled. - default_session (str): The default session name. - """ - def __init__( self, region: Optional[str] = None, identifier: Optional[str] = None, - auto_session: bool = True, - default_session: str = "default", + session_name: Optional[str] = None, + auto_create: bool = True, ) -> None: """ Initialize the Bedrock AgentCore code interpreter. Args: - region (Optional[str]): AWS region for the sandbox service. - identifier (Optional[str]): Custom code interpreter identifier. - auto_session (bool): Enable automatic session creation. Defaults to True. - default_session (str): Default session name. Defaults to "default". + region: AWS region for the sandbox service. + identifier: Custom code interpreter identifier. + session_name: Session name or strategy: + - None (default): Generate random session ID per instance + - "runtime": Use AgentCore runtime session_id (set at invocation time) + - Specific string: Use this exact session name + auto_create: Automatically create sessions if they don't exist. + - True (default): Create sessions on-demand + - False: Fail if session doesn't exist (strict mode) + + Examples: + # Case 1: Random session per instance (default) + interpreter = AgentCoreCodeInterpreter() + + # Case 2: Bind to runtime session (recommended for production) + session_id = getattr(context, 'session_id', None) + interpreter = AgentCoreCodeInterpreter(session_name=session_id) + + # Case 3: Named session with auto-create + interpreter = AgentCoreCodeInterpreter(session_name="my-analysis") + + # Case 4: Strict mode - must pre-initialize + interpreter = AgentCoreCodeInterpreter( + session_name="must-exist", + auto_create=False + ) """ super().__init__() self.region = resolve_region(region) self.identifier = identifier or "aws.codeinterpreter.v1" - self.auto_session = auto_session - self.default_session = default_session + self.auto_create = auto_create + + # Generate session name strategy + if session_name is None: + import uuid + + self.default_session = f"session-{uuid.uuid4().hex[:12]}" + else: + self.default_session = session_name + self._sessions: Dict[str, SessionInfo] = {} + logger.info(f"Initialized CodeInterpreter with session='{self.default_session}', " f"auto_create={auto_create}") + def start_platform(self) -> None: """Initialize the Bedrock AgentCoreplatform connection.""" pass @@ -212,42 +215,53 @@ def list_local_sessions(self) -> Dict[str, Any]: def _ensure_session(self, session_name: Optional[str]) -> tuple[str, Optional[Dict[str, Any]]]: """ - Ensure a session exists, creating it automatically if needed. + Ensure a session exists based on configuration. - This method checks if the specified session exists. If auto_session is enabled - and the session doesn't exist, it will be created automatically. + Behavior matrix: + | session_name | auto_create | Behavior | + |--------------|-------------|----------| + | None | True | Use default_session, create if needed | + | None | False | Use default_session, error if missing | + | "my-session" | True | Use "my-session", create if needed | + | "my-session" | False | Use "my-session", error if missing | Args: - session_name (Optional[str]): The session name to ensure exists. If None, - uses the default_session. + session_name: Explicit session name from action, or None to use default Returns: - tuple[str, Optional[Dict[str, Any]]]: A tuple containing: - - The session name (either provided or default) - - An error dictionary if session creation failed, None otherwise + Tuple of (session_name, error_dict or None) """ - if not session_name: - session_name = self.default_session - - if session_name in self._sessions: - return session_name, None - - if self.auto_session: - logger.info(f"Auto-creating session: {session_name}") + # Determine which session to use + target_session = session_name if session_name else self.default_session + + # Session already exists - use it + if target_session in self._sessions: + logger.debug(f"Using existing session: {target_session}") + return target_session, None + + # Session doesn't exist - check auto_create + if self.auto_create: + # Auto-create the session + logger.info(f"Auto-creating session: {target_session}") init_action = InitSessionAction( - type="initSession", - session_name=session_name, - description=f"Auto-initialized session for {session_name}", + type="initSession", session_name=target_session, description="Auto-initialized session" ) result = self.init_session(init_action) if result.get("status") != "success": - return session_name, result - - logger.info(f"Successfully auto-created session: {session_name}") - return session_name, None - - return session_name, {"status": "error", "content": [{"text": f"Session '{session_name}' not found"}]} + return target_session, result + + logger.info(f"Successfully auto-created session: {target_session}") + return target_session, None + + # auto_create=False and session doesn't exist - error + logger.error(f"Session '{target_session}' not found and auto_create is disabled") + return target_session, { + "status": "error", + "content": [ + {"text": f"Session '{target_session}' not found. " f"Create it with initSession or enable auto_create."} + ], + } def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: """Execute code in a Bedrock AgentCore session with automatic session management.""" diff --git a/src/strands_tools/code_interpreter/code_interpreter.py b/src/strands_tools/code_interpreter/code_interpreter.py index 9c4bd564..ff151d17 100644 --- a/src/strands_tools/code_interpreter/code_interpreter.py +++ b/src/strands_tools/code_interpreter/code_interpreter.py @@ -31,132 +31,191 @@ class CodeInterpreter(ABC): def __init__(self): self._started = False - self.code_interpreter.tool_spec["description"] = """ -Execute code in isolated sandbox environments with automatic session management. - -COMMON USE CASE - Execute Code Directly: - -To execute code, provide the code to run. Sessions are created and managed automatically: - -{{ - "action": {{ - "type": "executeCode", - "code": "result = 25 * 48\\nprint(f'Result: {{result}}')", - "language": "python" - }} -}} - -The session_name parameter is optional. If not provided, a default session will be used -and created automatically if it doesn't exist. - -Supported Languages: {supported_languages_list} - -KEY FEATURES: - -1. Automatic Session Management - - Sessions are created on-demand when executing code - - No manual initialization required for simple use cases - - State persists within a session between executions - -2. File System Operations - - Read, write, list, and remove files in the sandbox - - Files persist within a session - -3. Shell Command Execution - - Execute system commands in the sandbox environment - -4. Multiple Programming Languages - - Python, JavaScript, and TypeScript supported - -OPERATION EXAMPLES: - -1. Execute Python Code: - {{ - "action": {{ - "type": "executeCode", - "code": "import pandas as pd\\ndf = pd.DataFrame({{'a': [1,2,3]}})\\nprint(df)", - "language": "python" - }} - }} - -2. Execute Shell Command: - {{ - "action": {{ - "type": "executeCommand", - "command": "ls -la" - }} - }} - -3. Write Files: - {{ - "action": {{ - "type": "writeFiles", - "content": [ - {{"path": "data.txt", "text": "Hello World"}}, - {{"path": "config.json", "text": '{{"debug": true}}'}} - ] - }} - }} - -4. Read Files: - {{ - "action": {{ - "type": "readFiles", - "paths": ["data.txt", "config.json"] - }} - }} - -5. List Files: - {{ - "action": {{ - "type": "listFiles", - "path": "." - }} - }} - -ADVANCED: Manual Session Management - -For organizing multiple execution contexts, you can explicitly create named sessions: - -{{ - "action": {{ - "type": "initSession", - "session_name": "data-analysis", - "description": "Session for data analysis tasks" - }} -}} - -Then reference the session in subsequent operations: - -{{ - "action": {{ - "type": "executeCode", - "session_name": "data-analysis", - "code": "print('Using named session')" - }} -}} - -RESPONSE FORMAT: - -Success: -{{ - "status": "success", - "content": [{{"text": "execution output"}}] -}} - -Error: -{{ - "status": "error", - "content": [{{"text": "error description"}}] -}} - -Args: - code_interpreter_input: Structured input containing the action to perform. - -Returns: - Dictionary containing execution results with status and content. - """.format( - supported_languages_list=", ".join([f"{lang.name}" for lang in self.get_supported_languages()]), + # Existing description (keep all of it) + existing_description = """ + Code Interpreter tool for executing code in isolated sandbox environments. + + This tool provides a comprehensive code execution platform that supports multiple programming + languages with persistent session management, file operations, and shell command execution. + Built on the Bedrock AgentCore Code Sandbox platform, it offers secure, isolated environments + for code execution with full lifecycle management. + + Key Features: + 1. Multi-Language Support: + The tool supports the following programming languages: {supported_languages_list} + • Full standard library access for each supported language + • Runtime environment appropriate for each language + • Shell command execution for system operations + + 2. Session Management: + • Create named, persistent sessions for stateful code execution + • List and manage multiple concurrent sessions + • Automatic session cleanup and resource management + • Session isolation for security and resource separation + + 3. File System Operations: + • Read files from the sandbox environment + • Write multiple files with custom content + • List directory contents and navigate file structures + • Remove files and manage sandbox storage + + 4. Advanced Execution Features: + • Context preservation across code executions within sessions + • Optional context clearing for fresh execution environments + • Real-time output capture and error handling + • Support for long-running processes and interactive code + + How It Works: + ------------ + 1. The tool accepts structured action inputs defining the operation type + 2. Sessions are created on-demand with isolated sandbox environments + 3. Code is executed within the Bedrock AgentCore platform with full runtime support + 4. Results, outputs, and errors are captured and returned in structured format + 5. File operations interact directly with the sandbox file system + 6. Platform lifecycle is managed automatically with cleanup on completion + + Operation Types: + -------------- + - initSession: Create a new isolated code execution session + - listLocalSessions: View all active sessions and their status + - executeCode: Run code in a specified programming language + - executeCommand: Execute shell commands in the sandbox + - readFiles: Read file contents from the sandbox file system + - writeFiles: Create or update files in the sandbox + - listFiles: Browse directory contents and file structures + - removeFiles: Delete files from the sandbox environment + + Common Usage Scenarios: + --------------------- + - Data analysis: Execute Python scripts for data processing and visualization + - Web development: Run JavaScript/TypeScript for frontend/backend development + - System administration: Execute shell commands for environment setup + - File processing: Read, transform, and write files programmatically + - Educational coding: Provide safe environments for learning and experimentation + - CI/CD workflows: Execute build scripts and deployment commands + - API testing: Run code to test external services and APIs + + Usage with Strands Agent: +```python + from strands import Agent + from strands_tools.code_interpreter import AgentCoreCodeInterpreter + + # Create the code interpreter tool + bedrock_agent_core_code_interpreter = AgentCoreCodeInterpreter(region="us-west-2") + agent = Agent(tools=[bedrock_agent_core_code_interpreter.code_interpreter]) + + # Create a session + agent.tool.code_interpreter( + code_interpreter_input={{ + "action": {{ + "type": "initSession", + "description": "Data analysis session", + "session_name": "analysis-session" + }} + }} + ) + + # Execute Python code + agent.tool.code_interpreter( + code_interpreter_input={{ + "action": {{ + "type": "executeCode", + "session_name": "analysis-session", + "code": "import pandas as pd\\ndf = pd.read_csv('data.csv')\\nprint(df.head())", + "language": "python" + }} + }} + ) + + # Write files to the sandbox + agent.tool.code_interpreter( + code_interpreter_input={{ + "action": {{ + "type": "writeFiles", + "session_name": "analysis-session", + "content": [ + {{"path": "config.json", "text": '{{"debug": true}}'}}, + {{"path": "script.py", "text": "print('Hello, World!')"}} + ] + }} + }} + ) + + # Execute shell commands + agent.tool.code_interpreter( + code_interpreter_input={{ + "action": {{ + "type": "executeCommand", + "session_name": "analysis-session", + "command": "ls -la && python script.py" + }} + }} + ) +``` + + Args: + code_interpreter_input: Structured input containing the action to perform. + Must be a CodeInterpreterInput object with an 'action' field specifying + the operation type and required parameters. + + Action Types and Required Fields: + - InitSessionAction: type="initSession", description (required), session_name (optional) + - ExecuteCodeAction: type="executeCode", session_name, code, language, clear_context (optional) + * language must be one of: {{supported_languages_enum}} + - ExecuteCommandAction: type="executeCommand", session_name, command + - ReadFilesAction: type="readFiles", session_name, paths (list) + - WriteFilesAction: type="writeFiles", session_name, content (list of FileContent objects) + - ListFilesAction: type="listFiles", session_name, path + - RemoveFilesAction: type="removeFiles", session_name, paths (list) + - ListLocalSessionsAction: type="listLocalSessions" + + Returns: + Dict containing execution results in the format: + {{ + "status": "success|error", + "content": [{{"text": "...", "json": {{...}}}}] + }} + + Success responses include: + - Session information for session operations + - Code execution output and results + - File contents for read operations + - Operation confirmations for write/delete operations + + Error responses include: + - Session not found errors + - Code compilation/execution errors + - File system operation errors + - Platform connectivity issues + """ + + auto_session_note = """**IMPORTANT UPDATE: session_name is now optional in most operations** + + Sessions are automatically created when needed. You can now: + • Omit session_name to use an auto-generated session (recommended for simple use cases) + • Provide session_name for named sessions (useful for multi-step workflows) + + Quick example without session_name: +```python + # No need to call initSession first + agent.tool.code_interpreter( + code_interpreter_input={{ + "action": {{ + "type": "executeCode", + "code": "print('Hello, World!')", + "language": "python" + }} + }} + ) +``` + + --- + + """ + + # Set description: note + existing content + self.code_interpreter.tool_spec["description"] = auto_session_note + existing_description.format( + supported_languages_list=", ".join([f"{lang.name}" for lang in self.get_supported_languages()]) ) @tool diff --git a/tests/code_interpreter/test_agent_core_code_interpreter.py b/tests/code_interpreter/test_agent_core_code_interpreter.py index e6feaefa..6541344d 100644 --- a/tests/code_interpreter/test_agent_core_code_interpreter.py +++ b/tests/code_interpreter/test_agent_core_code_interpreter.py @@ -47,8 +47,7 @@ def test_initialization(interpreter): assert interpreter.identifier == "aws.codeinterpreter.v1" # Should use default identifier assert interpreter._sessions == {} assert not interpreter._started - assert interpreter.auto_session is True # Check default value - assert interpreter.default_session == "default" # Check default value + assert interpreter.default_session.startswith("session-") def test_ensure_session_existing_session(interpreter, mock_client): @@ -81,18 +80,19 @@ def test_ensure_session_new_session_with_auto_session(interpreter): assert "Auto-initialized" in call_args.description -def test_ensure_session_no_auto_session(interpreter): - """Test _ensure_session with auto session disabled.""" - # Disable auto session - interpreter.auto_session = False +def test_ensure_session_no_auto_session(): + """Test _ensure_session with auto create disabled.""" + with patch("strands_tools.code_interpreter.agent_core_code_interpreter.resolve_region") as mock_resolve: + mock_resolve.return_value = "us-west-2" + interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_create=False) # Disable auto_create - # Test with non-existent session - session_name, error = interpreter._ensure_session("non-existent") + # Test with non-existent session + session_name, error = interpreter._ensure_session("non-existent") - assert session_name == "non-existent" - assert error is not None - assert error["status"] == "error" - assert "not found" in error["content"][0]["text"] + assert session_name == "non-existent" + assert error is not None + assert error["status"] == "error" + assert "not found" in error["content"][0]["text"] def test_ensure_session_default_session_name(interpreter): @@ -103,7 +103,7 @@ def test_ensure_session_default_session_name(interpreter): # Test with None session name session_name, error = interpreter._ensure_session(None) - assert session_name == "default" # Should use default session name + assert session_name == interpreter.default_session # Use actual default session name assert error is None @@ -556,19 +556,19 @@ def test_execute_code_success(interpreter, mock_client): ) -def test_execute_code_session_not_found(interpreter): - """Test code execution with non-existent session.""" - # Disable auto session creation - interpreter.auto_session = False +def test_execute_code_session_not_found(): + """Test code execution with non-existent session when auto_create is disabled.""" + with patch("strands_tools.code_interpreter.agent_core_code_interpreter.resolve_region") as mock_resolve: + mock_resolve.return_value = "us-west-2" + interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_create=False) # Disable auto_create - action = ExecuteCodeAction( - type="executeCode", session_name="non-existent", code="print('Hello')", language=LanguageType.PYTHON - ) + action = ExecuteCodeAction( + type="executeCode", session_name="non-existent", code="print('Hello')", language=LanguageType.PYTHON + ) - result = interpreter.execute_code(action) + result = interpreter.execute_code(action) - assert result["status"] == "error" - assert "not found" in result["content"][0]["text"] + assert result["status"] == "error" def test_execute_command_success(interpreter, mock_client): @@ -584,17 +584,17 @@ def test_execute_command_success(interpreter, mock_client): mock_client.invoke.assert_called_once_with("executeCommand", {"command": "ls -la"}) -def test_execute_command_session_not_found(interpreter): - """Test command execution with non-existent session.""" - # Disable auto session creation - interpreter.auto_session = False +def test_execute_command_session_not_found(): + """Test command execution with non-existent session when auto_create is disabled.""" + with patch("strands_tools.code_interpreter.agent_core_code_interpreter.resolve_region") as mock_resolve: + mock_resolve.return_value = "us-west-2" + interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_create=False) # Disable auto_create - action = ExecuteCommandAction(type="executeCommand", session_name="non-existent", command="ls -la") + action = ExecuteCommandAction(type="executeCommand", session_name="non-existent", command="ls -la") - result = interpreter.execute_command(action) + result = interpreter.execute_command(action) - assert result["status"] == "error" - assert "not found" in result["content"][0]["text"] + assert result["status"] == "error" def test_read_files_success(interpreter, mock_client): @@ -753,8 +753,11 @@ def test_execute_code_with_auto_session_creation(mock_client_class, interpreter) assert result["status"] == "success" - # Verify session was created - assert "default" in interpreter._sessions + # Verify session was created (will be random UUID, not "default") + assert len(interpreter._sessions) == 1 + auto_created_session = list(interpreter._sessions.keys())[0] + assert auto_created_session.startswith("session-") # Check pattern instead of exact name + # Verify code was executed mock_client.invoke.assert_called_with( "executeCode", {"code": "print('Auto session')", "language": "python", "clearContext": False} diff --git a/tests/code_interpreter/test_code_interpreter.py b/tests/code_interpreter/test_code_interpreter.py index 6c72c23f..5e197c93 100644 --- a/tests/code_interpreter/test_code_interpreter.py +++ b/tests/code_interpreter/test_code_interpreter.py @@ -292,6 +292,6 @@ def test_multiple_tool_calls_only_start_once(mock_interpreter): def test_dynamic_tool_spec(mock_interpreter): assert ( - "Supported Languages: PYTHON, JAVASCRIPT, TYPESCRIPT" + "The tool supports the following programming languages: PYTHON, JAVASCRIPT, TYPESCRIPT" in mock_interpreter.code_interpreter.tool_spec["description"] ) diff --git a/tests_integ/code_interpreter/test_agent_core_code_interpreter.py b/tests_integ/code_interpreter/test_agent_core_code_interpreter.py index 857df82c..94a332c8 100644 --- a/tests_integ/code_interpreter/test_agent_core_code_interpreter.py +++ b/tests_integ/code_interpreter/test_agent_core_code_interpreter.py @@ -49,29 +49,17 @@ def test_direct_tool_call(agent): @skip_if_github_action.mark def test_complex_natural_language_workflow(agent): """Test complex multistep workflow through natural language instructions.""" - # Complex natural language instruction that requires multiple operations - result: AgentResult = agent(""" - I need you to help me with a complex data processing task. Please: - - 1. Create a new code session for this data processing project - - 2. Write a Python script called 'data_generator.py' that: - - Creates a CSV file called 'sales_data.csv' with sample sales data - - Include columns: date, product, quantity, price, customer_id - - Generate at least 20 rows of realistic sample data - - Save this data to the CSV file - - 3. Write another Javascript script called 'data_processor.js' that: - - Reads the 'sales_data.csv' file you just created - - Creates a summary report and saves it to 'sales_report.txt' - - 4. Run both scripts in sequence - - 5. Read the contents of both the CSV file and the report file to verify they were created correctly - - 6. Finally, create a shell script called 'cleanup.sh' that lists all the files we created, then run it - 7. If and only if all steps succeed RESPOND WITH ONLY "PASS" IF NOT RETURN "FAIL" + result: AgentResult = agent(""" +You have code execution capabilities. Complete this workflow efficiently by combining operations where possible: + +1. Create a Python script 'data_generator.py' that generates a CSV file 'sales_data.csv' with 20 rows (date, product, quantity, price, customer_id columns) and executes it + +2. Verify the CSV was created by reading 'sales_data.csv' + +3. Create and run a shell script 'cleanup.sh' that lists all files + +After ALL steps complete successfully, respond with ONLY the word "PASS". If any step fails, respond with "FAIL". """) assert "PASS" in result.message["content"][0]["text"] @@ -92,8 +80,10 @@ def test_auto_session_creation(bedrock_agent_core_code_interpreter): assert result['status'] == 'success' - # Verify the default session exists - assert 'default' in bedrock_agent_core_code_interpreter._sessions + # Verify a session was auto-created (will be random UUID, not "default") + assert len(bedrock_agent_core_code_interpreter._sessions) == 1 + auto_created_session = list(bedrock_agent_core_code_interpreter._sessions.keys())[0] + assert auto_created_session.startswith("session-") # Check pattern instead of exact name # Execute a second command in the same auto-created session result2 = bedrock_agent_core_code_interpreter.code_interpreter( @@ -106,4 +96,7 @@ def test_auto_session_creation(bedrock_agent_core_code_interpreter): } ) - assert result2['status'] == 'success' \ No newline at end of file + assert result2['status'] == 'success' + + # Verify still only one session exists (reused the auto-created one) + assert len(bedrock_agent_core_code_interpreter._sessions) == 1 \ No newline at end of file diff --git a/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py b/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py index df798ece..684fba2b 100644 --- a/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py +++ b/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py @@ -9,6 +9,7 @@ import logging from unittest.mock import MagicMock, patch +import time import pytest from strands import Agent @@ -29,7 +30,7 @@ def custom_identifier_interpreter() -> AgentCoreCodeInterpreter: """Create a real AgentCoreCodeInterpreter with custom parameters but default identifier.""" # Use default identifier but customize other parameters - return AgentCoreCodeInterpreter(region="us-west-2", auto_session=True, default_session="custom-default") + return AgentCoreCodeInterpreter(region="us-west-2", session_name="custom-default") @pytest.fixture @@ -49,6 +50,11 @@ def agent_with_default_identifier(default_identifier_interpreter: AgentCoreCodeI """Create an agent with default identifier code interpreter tool.""" return Agent(tools=[default_identifier_interpreter.code_interpreter]) +@pytest.fixture(autouse=True, scope="function") +def rate_limit_protection(): + """Prevent throttling between tests.""" + yield + time.sleep(2) class TestCustomIdentifierEndToEnd: """Test complete end-to-end functionality with custom identifiers.""" @@ -100,11 +106,11 @@ def test_multiple_sessions_with_different_configurations(self): # Create interpreters with different session configurations but same valid identifier interpreter1 = AgentCoreCodeInterpreter( region="us-west-2", - default_session="session1-default" + session_name="session1-default" ) interpreter2 = AgentCoreCodeInterpreter( region="us-west-2", - default_session="session2-default" + session_name="session2-default" ) # Create sessions with each interpreter @@ -140,25 +146,22 @@ def test_multiple_sessions_with_different_configurations(self): assert 'session-2' in interpreter2._sessions assert 'session-2' not in interpreter1._sessions - @skip_if_github_action.mark + @skip_if_github_action.mark def test_natural_language_workflow_with_custom_identifier(self, agent_with_custom_identifier): - """Test complex natural language workflow with custom identifier.""" - result: AgentResult = agent_with_custom_identifier(""" - I need you to help me test the custom identifier functionality. Please: - - 1. Create a new code session called 'custom-test-session' - - 2. Write a Python script that creates a simple text file with the content 'Custom identifier test successful' - - 3. Execute the script to create the file + """Test custom identifier with consolidated operations.""" - 4. Read the file back to verify it was created correctly - - 5. If all steps succeed, respond with "CUSTOM_IDENTIFIER_TEST_PASS" - """) + result = agent_with_custom_identifier(""" + Using custom identifier, complete this in ONE code execution: + 1. Create 'test.txt' with content 'Custom identifier test' + 2. Read it back + 3. Return the contents - # Verify the workflow completed successfully - assert "CUSTOM_IDENTIFIER_TEST_PASS" in result.message["content"][0]["text"] + Respond with "PASS:" followed by the file contents. + """) + + response = result.message["content"][0]["text"] + assert "PASS:" in response + assert "Custom identifier test" in response class TestBackwardCompatibility: @@ -195,7 +198,7 @@ def test_existing_functionality_unchanged_with_default_identifier(self, default_ assert code_result['status'] == 'success' - @skip_if_github_action.mark + @skip_if_github_action.mark def test_existing_natural_language_workflow_unchanged(self, agent_with_default_identifier): """Test that existing natural language workflows work unchanged.""" # This replicates the complex workflow from the original integration test @@ -329,7 +332,7 @@ def test_logging_includes_identifier_context_on_errors(self): def test_session_not_found_error_with_custom_identifier(self): """Test that session not found errors work correctly with custom identifiers.""" custom_id = "session-not-found-test" - interpreter = AgentCoreCodeInterpreter(region="us-west-2", identifier=custom_id, auto_session=False) # Disable auto session + interpreter = AgentCoreCodeInterpreter(region="us-west-2", identifier=custom_id, auto_create=False) # Try to execute code in non-existent session result = interpreter.code_interpreter( @@ -466,12 +469,12 @@ def test_complex_identifier_end_to_end(self): assert interpreter.identifier == complex_id @skip_if_github_action.mark - def test_auto_session_creation_with_custom_identifier(self): # Fixed signature with self + def test_auto_session_creation_with_custom_identifier(self): """Test automatic session creation when executing code directly.""" - # Create interpreter with auto_session=True - interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_session=True) + # Use default identifier since custom identifiers may not exist in AWS + interpreter = AgentCoreCodeInterpreter(region="us-west-2") # Remove identifier="custom-auto-test" - # Execute code without creating a session first + # Execute code without creating a session first - should auto-create with random UUID result = interpreter.code_interpreter( code_interpreter_input={ "action": { @@ -484,28 +487,31 @@ def test_auto_session_creation_with_custom_identifier(self): # Fixed signature assert result['status'] == 'success' - # Verify default session was created - FIX THIS LINE: - assert 'default' in interpreter._sessions # Use interpreter, not custom_identifier_interpreter + # Verify a session was created (will be random UUID, not "default") + assert len(interpreter._sessions) == 1 + + # Get the auto-created session name (should start with "session-") + auto_created_session = list(interpreter._sessions.keys())[0] + assert auto_created_session.startswith("session-") # Verify we can list our auto-created session - list_result = interpreter.code_interpreter( # Use interpreter here too + list_result = interpreter.code_interpreter( code_interpreter_input={"action": {"type": "listLocalSessions"}} ) assert list_result['status'] == 'success' - assert list_result['content'][0]['json']['totalSessions'] >= 1 + assert list_result['content'][0]['json']['totalSessions'] == 1 - # At least one session should be the default session + # The session name should be the auto-created one session_names = [s['sessionName'] for s in list_result['content'][0]['json']['sessions']] - assert 'default' in session_names + assert auto_created_session in session_names @skip_if_github_action.mark def test_auto_session_creation_with_default_identifier(self): """Test automatic session creation when executing code directly.""" - # Create interpreter with auto_session=True - interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_session=True) + interpreter = AgentCoreCodeInterpreter(region="us-west-2") - # Execute code without creating a session first + # Execute code without creating a session first - should auto-create with random UUID result = interpreter.code_interpreter( code_interpreter_input={ "action": { @@ -518,8 +524,12 @@ def test_auto_session_creation_with_default_identifier(self): assert result['status'] == 'success' - # Verify default session was created - assert 'default' in interpreter._sessions + # Verify a session was created (will be random UUID, not "default") + assert len(interpreter._sessions) == 1 + + # Get the auto-created session name (should start with "session-") + auto_created_session = list(interpreter._sessions.keys())[0] + assert auto_created_session.startswith("session-") # Verify we can list our auto-created session list_result = interpreter.code_interpreter( @@ -527,4 +537,52 @@ def test_auto_session_creation_with_default_identifier(self): ) assert list_result['status'] == 'success' - assert list_result['content'][0]['json']['totalSessions'] >= 1 \ No newline at end of file + assert list_result['content'][0]['json']['totalSessions'] == 1 + + # The session name should be the auto-created one + session_names = [s['sessionName'] for s in list_result['content'][0]['json']['sessions']] + assert auto_created_session in session_names + + @skip_if_github_action.mark + def test_session_name_strategies(self): + """Test different session_name strategies.""" + # Case 1: None -> random UUID + interpreter1 = AgentCoreCodeInterpreter(region="us-west-2", session_name=None) + assert interpreter1.default_session.startswith("session-") + assert len(interpreter1.default_session) == len("session-") + 12 # UUID hex[:12] + + # Case 2: Specific string -> use that string + interpreter2 = AgentCoreCodeInterpreter(region="us-west-2", session_name="my-analysis") + assert interpreter2.default_session == "my-analysis" + + # Case 3: Runtime session (simulated) + runtime_session_id = "runtime-abc123" + interpreter3 = AgentCoreCodeInterpreter(region="us-west-2", session_name=runtime_session_id) + assert interpreter3.default_session == runtime_session_id + + def test_auto_create_flag_behavior(self): + """Test auto_create flag behavior.""" + # Case 1: auto_create=True (default) - should succeed + interpreter1 = AgentCoreCodeInterpreter(region="us-west-2", auto_create=True) + assert interpreter1.auto_create == True + + # Case 2: auto_create=False - strict mode + interpreter2 = AgentCoreCodeInterpreter(region="us-west-2", auto_create=False) + assert interpreter2.auto_create == False + + # Test strict mode behavior - should fail when session doesn't exist + result = interpreter2.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "session_name": "non-existent-session", + "code": "print('This should fail')", + "language": "python" + } + } + ) + + assert result['status'] == 'error' + error_message = result['content'][0]['text'] + assert "not found" in error_message + assert "auto_create" in error_message \ No newline at end of file From 73f1543407d57a7f81dca56989bda882b8f4b688 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Thu, 16 Oct 2025 12:03:32 -0700 Subject: [PATCH 4/6] fix: address comments on docstrings, comments --- .../agent_core_code_interpreter.py | 11 +-- .../code_interpreter/code_interpreter.py | 56 ++++++++++----- src/strands_tools/code_interpreter/models.py | 71 ++++++++++++------- .../test_agent_core_code_interpreter.py | 22 +++--- 4 files changed, 95 insertions(+), 65 deletions(-) diff --git a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py index c3af6006..42788dc7 100644 --- a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py +++ b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py @@ -254,14 +254,9 @@ def _ensure_session(self, session_name: Optional[str]) -> tuple[str, Optional[Di logger.info(f"Successfully auto-created session: {target_session}") return target_session, None - # auto_create=False and session doesn't exist - error - logger.error(f"Session '{target_session}' not found and auto_create is disabled") - return target_session, { - "status": "error", - "content": [ - {"text": f"Session '{target_session}' not found. " f"Create it with initSession or enable auto_create."} - ], - } + # auto_create=False and session doesn't exist - raise exception + logger.debug(f"Session '{target_session}' not found (auto_create disabled)") + raise ValueError(f"Session '{target_session}' not found. Create it first using initSession.") def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: """Execute code in a Bedrock AgentCore session with automatic session management.""" diff --git a/src/strands_tools/code_interpreter/code_interpreter.py b/src/strands_tools/code_interpreter/code_interpreter.py index ff151d17..45183cdd 100644 --- a/src/strands_tools/code_interpreter/code_interpreter.py +++ b/src/strands_tools/code_interpreter/code_interpreter.py @@ -32,7 +32,7 @@ def __init__(self): self._started = False # Existing description (keep all of it) - existing_description = """ + description_template = """ Code Interpreter tool for executing code in isolated sandbox environments. This tool provides a comprehensive code execution platform that supports multiple programming @@ -189,7 +189,7 @@ def __init__(self): - Platform connectivity issues """ - auto_session_note = """**IMPORTANT UPDATE: session_name is now optional in most operations** + auto_session_note = """**session_name is now optional in most operations** Sessions are automatically created when needed. You can now: • Omit session_name to use an auto-generated session (recommended for simple use cases) @@ -214,7 +214,7 @@ def __init__(self): """ # Set description: note + existing content - self.code_interpreter.tool_spec["description"] = auto_session_note + existing_description.format( + self.code_interpreter.tool_spec["description"] = auto_session_note + description_template.format( supported_languages_list=", ".join([f"{lang.name}" for lang in self.get_supported_languages()]) ) @@ -271,41 +271,63 @@ def __del__(self): """Cleanup on destruction.""" try: if self._started: - logger.debug("Cleaning up in destructor") + logger.debug("Platform cleanup completed successfully") self._cleanup() except Exception as e: - logger.debug(f"Cleanup during destruction skipped: {e}") + logger.debug("exception=<%s> | platform cleanup during destruction skipped", str(e)) - # Abstract methods + # Abstract methods that must be implemented by subclasses @abstractmethod - def start_platform(self) -> None: ... + def start_platform(self) -> None: + """Initialize the platform connection and resources.""" + ... @abstractmethod - def cleanup_platform(self) -> None: ... + def cleanup_platform(self) -> None: + """Clean up platform resources and connections.""" + ... @abstractmethod - def init_session(self, action: InitSessionAction) -> Dict[str, Any]: ... + def init_session(self, action: InitSessionAction) -> Dict[str, Any]: + """Initialize a new sandbox session.""" + ... @abstractmethod - def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: ... + def execute_code(self, action: ExecuteCodeAction) -> Dict[str, Any]: + """Execute code in a sandbox session.""" + ... @abstractmethod - def execute_command(self, action: ExecuteCommandAction) -> Dict[str, Any]: ... + def execute_command(self, action: ExecuteCommandAction) -> Dict[str, Any]: + """Execute a shell command in a sandbox session.""" + ... @abstractmethod - def read_files(self, action: ReadFilesAction) -> Dict[str, Any]: ... + def read_files(self, action: ReadFilesAction) -> Dict[str, Any]: + """Read files from a sandbox session.""" + ... @abstractmethod - def list_files(self, action: ListFilesAction) -> Dict[str, Any]: ... + def list_files(self, action: ListFilesAction) -> Dict[str, Any]: + """List files in a session directory.""" + ... @abstractmethod - def remove_files(self, action: RemoveFilesAction) -> Dict[str, Any]: ... + def remove_files(self, action: RemoveFilesAction) -> Dict[str, Any]: + """Remove files from a sandbox session.""" + ... @abstractmethod - def write_files(self, action: WriteFilesAction) -> Dict[str, Any]: ... + def write_files(self, action: WriteFilesAction) -> Dict[str, Any]: + """Write files to a sandbox session.""" + ... @abstractmethod - def list_local_sessions(self) -> Dict[str, Any]: ... + def list_local_sessions(self) -> Dict[str, Any]: + """List all sessions created by this platform instance.""" + ... @abstractmethod - def get_supported_languages(self) -> List[LanguageType]: ... + def get_supported_languages(self) -> List[LanguageType]: + """list supported languages""" + ... diff --git a/src/strands_tools/code_interpreter/models.py b/src/strands_tools/code_interpreter/models.py index 88fca478..b02f9b85 100644 --- a/src/strands_tools/code_interpreter/models.py +++ b/src/strands_tools/code_interpreter/models.py @@ -1,5 +1,8 @@ """ -Pydantic models for Code Interpreter +Pydantic models for BedrockAgentCore Code Sandbox Strands tool. + +This module contains all the Pydantic models used for type-safe action definitions +with discriminated unions, ensuring required fields are present for each action type. """ from enum import Enum @@ -9,7 +12,7 @@ class LanguageType(str, Enum): - """Supported programming languages.""" + """Supported programming languages for code execution.""" PYTHON = "python" JAVASCRIPT = "javascript" @@ -17,57 +20,68 @@ class LanguageType(str, Enum): class FileContent(BaseModel): - """File content for writing to sandbox.""" + """Represents a file with its path and text content for writing to the sandbox file system. Used when creating or + updating files during code execution sessions.""" - path: str = Field(description="File path") - text: str = Field(description="File content") + path: str = Field(description="The file path where content should be written") + text: str = Field(description="Text content for the file") +# Action-specific Pydantic models using discriminated unions class InitSessionAction(BaseModel): - """Create a new session.""" + """Create a new isolated code execution environment. Use this when starting a new coding task, data analysis + project, or when you need a fresh sandbox environment. Each session maintains its own state, variables, + and file system.""" - type: Literal["initSession"] = Field(description="Initialize session") - description: str = Field(description="Session purpose") - session_name: str = Field(description="Session name") + type: Literal["initSession"] = Field(description="Initialize a new code interpreter session") + description: str = Field(description="Required description of what this session will be used for") + session_name: Optional[str] = Field( + default=None, description="Session name. If not provided, a default session will be used." + ) class ListLocalSessionsAction(BaseModel): - """List all sessions.""" + """View all active code interpreter sessions managed by this tool instance. Use this to see what sessions are + available, check their status, or find the session name you need for other operations.""" - type: Literal["listLocalSessions"] = Field(description="List sessions") + type: Literal["listLocalSessions"] = Field(description="List all local sessions managed by this tool instance") class ExecuteCodeAction(BaseModel): - """Execute code in a specific programming language within an existing session.""" + """Execute code in a specific programming language within an existing session. Use this for running Python + scripts, JavaScript/TypeScript code, data analysis, calculations, or any programming task. The session maintains + state between executions.""" - type: Literal["executeCode"] = Field(description="Execute code") + type: Literal["executeCode"] = Field(description="Execute code in the code interpreter") session_name: Optional[str] = Field( default=None, - description="Session name. If not provided, uses the " "default session which will be auto-created.", + description="Session name. If not provided, uses the default session which will be auto-created if needed.", ) - code: str = Field(description="Code to execute") + code: str = Field(description="Required code to execute") language: LanguageType = Field(default=LanguageType.PYTHON, description="Programming language for code execution") clear_context: bool = Field(default=False, description="Whether to clear the execution context before running code") class ExecuteCommandAction(BaseModel): - """Execute shell command within the sandbox environment.""" + """Execute shell/terminal commands within the sandbox environment. Use this for system operations like installing + packages, running scripts, file management, or any command-line tasks that need to be performed in the session.""" - type: Literal["executeCommand"] = Field(description="Execute a shell command") + type: Literal["executeCommand"] = Field(description="Execute a shell command in the code interpreter") session_name: Optional[str] = Field( default=None, description="Session name. If not provided, uses the default session." ) - command: str = Field(description="Shell command to execute") + command: str = Field(description="Required shell command to execute") class ReadFilesAction(BaseModel): - """Read the contents of one or more files from the sandbox file system.""" + """Read the contents of one or more files from the sandbox file system. Use this to examine data files, + configuration files, code files, or any other files that have been created or uploaded to the session.""" - type: Literal["readFiles"] = Field(description="Read files") + type: Literal["readFiles"] = Field(description="Read files from the code interpreter") session_name: Optional[str] = Field( default=None, description="Session name. If not provided, uses the default session." @@ -77,7 +91,8 @@ class ReadFilesAction(BaseModel): class ListFilesAction(BaseModel): - """Browse and list files and directories within the sandbox file system.""" + """Browse and list files and directories within the sandbox file system. Use this to explore the directory + structure, find files, or understand what's available in the session before reading or manipulating files.""" type: Literal["listFiles"] = Field(description="List files in a directory") @@ -89,27 +104,29 @@ class ListFilesAction(BaseModel): class RemoveFilesAction(BaseModel): - """Delete one or more files from the sandbox file system.""" + """Delete one or more files from the sandbox file system. Use this to clean up temporary files, remove outdated + data, or manage storage space within the session. Be careful as this permanently removes files.""" - type: Literal["removeFiles"] = Field(description="Remove files") + type: Literal["removeFiles"] = Field(description="Remove files from the code interpreter") session_name: Optional[str] = Field( default=None, description="Session name. If not provided, uses the default session." ) - paths: List[str] = Field(description="List of file paths to remove") + paths: List[str] = Field(description="Required list of file paths to remove") class WriteFilesAction(BaseModel): - """Create or update multiple files in the sandbox file system with specified content.""" + """Create or update multiple files in the sandbox file system with specified content. Use this to save data, + create configuration files, write code files, or store any text-based content that your code execution will need.""" - type: Literal["writeFiles"] = Field(description="Write files") + type: Literal["writeFiles"] = Field(description="Write files to the code interpreter") session_name: Optional[str] = Field( default=None, description="Session name. If not provided, uses the default session." ) - content: List[FileContent] = Field(description="List of file content to write") + content: List[FileContent] = Field(description="Required list of file content to write") class CodeInterpreterInput(BaseModel): diff --git a/tests/code_interpreter/test_agent_core_code_interpreter.py b/tests/code_interpreter/test_agent_core_code_interpreter.py index 6541344d..8f50506f 100644 --- a/tests/code_interpreter/test_agent_core_code_interpreter.py +++ b/tests/code_interpreter/test_agent_core_code_interpreter.py @@ -86,13 +86,9 @@ def test_ensure_session_no_auto_session(): mock_resolve.return_value = "us-west-2" interpreter = AgentCoreCodeInterpreter(region="us-west-2", auto_create=False) # Disable auto_create - # Test with non-existent session - session_name, error = interpreter._ensure_session("non-existent") - - assert session_name == "non-existent" - assert error is not None - assert error["status"] == "error" - assert "not found" in error["content"][0]["text"] + # Test with non-existent session - should raise ValueError + with pytest.raises(ValueError, match="Session 'non-existent' not found. Create it first using initSession."): + interpreter._ensure_session("non-existent") def test_ensure_session_default_session_name(interpreter): @@ -566,9 +562,9 @@ def test_execute_code_session_not_found(): type="executeCode", session_name="non-existent", code="print('Hello')", language=LanguageType.PYTHON ) - result = interpreter.execute_code(action) - - assert result["status"] == "error" + # Expect ValueError to be raised + with pytest.raises(ValueError, match="Session 'non-existent' not found. Create it first using initSession."): + interpreter.execute_code(action) def test_execute_command_success(interpreter, mock_client): @@ -592,9 +588,9 @@ def test_execute_command_session_not_found(): action = ExecuteCommandAction(type="executeCommand", session_name="non-existent", command="ls -la") - result = interpreter.execute_command(action) - - assert result["status"] == "error" + # Expect ValueError to be raised + with pytest.raises(ValueError, match="Session 'non-existent' not found. Create it first using initSession."): + interpreter.execute_command(action) def test_read_files_success(interpreter, mock_client): From 387f6f2dab1d8fdaa6f6c6cff7b3b36d430f1e30 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Thu, 16 Oct 2025 12:09:23 -0700 Subject: [PATCH 5/6] fix: updated import uuid --- .../code_interpreter/agent_core_code_interpreter.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py index 42788dc7..9c55acb7 100644 --- a/src/strands_tools/code_interpreter/agent_core_code_interpreter.py +++ b/src/strands_tools/code_interpreter/agent_core_code_interpreter.py @@ -1,4 +1,5 @@ import logging +import uuid from dataclasses import dataclass from typing import Any, Dict, List, Optional @@ -87,8 +88,6 @@ def __init__( # Generate session name strategy if session_name is None: - import uuid - self.default_session = f"session-{uuid.uuid4().hex[:12]}" else: self.default_session = session_name From 3384adefce02dbd33c9702855f199bc6bc7466e9 Mon Sep 17 00:00:00 2001 From: Sundar Raghavan Date: Tue, 21 Oct 2025 13:11:23 -0700 Subject: [PATCH 6/6] fix: integration test issues --- ...core_code_interpreter_custom_identifier.py | 55 ++++++++++--------- 1 file changed, 29 insertions(+), 26 deletions(-) diff --git a/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py b/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py index 684fba2b..a1cdfe65 100644 --- a/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py +++ b/tests_integ/code_interpreter/test_agent_core_code_interpreter_custom_identifier.py @@ -334,23 +334,24 @@ def test_session_not_found_error_with_custom_identifier(self): custom_id = "session-not-found-test" interpreter = AgentCoreCodeInterpreter(region="us-west-2", identifier=custom_id, auto_create=False) - # Try to execute code in non-existent session - result = interpreter.code_interpreter( - code_interpreter_input={ - "action": { - "type": "executeCode", - "session_name": "non-existent-session", - "code": "print('This should fail')", - "language": "python" + # Try to execute code in non-existent session - should raise ValueError + with pytest.raises(ValueError) as exc_info: + interpreter.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "session_name": "non-existent-session", + "code": "print('This should fail')", + "language": "python" + } } - } - ) + ) - # Verify error response - assert result['status'] == 'error' - error_message = result['content'][0]['text'] + # Verify error message contains expected information + error_message = str(exc_info.value) assert "non-existent-session" in error_message assert "not found" in error_message + assert "initSession" in error_message def test_multiple_error_scenarios_with_different_identifiers(self): """Test various error scenarios with different custom identifiers.""" @@ -570,19 +571,21 @@ def test_auto_create_flag_behavior(self): interpreter2 = AgentCoreCodeInterpreter(region="us-west-2", auto_create=False) assert interpreter2.auto_create == False - # Test strict mode behavior - should fail when session doesn't exist - result = interpreter2.code_interpreter( - code_interpreter_input={ - "action": { - "type": "executeCode", - "session_name": "non-existent-session", - "code": "print('This should fail')", - "language": "python" + # Test strict mode behavior - should raise ValueError when session doesn't exist + with pytest.raises(ValueError) as exc_info: + interpreter2.code_interpreter( + code_interpreter_input={ + "action": { + "type": "executeCode", + "session_name": "non-existent-session", + "code": "print('This should fail')", + "language": "python" + } } - } - ) + ) - assert result['status'] == 'error' - error_message = result['content'][0]['text'] + # Verify error message contains expected information + error_message = str(exc_info.value) + assert "non-existent-session" in error_message assert "not found" in error_message - assert "auto_create" in error_message \ No newline at end of file + assert "initSession" in error_message \ No newline at end of file