Implementation:CrewAIInc CrewAI Bedrock Code Interpreter Toolkit

Knowledge Sources	CrewAI
Domains	AWS, Code_Execution, Tools
Last Updated	2026-02-11 00:00 GMT

Overview

Comprehensive toolkit for executing code and managing files in AWS Bedrock Code Interpreter environments through CrewAI tools, with 9 tool classes and thread-based session management.

Description

This module implements the CodeInterpreterToolkit class and 9 individual tool classes that wrap AWS Bedrock AgentCore code interpreter API calls. Each tool extends BaseTool and provides both synchronous and asynchronous execution methods.

Tool classes:

• ExecuteCodeTool (execute_code): Run code in various languages (primarily Python) with optional context clearing
• ExecuteCommandTool (execute_command): Run shell commands in the interpreter environment
• ReadFilesTool (read_files): Read content of files by path list
• ListFilesTool (list_files): List files in a directory
• DeleteFilesTool (delete_files): Remove files by path list
• WriteFilesTool (write_files): Create or update files with content dictionaries
• StartCommandTool (start_command_execution): Start long-running commands asynchronously
• GetTaskTool (get_task): Check status of async tasks by task ID
• StopTaskTool (stop_task): Stop running tasks by task ID

CodeInterpreterToolkit manages the lifecycle:

• Lazy initialization of code interpreter sessions on first use
• Thread-based session isolation via _code_interpreters dictionary keyed by thread_id
• Cleanup of individual or all sessions via async cleanup() method
• Tool access via get_tools() and get_tools_by_name()

The extract_output_from_stream() helper function processes code interpreter response streams, extracting text content and file resources.

Usage

Use this toolkit when CrewAI agents need to execute arbitrary code, run shell commands, or perform file operations in isolated, secure AWS environments. The thread-based session management enables multiple parallel code execution contexts for multi-agent scenarios.

Code Reference

Source Location

• Repository: CrewAI
• File: lib/crewai-tools/src/crewai_tools/aws/bedrock/code_interpreter/code_interpreter_toolkit.py
• Lines: 1-625

Signature (CodeInterpreterToolkit)

class CodeInterpreterToolkit:
    """Toolkit for working with AWS Bedrock code interpreter environment."""

    def __init__(self, region: str = "us-west-2"):
        self.region = region
        self._code_interpreters: dict[str, CodeInterpreter] = {}
        self.tools: list[BaseTool] = []

Signature (ExecuteCodeTool)

class ExecuteCodeTool(BaseTool):
    """Tool for executing code in various languages."""

    name: str = "execute_code"
    description: str = "Execute code in various languages (primarily Python)"
    args_schema: type[BaseModel] = ExecuteCodeInput
    toolkit: Any = Field(default=None, exclude=True)

Factory Function

def create_code_interpreter_toolkit(
    region: str = "us-west-2",
) -> tuple[CodeInterpreterToolkit, list[BaseTool]]:
    """Create a CodeInterpreterToolkit and return (toolkit, tools)."""

Import

from crewai_tools.aws.bedrock.code_interpreter.code_interpreter_toolkit import (
    CodeInterpreterToolkit,
    create_code_interpreter_toolkit,
)

I/O Contract

Inputs (CodeInterpreterToolkit)

Name	Type	Required	Description
region	str	No	AWS region for the code interpreter (default: "us-west-2")

Tool Input Schemas

Tool	Parameters	Description
ExecuteCodeTool	code (str), language (str), clear_context (bool), thread_id (str)	Execute code with optional language and context clearing
ExecuteCommandTool	command (str), thread_id (str)	Run a shell command
ReadFilesTool	paths (list[str]), thread_id (str)	Read files at specified paths
ListFilesTool	directory_path (str), thread_id (str)	List files in a directory
DeleteFilesTool	paths (list[str]), thread_id (str)	Delete files at specified paths
WriteFilesTool	files (list[dict[str, str]]), thread_id (str)	Write files with path and text content
StartCommandTool	command (str), thread_id (str)	Start a long-running command asynchronously
GetTaskTool	task_id (str), thread_id (str)	Check status of an async task
StopTaskTool	task_id (str), thread_id (str)	Stop a running task

Outputs

Name	Type	Description
get_tools()	list[BaseTool]	List of 9 code interpreter tools
get_tools_by_name()	dict[str, BaseTool]	Dictionary mapping tool names to tool instances
Tool _run()	str	Extracted output from the code interpreter response stream
cleanup()	None	Stops and removes code interpreter sessions

Key Functions

extract_output_from_stream

def extract_output_from_stream(response):
    """Extract output from code interpreter response stream.

    Processes 'result' events, extracting 'text' content items directly
    and 'resource' content items as file content with paths.
    Returns newline-joined output string.
    """

Usage Examples

Basic Usage

from crewai import Agent, Task, Crew
from crewai_tools.aws.bedrock.code_interpreter import create_code_interpreter_toolkit

# Create the code interpreter toolkit
toolkit, code_tools = create_code_interpreter_toolkit(region="us-west-2")

# Create a CrewAI agent with code interpreter tools
developer_agent = Agent(
    role="Python Developer",
    goal="Create and execute Python code to solve problems",
    backstory="You're a skilled Python developer.",
    tools=code_tools,
)

# Create a task
coding_task = Task(
    description="Write a Python function that calculates factorial and test it.",
    agent=developer_agent,
)

# Run the crew
crew = Crew(agents=[developer_agent], tasks=[coding_task])
result = crew.kickoff()

# Clean up resources
import asyncio
asyncio.run(toolkit.cleanup())

Cleaning Up a Specific Thread

# Clean up only a specific thread's session
await toolkit.cleanup(thread_id="my_thread")

# Clean up all sessions
await toolkit.cleanup()