
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-agentcore1.18.1,2.0.0同时附带playwright、nest-asyncio、beautifulsoup4这些是 Bedrock 浏览器工具等整个 extra 共用的依赖。安装命令来自原 READMEuv add crewai-tools bedrock-agentcore使用前提原 README Requirements 部分拥有可访问Bedrock AgentCore API的 AWS 账户正确配置了 AWS 凭据标准 AWS 凭据链即可如环境变量、~/.aws/credentials、IAM Roleregion参数需与你的 AgentCore 资源所在区域一致所有示例默认使用us-west-2。三、九大内置工具与底层方法映射工具包共提供 9 个工具。下表把每个工具的名称、输入参数与它实际调用的 Bedrock AgentCore 方法对应起来依据 code_interpreter_toolkit.py 中各_run实现的invoke(method..., params...)调用工具名 (name)输入参数对应 AgentCore 方法说明execute_codecode、language默认python、clear_context默认False、thread_idexecuteCode执行代码clear_contextTrue时清空执行上下文execute_commandcommand、thread_idexecuteCommand在沙箱内运行 Shell 命令read_filespaths: list[str]、thread_idreadFiles批量读取文件内容list_filesdirectory_path默认、thread_idlistFiles列出目录内容delete_filespaths: list[str]、thread_idremoveFiles批量删除文件注意方法名与工具名不同write_filesfiles: list[dict]含path/text字段、thread_idwriteFiles创建或更新文件参数字段名为contentstart_command_executioncommand、thread_idstartCommandExecution异步启动长时命令get_tasktask_id、thread_idgetTask查询异步任务状态stop_tasktask_id、thread_idstopTask停止运行中的任务两个值得注意的实现细节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(regionus-west-2) # Create the Bedrock LLM llm LLM( modelbedrock/us.anthropic.claude-3-7-sonnet-20250219-v1:0, region_nameus-west-2, ) # Create a CrewAI agent that uses the code interpreter tools developer_agent Agent( rolePython Developer, goalCreate and execute Python code to solve problems., backstoryYoure a skilled Python developer with expertise in data analysis., toolscode_tools, llmllm ) # Create a task for the agent coding_task Task( descriptionWrite 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_outputThe Python function created, and the test results., agentdeveloper_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(regionus-west-2)返回二元组(toolkit, tools)。注意此时并不会创建任何远程会话——源码_setup_tools()只实例化 9 个工具对象真正的CodeInterpreter会话直到第一次工具调用时才由_get_or_create_interpreter()创建内部调用CodeInterpreter(regionself.region)并start()。这种懒加载意味着即使创建了工具包但 Agent 从未触发代码执行也不会产生任何云端会话开销toolscode_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(regionus-west-2) tools_by_name toolkit.get_tools_by_name() # Create the Bedrock LLM llm LLM( modelbedrock/us.anthropic.claude-3-7-sonnet-20250219-v1:0, region_nameus-west-2, ) # Create agents with specific tools code_agent Agent( roleCode Developer, goalWrite and execute code, backstoryYou 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] ], llmllm ) file_agent Agent( roleFile Manager, goalManage files in the environment, backstoryYou 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] ], llmllm ) # Create tasks for the agents coding_task Task( descriptionWrite a Python script to analyze data from a CSV file. Do not use any imports from outside the Python standard library., expected_outputThe Python function created., agentcode_agent ) file_task Task( descriptionOrganize the created files into separate directories., agentfile_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_iddefault因此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(regionus-west-2) # Create the Bedrock LLM llm LLM( modelbedrock/us.anthropic.claude-3-7-sonnet-20250219-v1:0, region_nameus-west-2, ) # Create a data analyst agent analyst_agent Agent( roleData Analyst, goalAnalyze data using Python, backstoryYoure an expert data analyst who uses Python for data processing., toolscode_tools, llmllm ) # 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 , agentanalyst_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 共享上下文与文件系统languageexecute_codepython代码语言标识clear_contextexecute_codeFalseTrue时清空该线程执行上下文适合切换任务场景directory_pathlist_files根目录要列出的沙箱内目录toolkit.cleanup(thread_idNone)全部会话按需清理单个或全部会话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 包导出 依赖声明【免费下载链接】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创作声明:本文部分内容由AI辅助生成(AIGC),仅供参考