Skip to main content
Amazon Bedrock AgentCore Code Interpreter 使智能体能够在安全的托管沙箱环境中执行代码。智能体可以运行 Python、JavaScript 和 TypeScript 代码，用于计算、数据分析、文件操作和可视化。

​
概述

​
集成详情

类包可序列化JS 支持版本
CodeInterpreterToolkitlangchain-aws✅❌PyPI - Version

​
工具特性

返回制品原生异步返回数据定价
✅✅文本、文件、图像按需计费 (AWS)

​
可用工具

该工具包提供了多种代码执行和文件管理工具：
工具描述
execute_code运行具有持久状态的 Python/JavaScript/TypeScript 代码
execute_command在环境中运行 Shell 命令
read_files读取环境中的文件内容
write_files创建或更新文件
list_files列出目录中的文件
delete_files从环境中删除文件
upload_file上传带有语义描述的文件
install_packages安装 Python 包
start_command_execution异步启动长时间运行的命令
get_task通过 task_id 检查异步任务的状态
stop_task通过 task_id 停止正在运行的异步任务

​
设置

该集成位于 langchain-aws 包中，它封装了 bedrock-agentcore SDK。 
pip install -U langchain-aws bedrock-agentcore

​
凭证

您需要配置具有 Bedrock AgentCore Code Interpreter 权限的 AWS 凭证。有关所需 IAM 权限，请参阅 Amazon Bedrock AgentCore 文档。 同时建议（但非必需）设置 LangSmith 以获得一流的可观测性： 
import os

os.environ["LANGSMITH_API_KEY"] = "your-api-key"
os.environ["LANGSMITH_TRACING"] = "true"

​
实例化

工具包使用异步工厂函数创建： 
from langchain_aws.tools import create_code_interpreter_toolkit

# Create toolkit and get tools (async)
toolkit, code_tools = await create_code_interpreter_toolkit(region="us-west-2")

​
调用

​
直接使用工具

获取特定工具并调用它们： 
# Get tools by name
tools_by_name = toolkit.get_tools_by_name()

# Execute Python code
result = tools_by_name["execute_code"].invoke({
    "code": """
import numpy as np
data = [1, 2, 3, 4, 5]
print(f"Mean: {np.mean(data)}")
print(f"Sum: {np.sum(data)}")
""",
    "language": "python"
})
print(result)

​
在智能体中使用

import asyncio
from langchain.agents import create_react_agent
from langchain.chat_models import init_chat_model
from langchain_aws.tools import create_code_interpreter_toolkit

async def main():
    # Create toolkit
    toolkit, code_tools = await create_code_interpreter_toolkit(region="us-west-2")

    # Initialize chat model
    llm = init_chat_model(
        "us.anthropic.claude-sonnet-4-20250514-v1:0",
        model_provider="bedrock_converse",
    )

    # Create agent with code interpreter tools
    agent = create_react_agent(
        model=llm,
        tools=code_tools,
    )

    # Create config with thread_id for session isolation
    config = {"configurable": {"thread_id": "session-123"}}

    # Run the agent
    result = await agent.ainvoke(
        {"messages": [{"role": "user", "content": "Calculate the factorial of 10"}]},
        config=config
    )
    print(result["messages"][-1].content)

    # Clean up when done
    await toolkit.cleanup()

asyncio.run(main())

​
基于线程的会话隔离

工具包通过 thread_id 支持多个并发会话。每个线程维护自己的代码解释器会话，状态相互隔离： 
# Different threads have isolated sessions
config_user1 = {"configurable": {"thread_id": "user-1"}}
config_user2 = {"configurable": {"thread_id": "user-2"}}

# Variables defined in user-1's session won't exist in user-2's session
await agent.ainvoke(
    {"messages": [{"role": "user", "content": "Set x = 100"}]},
    config=config_user1
)

await agent.ainvoke(
    {"messages": [{"role": "user", "content": "What is x?"}]},  # x is undefined here
    config=config_user2
)

​
文件操作

​
写入和读取文件

tools_by_name = toolkit.get_tools_by_name()

# Write a file
tools_by_name["write_files"].invoke({
    "files": [{"path": "data.csv", "text": "name,value\nAlice,100\nBob,200"}]
})

# Read it back
content = tools_by_name["read_files"].invoke({"paths": ["data.csv"]})
print(content)

# List files in current directory
files = tools_by_name["list_files"].invoke({"directory_path": "."})
print(files)

​
上传带描述的文件

tools_by_name["upload_file"].invoke({
    "path": "sales.csv",
    "content": "date,revenue,product\n2024-01-01,1000,Widget\n2024-01-02,1500,Gadget",
    "description": "Sales data with columns: date, revenue, product_id"
})

​
安装包

tools_by_name["install_packages"].invoke({
    "packages": ["pandas>=2.0", "matplotlib", "scikit-learn"],
    "upgrade": False
})

​
异步任务管理

对于长时间运行的命令，您可以异步启动它们并检查其状态： 
tools_by_name = toolkit.get_tools_by_name()
config = {"configurable": {"thread_id": "session-123"}}

# Start a long-running command asynchronously
result = tools_by_name["start_command_execution"].invoke(
    {"command": "python long_running_script.py"},
    config=config
)
# Returns a task_id

# Check task status
status = tools_by_name["get_task"].invoke(
    {"task_id": "task-abc123"},
    config=config
)
print(status)

# Stop a running task if needed
tools_by_name["stop_task"].invoke(
    {"task_id": "task-abc123"},
    config=config
)

​
会话清理

完成后请务必清理会话以释放资源： 
# Clean up all sessions
await toolkit.cleanup()

# Or clean up a specific thread's session
await toolkit.cleanup(thread_id="session-123")

​
API 参考

有关所有功能和配置的详细文档，请参阅：
Edit this page on GitHub or file an issue.
Connect these docs to Claude, VSCode, and more via MCP for real-time answers.