CrewAI 接入 AWS Bedrock Code Interpreter：构建安全隔离的代码执行 Agent 工具链-平芜编程栈

CrewAI 接入 AWS Bedrock Code Interpreter：构建安全隔离的代码执行 Agent 工具链

【免费下载链接】crewAIFramework for orchestrating role-playing, autonomous AI agents. By fostering collaborative intelligence, CrewAI empowers agents to work together seamlessly, tackling complex tasks.项目地址: https://gitcode.com/GitHub_Trending/cr/crewAI

本文围绕 CrewAI 的 AWS Bedrock Code Interpreter 工具包（位于lib/crewai-tools包中的crewai_tools.aws.bedrock.code_interpreter模块）展开。读完后，你将能够：在 CrewAI Agent 中接入 AWS Bedrock AgentCore 的远程代码解释器环境，让 Agent 安全地执行 Python 代码、运行 Shell 命令、管理文件，并通过thread_id实现多会话隔离；同时理解工具包底层的懒加载会话机制、流式输出解析与资源清理的实现原理，避免云端会话泄漏。

一、模块定位与整体架构

该工具包是crewai_tools中aws子包的一部分，与 Browser 工具包、Agent 调用工具、知识库检索工具、S3 读写工具并列，统一通过crewai_tools.aws对外导出：

入口导出：crewai_tools/aws/init.py 中create_code_interpreter_toolkit被列为公开 API；
模块导出：code_interpreter/init.py 暴露CodeInterpreterToolkit与create_code_interpreter_toolkit两个符号；
核心实现：全部集中在 code_interpreter_toolkit.py 一个文件内（约 600 行）。

它的价值在于：Agent 不再在本地进程里执行模型生成的代码，而是把代码发送到 AWS 侧的 Bedrock AgentCore Code Interpreter 沙箱中运行，获得一个安全、隔离、有文件系统与 Shell 能力的远程执行环境，适用于数据分析、脚本调试、批量文件处理等场景。

从源码结构看，整个工具包可以拆成三层：

输入 Schema 层：9 个 PydanticBaseModel（如ExecuteCodeInput、WriteFilesInput），为每个工具定义参数名、类型、默认值与描述。这些描述会进入 LLM 的工具调用提示词，直接影响 Agent 传参的准确性；
Tool 类层：9 个继承自crewai.tools.BaseTool的工具类，每个工具的_run方法负责取/建对应会话并调用底层CodeInterpreter.invoke(method=..., params=...)，_arun则直接委托给同步实现（源码注释说明底层同步 API 是线程安全的）；
Toolkit 会话管理层：CodeInterpreterToolkit负责按thread_id懒加载并缓存多个CodeInterpreter会话，并提供cleanup()回收资源。

二、安装与前置条件

工具包依赖独立的第三方包bedrock-agentcore，在 pyproject.toml 中以bedrock可选依赖（extra）声明，版本约束为bedrock-agentcore>=1.18.1,<2.0.0，同时附带playwright、nest-asyncio、beautifulsoup4（这些是 Bedrock 浏览器工具等整个 extra 共用的依赖）。

安装命令（来自原 README）：

uv add crewai-tools bedrock-agentcore

使用前提（原 README "Requirements" 部分）：

拥有可访问Bedrock AgentCore API的 AWS 账户；
正确配置了 AWS 凭据（标准 AWS 凭据链即可，如环境变量、~/.aws/credentials、IAM Role）；
region参数需与你的 AgentCore 资源所在区域一致，所有示例默认使用us-west-2。

三、九大内置工具与底层方法映射

工具包共提供 9 个工具。下表把每个工具的名称、输入参数与它实际调用的 Bedrock AgentCore 方法对应起来（依据 code_interpreter_toolkit.py 中各_run实现的invoke(method=..., params=...)调用）：

工具名 (`name`)	输入参数	对应 AgentCore 方法	说明
`execute_code`	`code`、`language`（默认`"python"`）、`clear_context`（默认`False`）、`thread_id`	`executeCode`	执行代码；`clear_context=True`时清空执行上下文
`execute_command`	`command`、`thread_id`	`executeCommand`	在沙箱内运行 Shell 命令
`read_files`	`paths: list[str]`、`thread_id`	`readFiles`	批量读取文件内容
`list_files`	`directory_path`（默认`""`）、`thread_id`	`listFiles`	列出目录内容
`delete_files`	`paths: list[str]`、`thread_id`	`removeFiles`	批量删除文件（注意方法名与工具名不同）
`write_files`	`files: list[dict]`（含`path`/`text`字段）、`thread_id`	`writeFiles`	创建或更新文件，参数字段名为`content`
`start_command_execution`	`command`、`thread_id`	`startCommandExecution`	异步启动长时命令
`get_task`	`task_id`、`thread_id`	`getTask`	查询异步任务状态
`stop_task`	`task_id`、`thread_id`	`stopTask`	停止运行中的任务

两个值得注意的实现细节：

