PythonとE2BでAIエージェント用セキュア実行環境を作る方法

pip install e2b-code-interpreter anthropic python-dotenv

ANTHROPIC_API_KEY=sk-ant-xxx...
E2B_API_KEY=e2b_xxx...

import os
from dotenv import load_dotenv
from e2b_code_interpreter import Sandbox

# 環境変数の読み込み
load_dotenv()

def test_sandbox():
    # クラウド上に隔離環境（サンドボックス）を作成
    # デフォルトでUbuntuベースの環境が立ち上がる
    with Sandbox(api_key=os.getenv("E2B_API_KEY")) as sandbox:
        print("サンドボックスを起動しました...")

        # サンドボックス内でPythonコードを実行
        # なぜこれが必要か：LLMが書いたコードをこのrun_pythonに渡すため
        code = "print('Hello from E2B!')"
        execution = sandbox.run_python(code)

        if execution.error:
            print(f"エラー発生: {execution.error}")
        else:
            print(f"実行結果: {execution.results[0] if execution.results else 'なし'}")
            print(f"標準出力: {execution.logs.stdout}")

if __name__ == "__main__":
    test_sandbox()

import os
from dotenv import load_dotenv
from anthropic import Anthropic
from e2b_code_interpreter import Sandbox

load_dotenv()

# クライアントの初期化
client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
E2B_API_KEY = os.getenv("E2B_API_KEY")

# Claudeに教える「道具」の定義
# どんな時にこの道具を使うべきか、詳細に説明文を書くのがAIを賢く動かすコツ
tools = [
    {
        "name": "execute_python",
        "description": "Pythonコードを実行して、データ分析やグラフ作成、計算を行う。結果や生成された画像を取得できる。",
        "input_schema": {
            "type": "object",
            "properties": {
                "code": {
                    "type": "string",
                    "description": "実行するPythonコード。必要なライブラリのimportも含めること。"
                }
            },
            "required": ["code"]
        }
    }
]

def run_agent(prompt):
    # メッセージの送信
    response = client.messages.create(
        model="claude-3-5-sonnet-20240620",
        max_tokens=2000,
        tools=tools,
        messages=[{"role": "user", "content": prompt}]
    )

    print(f"Claudeの思考: {response.content[0].text if response.content[0].type == 'text' else 'ツール使用中'}")

    # Claudeがツールを使いたいと言っているかチェック
    for content in response.content:
        if content.type == "tool_use" and content.name == "execute_python":
            code = content.input["code"]
            print(f"--- 実行されるコード ---\n{code}\n-------------------------")

            # E2Bでコード実行
            with Sandbox(api_key=E2B_API_KEY) as sandbox:
                execution = sandbox.run_python(code)

                if execution.error:
                    return f"実行エラー: {execution.error}"

                # 実行結果を返す（今回は簡単のため標準出力のみ）
                return f"実行成功。出力: {execution.logs.stdout}"

if __name__ == "__main__":
    result = run_agent("1から100までの素数を計算して、その合計を教えてください。")
    print(result)

Claudeの思考: 1から100までの素数を計算するために、Pythonコードを実行します。
--- 実行されるコード ---
def is_prime(n):
    if n < 2: return False
    for i in range(2, int(n**0.5) + 1):
        if n % i == 0: return False
    return True

primes = [n for n in range(1, 101) if is_prime(n)]
print(sum(primes))
-------------------------
実行成功。出力: ['1060\n']

import os
import base64
from dotenv import load_dotenv
from anthropic import Anthropic
from e2b_code_interpreter import Sandbox

load_dotenv()

class DataAgent:
    def __init__(self):
        self.anthropic = Anthropic()
        self.e2b_api_key = os.getenv("E2B_API_KEY")

    def process_query(self, user_query):
        print(f"ユーザーの依頼: {user_query}")

        # グラフ作成が含まれる可能性が高いので、予めsandboxを用意
        with Sandbox(api_key=self.e2b_api_key) as sandbox:
            # メッセージ履歴の初期化
            messages = [{"role": "user", "content": user_query}]

            # 最大3回までやり取りを許可（ループ防止とエラーリトライのため）
            for _ in range(3):
                response = self.anthropic.messages.create(
                    model="claude-3-5-sonnet-20240620",
                    max_tokens=4000,
                    tools=tools, # Step 3で定義したもの
                    messages=messages
                )

                # Claudeの返答を履歴に追加
                messages.append({"role": "assistant", "content": response.content})

                # ツール使用の判定
                tool_use = next((c for c in response.content if c.type == "tool_use"), None)

                if not tool_use:
                    # ツールを使わずに返答してきたら終了
                    return response.content[0].text

                if tool_use.name == "execute_python":
                    code = tool_use.input["code"]
                    print("AIがコードを実行中...")

                    execution = sandbox.run_python(code)

                    if execution.error:
                        # エラー内容をAIにフィードバックして修正を促す
                        tool_result = f"エラーが発生しました: {execution.error.value}\n{execution.error.traceback}"
                    else:
                        # 成功した場合、標準出力と生成されたファイルを確認
                        output = "".join(execution.logs.stdout)

                        # もし画像（グラフ）が生成されていたら保存
                        if execution.results:
                            for idx, res in enumerate(execution.results):
                                # pngなどの画像データがあるかチェック
                                if res.png:
                                    filename = f"chart_{idx}.png"
                                    # base64デコードして保存
                                    with open(filename, "wb") as f:
                                        f.write(base64.b64decode(res.png))
                                    output += f"\n[画像 {filename} を保存しました]"

                        tool_result = output if output else "実行完了（出力なし）"

                    # 実行結果を履歴に追加してループ継続
                    messages.append({
                        "role": "user",
                        "content": [
                            {
                                "type": "tool_result",
                                "tool_use_id": tool_use.id,
                                "content": tool_result
                            }
                        ]
                    })

if __name__ == "__main__":
    agent = DataAgent()
    # 複雑な依頼
    final_ans = agent.process_query("1から10までのxに対して、y=x^2のグラフを描画して保存してください。")
    print(f"\n最終回答: {final_ans}")