PAI-Instant Service は、E2B オープンソース SDK インターフェースとネイティブに互換性があります。E2B ベースのエージェントアプリケーションを移行する際は、E2B_DOMAIN と E2B_API_KEY の 2 つの環境変数を変更するだけで、コードの変更は不要です。
SDK のインストール
E2B Python SDK の、v2.25.0 より前のバージョンをインストールします。
pip install "e2b<2.25.0" Code Interpreter や Desktop などのシナリオ固有の機能については、対応する拡張パッケージをインストールします。
pip install e2b_code_interpreter # Code Interpreter シナリオ用
pip install e2b_desktop # Desktop シナリオ用環境変数の設定
import os
os.environ["E2B_DOMAIN"] = "sandbox01.cn-shanghai.pai-eas.aliyuncs.com"
os.environ["E2B_API_KEY"] = "<YOUR-API-TOKEN>" 説明
テンプレートの詳細ページで[SDK Integration]をクリックし、E2B_DOMAIN と E2B_API_KEY の値を取得してください。
コア API
|
カテゴリ |
操作 |
Python SDK |
|
ライフサイクル |
インスタンスの作成 |
|
|
既存のインスタンスへの接続 |
|
|
|
一時停止/再開 |
|
|
|
インスタンスの強制終了 |
|
|
|
TTL 制御 |
|
|
|
コマンド実行 |
同期実行 |
|
|
ファイルシステム |
読み取り/書き込み |
|
サンプル
#!/usr/bin/env python
"""
E2B SDK テンプレート用テストデモ
このデモでは、E2B SDK を使用して、すべてのサンドボックス機能
(PTY ターミナルを除く) をテストする方法を示します。
実行前の準備：
1. e2b SDK をインストール: pip install "e2b<2.25.0"
"""
from e2b import Sandbox
import os
import time
import httpx
# 環境変数を設定
os.environ["E2B_DOMAIN"] = "<YOUR-DOMAIN>" # 例：sandbox01.cn-shanghai.pai-eas.aliyuncs.com
os.environ["E2B_API_KEY"] = "<YOUR-API-TOKEN>"
print("=" * 60)
print("E2B サンドボックステストデモ")
print("=" * 60)
# サンドボックスを作成
print("\n[1] テンプレート「code-interpreter」でサンドボックスを作成中")
sandbox = Sandbox.create(
template="code-interpreter",
timeout=300, # 5 分間のタイムアウト
)
print(f" ✓ サンドボックスが作成されました (ID: {sandbox.sandbox_id})")
# サンドボックスの準備完了を待機
print("\n[2] サンドボックスの準備完了を待機中...")
max_attempts = 30
for attempt in range(1, max_attempts + 1):
try:
is_running = sandbox.is_running()
status = "実行中" if is_running else "停止中"
print(f" [{attempt}/{max_attempts}] サンドボックスステータス： {status}")
if is_running:
print(" ✓ サンドボックスが実行中になりました！")
break
except Exception as e:
print(f" [{attempt}/{max_attempts}] ステータスの確認に失敗しました： {e}")
if attempt < max_attempts:
time.sleep(2)
# 基本的なコマンド実行をテスト
print("\n[3] 基本的なコマンド実行をテスト中 (SDK)...")
try:
result = sandbox.commands.run("echo 'Hello from sandbox!'")
if "Hello from sandbox!" in result.stdout:
print(" ✓ コマンド実行が動作します！")
print(f" 出力： {result.stdout.strip()}")
else:
print(f" ✗ 予期しない出力： {result.stdout}")
except Exception as e:
print(f" ✗ コマンド実行に失敗しました： {e}")
# ファイルシステム操作をテスト
print("\n[4] ファイルシステム操作をテスト中 (SDK)...")
try:
test_content = "Hello from filesystem test!"
sandbox.files.write("/tmp/test.txt", test_content)
print(" ✓ ファイルが正常に書き込まれました")
content = sandbox.files.read("/tmp/test.txt")
if content == test_content:
print(" ✓ ファイルシステムの読み取りが動作します！")
files = sandbox.files.list("/tmp")
file_names = [f.name for f in files]
if "test.txt" in file_names:
print(" ✓ ディレクトリの一覧表示が動作します！")
sandbox.files.remove("/tmp/test.txt")
print(" ✓ クリーンアップが完了しました")
except Exception as e:
print(f" ✗ ファイルシステムテストに失敗しました： {e}")
# Pythonコード実行をテスト
print("\n[5] Python コードの実行をテスト中 (SDK)...")
try:
result = sandbox.commands.run(
"python -c \"import sys; print('Python version:', sys.version)\"",
timeout=10
)
if "Python version:" in result.stdout:
print(" ✓ Python の実行が動作します！")
print(f" {result.stdout.strip()[:60]}...")
else:
print(" ✗ Python が利用できない可能性があります")
except Exception as e:
print(f" ✗ Python テストに失敗しました： {e}")
# 直接API経由でenvdヘルスエンドポイントをテスト
print("\n[6] 直接 API 経由で envd ヘルスエンドポイントをテスト中 (直接 API)...")
try:
envd_token = getattr(sandbox, '_SandboxBase__envd_access_token', None)
if envd_token:
envd_url = f"https://49983-{sandbox.sandbox_id}.sandbox01.cn-shanghai.pai-eas.aliyuncs.com"
headers = {
"Content-Type": "application/json",
"X-Access-Token": envd_token
}
resp = httpx.get(f"{envd_url}/health", headers=headers, timeout=5)
if resp.status_code in [200, 204]:
print(f" ✓ ヘルスチェックに成功しました！ (ステータス： {resp.status_code})")
else:
print(f" ✗ ヘルスチェックに失敗しました： {resp.status_code}")
else:
print(" ✗ サンドボックスオブジェクトに envd_access_token が見つかりません")
except Exception as e:
print(f" ✗ ヘルスチェックに失敗しました： {e}")
# データプレーンアクセスをテスト
print("\n[7] データプレーンアクセスをテスト中...")
try:
# Flaskをインストール
print(" Flask をインストール中...")
result = sandbox.commands.run("python -m pip install flask -q -i https://mirrors.cloud.aliyuncs.com/pypi/simple/ --trusted-host=mirrors.cloud.aliyuncs.com", timeout=60)
if result.exit_code == 0:
print(" ✓ Flask がインストールされました")
# Flaskアプリを作成
flask_code = '''
from flask import Flask
app = Flask(__name__)
@app.route('/sandbox_test')
def hello():
return 'hello sandbox'
if __name__ == '__main__':
app.run(host='0.0.0.0', port=50001)
'''
sandbox.files.write("/tmp/flask_app.py", flask_code)
print(" ✓ Flask アプリが作成されました")
# Flaskサーバーを起動
print(" ポート 50001 で Flask サーバーを起動中...")
sandbox.commands.run("nohup python /tmp/flask_app.py > /tmp/flask.log 2>&1 &", timeout=5)
time.sleep(2)
# サーバーが実行中か確認
result = sandbox.commands.run("pgrep -f flask_app.py || true", timeout=5)
if result.stdout.strip():
print(" ✓ Flask サーバーが起動しました")
# データプレーン経由でアクセス
envd_token = getattr(sandbox, '_SandboxBase__envd_access_token', None)
if envd_token:
data_plane_url = f"https://50001-{sandbox.sandbox_id}.sandbox01.cn-shanghai.pai-eas.aliyuncs.com/sandbox_test"
print(f" データプレーン URL にアクセス中...")
headers = {"X-Access-Token": envd_token}
max_retries = 5
for attempt in range(max_retries):
try:
resp = httpx.get(data_plane_url, headers=headers, timeout=10)
if resp.status_code == 200 and "hello sandbox" in resp.text:
print(" ✓ データプレーンアクセスに成功しました！")
print(f" レスポンス： {resp.text.strip()}")
break
except Exception as e:
if attempt < max_retries - 1:
time.sleep(1)
else:
print(f" ✗ データプレーンアクセスに失敗しました： {e}")
else:
print(" ✗ envd_access_token が見つかりません")
except Exception as e:
print(f" ✗ データプレーンテストに失敗しました： {e}")
# サンドボックスステータスを確認
print("\n[8] サンドボックスステータスをテスト中...")
try:
is_running = sandbox.is_running()
status = "実行中" if is_running else "停止中"
print(f" サンドボックス実行ステータス： {status}")
except Exception as e:
print(f" ステータスの確認に失敗しました： {e}")
# サマリー
print("\n" + "=" * 60)
print("✓ すべてのテストが完了しました！")
print("=" * 60)
# サンドボックスを強制終了
print("\nサンドボックスを強制終了中...")
sandbox.kill()
print("✓ サンドボックスが強制終了されました")