1. 所有工具都带thread_id参数（默认"default"）。这是多会话隔离的基石：工具包内部维护_code_interpreters: dict[str, CodeInterpreter]字典，每个thread_id对应一个独立的远程会话。不同thread_id之间的变量、文件与执行上下文互不干扰——你可以让一个线程做数据清洗、另一个线程做可视化，而彼此不会覆盖对方的工作区。

2. 输出统一经extract_output_from_stream()解析。该函数遍历响应中的stream事件，对result.content逐项处理：type == "text"的条目直接拼接文本；type == "resource"的条目若是带text的文件资源，则去掉file://前缀并格式化为==== File: <路径> ====的代码块输出，其他资源则序列化为 JSON。这意味着当 Agent 在沙箱里生成图片、CSV 等文件并被返回为资源时，工具能以 Agent 可理解的结构化文本呈现，而不是原始二进制。

四、基础用法：单 Agent 执行代码

以下示例完整继承自原 README：创建一个代码解释器工具包、一个使用 Bedrock 上 Claude 模型的 Agent，让它编写并自测一个阶乘函数。

from crewai import Agent, Task, Crew, LLM from crewai_tools.aws import create_code_interpreter_toolkit # Create the code interpreter toolkit toolkit, code_tools = create_code_interpreter_toolkit(region="us-west-2") # Create the Bedrock LLM llm = LLM( model="bedrock/us.anthropic.claude-3-7-sonnet-20250219-v1:0", region_name="us-west-2", ) # Create a CrewAI agent that uses the 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 with expertise in data analysis.", tools=code_tools, llm=llm ) # Create a task for the agent coding_task = Task( description="Write a Python function that calculates the factorial of a number and test it. Do not use any imports from outside the Python standard library.", expected_output="The Python function created, and the test results.", agent=developer_agent ) # Create and run the crew crew = Crew( agents=[developer_agent], tasks=[coding_task] ) result = crew.kickoff() print(f"\n***Final result:***\n\n{result}") # Clean up resources when done import asyncio asyncio.run(toolkit.cleanup())

关键点解析：

create_code_interpreter_toolkit(region="us-west-2")返回二元组(toolkit, tools)。注意此时并不会创建任何远程会话——源码_setup_tools()只实例化 9 个工具对象，真正的CodeInterpreter会话直到第一次工具调用时才由_get_or_create_interpreter()创建（内部调用CodeInterpreter(region=self.region)并start()）。这种懒加载意味着即使创建了工具包但 Agent 从未触发代码执行，也不会产生任何云端会话开销；
tools=code_tools表示把全部 9 个工具一次性交给 Agent，适合"全能开发者"型角色；
任务描述里显式约束"不使用标准库以外的 import"，是为了确保沙箱内无需额外依赖即可运行，这也是在远程受限环境中编写任务描述的一个实用技巧。

五、进阶用法：按名称精确装配工具

当 Crew 里有多个 Agent、职责不同时，更细粒度的做法是只把与角色匹配的工具子集交给对应 Agent。toolkit.get_tools_by_name()返回{工具名: 工具实例}字典，可按名取用：

from crewai import Agent, Task, Crew, LLM from crewai_tools.aws import create_code_interpreter_toolkit # Create the code interpreter toolkit toolkit, code_tools = create_code_interpreter_toolkit(region="us-west-2") tools_by_name = toolkit.get_tools_by_name() # Create the Bedrock LLM llm = LLM( model="bedrock/us.anthropic.claude-3-7-sonnet-20250219-v1:0", region_name="us-west-2", ) # Create agents with specific tools code_agent = Agent( role="Code Developer", goal="Write and execute code", backstory="You write and test code to solve complex problems.", tools=[ # Use specific tools by name tools_by_name["execute_code"], tools_by_name["execute_command"], tools_by_name["read_files"], tools_by_name["write_files"] ], llm=llm ) file_agent = Agent( role="File Manager", goal="Manage files in the environment", backstory="You help organize and manage files in the code environment.", tools=[ # Use specific tools by name tools_by_name["list_files"], tools_by_name["read_files"], tools_by_name["write_files"], tools_by_name["delete_files"] ], llm=llm ) # Create tasks for the agents coding_task = Task( description="Write a Python script to analyze data from a CSV file. Do not use any imports from outside the Python standard library.", expected_output="The Python function created.", agent=code_agent ) file_task = Task( description="Organize the created files into separate directories.", agent=file_agent ) # Create and run the crew crew = Crew( agents=[code_agent, file_agent], tasks=[coding_task, file_task] ) result = crew.kickoff() print(f"\n***Final result:***\n\n{result}") # Clean up code interpreter resources when done import asyncio asyncio.run(toolkit.cleanup())

这种拆分带来三个实际好处：

提示词更聚焦：Agent 每次推理看到的工具清单更短，参数混淆概率更低；
权限最小化：文件管理 Agent 拿不到execute_code，即使被提示词注入也难以在沙箱里执行任意代码；
共享同一工作区：两个 Agent 的工具都使用默认thread_id="default"，因此code_agent写入的 CSV 与脚本，file_agent能在同一个沙箱会话中直接看到并整理——这是多 Agent 协作的天然基础。若需要彼此隔离的工作区，则给不同 Agent 的任务/工具调用显式传不同的thread_id。

六、实战示例：数据分析全流程

下面这个示例（同样继承自原 README）展示了"生成数据 → 统计分析 → 可视化 → 落盘"的完整数据工程链路，全部只依赖 Python 标准库：

from crewai import Agent, Task, Crew, LLM from crewai_tools.aws import create_code_interpreter_toolkit # Create toolkit and tools toolkit, code_tools = create_code_interpreter_toolkit(region="us-west-2") # Create the Bedrock LLM llm = LLM( model="bedrock/us.anthropic.claude-3-7-sonnet-20250219-v1:0", region_name="us-west-2", ) # Create a data analyst agent analyst_agent = Agent( role="Data Analyst", goal="Analyze data using Python", backstory="You're an expert data analyst who uses Python for data processing.", tools=code_tools, llm=llm ) # Create a task for the agent analysis_task = Task( description=""" For all of the below, do not use any imports from outside the Python standard library. 1. Create a sample dataset with random data 2. Perform statistical analysis on the dataset 3. Generate visualizations of the results 4. Save the results and visualizations to files """, agent=analyst_agent ) # Create and run the crew crew = Crew( agents=[analyst_agent], tasks=[analysis_task] ) result = crew.kickoff() print(f"\n***Final result:***\n\n{result}") # Clean up resources import asyncio asyncio.run(toolkit.cleanup())

注意第 4 步"保存结果与可视化到文件"：Agent 会通过write_files（或在代码中写文件）把产物留在沙箱文件系统中，之后可以随时用read_files取回分析结论，或用list_files检查目录结构。而第 3 步生成的图表若以资源形式返回，extract_output_from_stream()会将其格式化为==== File: <uri> ====文本块，让 LLM 能"看到"文件路径与内容。

七、资源清理：防止云端会话泄漏

原 README 特别强调"始终在用完时清理资源"。对照源码 cleanup() 的实现，其行为是：

传入thread_id：只stop()并移除该线程对应的会话，失败仅记录 warning 不抛异常；
不传参数（None）：遍历所有已创建的会话逐个stop()，最后清空整个_code_interpreters字典。

import asyncio # Clean up all code interpreter sessions asyncio.run(toolkit.cleanup())

由于 Code Interpreter 是远程托管会话，不清理就意味着持续计费与占用的沙箱实例。建议将其放进脚本的try/finally或atexit中，保证异常路径也能回收。此外每个工具_run的实现都捕获了异常并返回"Error executing code: ..."这类字符串形式的错误信息而非抛出异常——这对 Agent 友好（错误会进入对话上下文供模型自我修正），但也意味着程序层面不应假设"工具返回非空字符串 = 执行成功"，重要场景应在 Agent 结果中校验关键输出。

八、关键参数速查与最佳实践

参数/方法	默认值	说明
`create_code_interpreter_toolkit(region=...)`	`"us-west-2"`	必须与 AgentCore 资源所在区域一致
`thread_id`（所有工具）	`"default"`	会话隔离键；同 ID 共享上下文与文件系统
`language`（execute_code）	`"python"`	代码语言标识
`clear_context`（execute_code）	`False`	`True`时清空该线程执行上下文，适合切换任务场景
`directory_path`（list_files）	`""`（根目录）	要列出的沙箱内目录
`toolkit.cleanup(thread_id=None)`	全部会话	按需清理单个或全部会话
`toolkit.get_tools()`/`get_tools_by_name()`	—	取工具列表 / 按名取工具字典

最佳实践汇总：

区域对齐：toolkit的region与LLM(region_name=...)通常应指向同一区域，避免跨区权限问题；
懒加载特性可利用：创建工具包本身零云端开销，适合在应用启动时预构建、按需触发的架构；
多 Agent 分工：按角色用get_tools_by_name()装配最小工具集；需要并行且互不污染的工作区时用不同thread_id；
错误即上下文：工具失败会返回错误字符串供 Agent 自我重试，编写 Task 时可提示模型"如果执行报错，先检查依赖与语法再重试"；
收尾必清理：asyncio.run(toolkit.cleanup())作为脚本最后一行（或finally块）调用。

九、适用边界说明

结合仓库现状补充两点边界：

从源码结构看，本模块的会话管理完全委托给bedrock-agentcore包的CodeInterpreter客户端（在 code_interpreter_toolkit.py 中于首次使用时延迟导入），CrewAI 侧只封装工具协议与线程会话映射，因此 AgentCore 会话本身的配额、超时与计费规则以 AWS 侧配置为准；
在lib/crewai-tools/tests/目录中未发现针对该模块的专用单元测试，集成验证以 README 中的三个端到端示例为主要参照；该功能属于crewai-tools的bedrock可选依赖集，未安装bedrock-agentcore时相关导入会失败，按需安装即可。

核心文件索引：README ｜核心实现｜ aws 包导出｜依赖声明

创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考