IBM Bob Shell を IBM Cloud Code Engine で REST API として動かしてみた

はじめに

IBM Bob は IBM が提供する AI コーディングアシスタントで、VS Code 拡張の Bob IDE と CLI 版の Bob Shell の 2 形態を提供しています。

Bob IDE については利用者が多いものの、Bob Shell はまだ少ない印象です。しかし、Bob Shell はターミナルで叩ける AI エージェントとして、スクリプトや Web アプリのバックエンドから subprocess 経由で呼び出せるのが利点の一つかと思います。

この記事では、Bob Shell を Flask の REST API でラップし、IBM Cloud Code Engine 上で動かすまでの実装を記載します。

---

Bob Shell とは

Bob IDE との違い

同じ AI エンジンを共有しつつ、UI と用途が大きく異なります。

観点	Bob IDE	Bob Shell
形態	VS Code 拡張	ターミナル CLI(bob コマンド)
セッション	対話のみ	対話 + 非対話の 2 系統
ファイル参照	開いているファイルが自動でコンテキストに入る	@path/to/file で明示参照
認証	IBMid ブラウザログイン	非対話モードは API キー必須
書き込み権限	承認フロー	デフォルトは read-only、--yolo で許可
CI/CD 組み込み	困難	可能
subprocess 呼び出し	困難	可能(本記事の主題)
コンテナ化	困難	可能

モード

Bob Shell には標準で 4 つのモードがあり、デフォルトは Code モード(コード変更可能、コマンド実行可能)です。

モード	用途
code(デフォルト)	コード生成・修正・リファクタリング
ask	コードベースに関する質問・調査(読み取りのみ)
plan	実装前の設計・計画立案
advanced	MCP ツール含む拡張機能

---

ローカルへのインストール

1. システム要件

• macOS / Linux / Windows
• Node.js 22.15.0 以上
• 4 GB RAM 以上
• 500 MB 以上の空き容量

2. インストール

公式インストールスクリプトを実行:

# macOS / Linux
curl -fsSL https://bob.ibm.com/download/bobshell.sh | bash

# Windows (PowerShell)
powershell -ep Bypass 'irm -Uri "https://bob.ibm.com/download/bobshell.ps1" | iex'

PATH 反映:

# zsh の場合
source ~/.zshrc

# 確認
bob --version

3. 初回起動と認証

cd ~/your-project
bob

ブラウザが開いて IBMid 認証が走ります。認証後、対話セッションでプロンプトを試せます:

> Explain the architecture of this project

---

API キーの取得

非対話モード(bob -p "...")では、ブラウザ認証ではなく API キー認証が必須です。

取得手順

1. Bob のダッシュボード( https://bob.ibm.com/ )にブラウザでアクセス
2. 設定画面から API キーを発行
3. 表示された API キー文字列(sk-... のような形式)をコピー

環境変数として設定

# .env ファイルに保存
cat > .env << 'EOF'
BOBSHELL_API_KEY=sk-xxxxxxxxxxxxxxx
EOF

# .gitignore に追加(必須)
echo '.env' >> .gitignore
chmod 600 .env

BOBSHELL_API_KEY が設定されていない状態で bob -p "..." を実行すると次のエラーになります:

API key required. Set BOBSHELL_API_KEY via environment variable or .env file.

非対話モードの基本コマンド

# 基本(読み取りのみ)
bob -p "Explain this project"

# ファイル参照
bob -p "Review @src/api.py and suggest improvements"

# 標準入力からパイプ
cat error.log | bob -p "Explain this build error"

# 書き込み・コマンド実行を許可
bob -p "Fix bugs in @app.py" --yolo

---

Dockerfile の書き方

Bob Shell を Flask アプリと一緒にコンテナ化します。ポイント:

• Node.js 22.x が必要(Bob Shell の前提)
• /root/.bob/bin を PATH に通す
• オプション: zskills などの MCP ツールは wheel 経由で持ち込む(後述)

完成版 Dockerfile

FROM python:3.11-slim-bookworm
WORKDIR /app
COPY . /app

# 基本パッケージ
RUN apt update
RUN apt install -y wget
RUN apt install -y jq procps

# Node.js 22.x と Bob Shell のインストール
RUN apt install -y curl && \
    curl -fsSL https://deb.nodesource.com/setup_22.x | bash - && \
    apt install -y nodejs
RUN curl -fsSL https://bob.ibm.com/download/bobshell.sh | bash
ENV PATH="/root/.bob/bin:${PATH}"

# uv 経由で MCP ツール(例: zskills)を wheel から導入
RUN curl -LsSf https://astral.sh/uv/install.sh | bash
ENV PATH="/root/.local/bin:${PATH}"
COPY wheels/zskills-*.whl /tmp/
RUN uv tool install /tmp/zskills-*.whl && rm /tmp/zskills-*.whl

# Bob 作業ディレクトリと .bob 設定の配置
RUN mkdir -p /app/bob-simple
RUN mkdir -p /app/bob-zskill-workspace
COPY .bob /app/bob-zskill-workspace/.bob

# Python パッケージ
RUN pip install --upgrade pip setuptools wheel
RUN pip install -r requirements.txt

EXPOSE 5001
CMD python server.py

この Dockerfile では .bob (skills の md や mcp.json などを格納) がないまっさらなプロジェクトフォルダである /app/bob-simple と .bob や必要なフォルダを配置して特定の skill を実行させる /app/bob-zskill-workspace というプロジェクトフォルダを用意します。
また、Bob Shell が利用する MCP として zskills というスキルをインストールしようとしています。

---

Flask での Bob Shell 呼び出し実装

ライセンス同意の初期化

非対話モードを初めて実行する際にライセンス同意プロンプトで止まることがあるため、Flask 起動時に --accept-license 付きで一度実行して同意マーカーを作成しておきます。

import os
import subprocess
import logging
from pathlib import Path

logger = logging.getLogger(__name__)


def initialize_bob_shell(cwd: str) -> bool:
    """Flask 起動時に Bob Shell の初回ライセンス同意を済ませる。"""
    if not os.environ.get("BOBSHELL_API_KEY"):
        logger.warning("BOBSHELL_API_KEY is not set.")
        return False

    bob_dir = Path.home() / ".bob"
    license_marker = bob_dir / ".license-accepted"
    if license_marker.exists():
        return True  # 既に同意済み

    try:
        result = subprocess.run(
            ["bob", "--accept-license", "-p", "Hello"],
            cwd=str(cwd),
            capture_output=True, text=True, timeout=120, check=False,
            env=os.environ.copy(),
        )
        if result.returncode == 0:
            bob_dir.mkdir(parents=True, exist_ok=True)
            license_marker.touch()
            return True
    except Exception as e:
        logger.error(f"Bob Shell init failed: {e}")
    return False

シンプルな非ストリーミング実行

def run_bob(
    prompt: str,
    cwd: str,
    allow_write: bool = False,
    timeout: int = 600,
) -> dict:
    """Bob Shell を非対話モードで実行(ブロッキング)。"""
    if not os.environ.get("BOBSHELL_API_KEY"):
        return {"stdout": "", "stderr": "BOBSHELL_API_KEY is not set", "returncode": -1}

    cmd = ["bob", "-p", prompt]
    if allow_write:
        cmd.append("--yolo")

    try:
        result = subprocess.run(
            cmd,
            cwd=str(cwd),
            capture_output=True, text=True, timeout=timeout,
            check=False, env=os.environ.copy(),
        )
        return {
            "stdout": result.stdout,
            "stderr": result.stderr,
            "returncode": result.returncode,
        }
    except subprocess.TimeoutExpired:
        return {"stdout": "", "stderr": f"Timeout after {timeout}s", "returncode": -1}

SSE ストリーミング実行

Bob Shell の標準出力には 思考プロセス(<thinking>...</thinking>)、ツール呼び出し、出力結果が逐次的に流れます。これを SSE で配信すれば、フロントエンドからリアルタイムで進捗を見せられます。

import json
import time

def run_bob_stream(
    prompt: str,
    cwd: str,
    allow_write: bool = False,
    timeout: int = 600,
):
    """Bob Shell の出力を SSE 形式でストリーミング。"""
    if not os.environ.get("BOBSHELL_API_KEY"):
        yield f"data: {json.dumps({'error': 'BOBSHELL_API_KEY is not set'})}\n\n"
        return

    cmd = ["bob", "-p", prompt]
    if allow_write:
        cmd.append("--yolo")

    process = None
    try:
        process = subprocess.Popen(
            cmd,
            cwd=str(cwd),
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
            bufsize=1,
            env=os.environ.copy(),
        )

        start_time = time.time()
        if process.stdout:
            for line in iter(process.stdout.readline, ''):
                if time.time() - start_time > timeout:
                    yield f"data: {json.dumps({'error': f'Timeout after {timeout}s'})}\n\n"
                    break
                if line:
                    yield f"data: {json.dumps({'data': line.rstrip()})}\n\n"

        returncode = process.wait(timeout=5)
        yield f"data: {json.dumps({'status': 'completed', 'returncode': returncode})}\n\n"

    finally:
        if process and process.poll() is None:
            process.terminate()
            try:
                process.wait(timeout=5)
            except subprocess.TimeoutExpired:
                process.kill()

Flask エンドポイント

from flask import Flask, Response, jsonify, stream_with_context, request

app = Flask(__name__)

@app.route("/api/bob-simple-non-dialog", methods=["POST"])
def bob_simple_non_dialog_endpoint():
    data = request.get_json()
    prompt = data["prompt"]

    return Response(
        stream_with_context(
            run_bob_stream(
                prompt=prompt,
                cwd="/app/bob-simple",
            )
        ),
        content_type='text/event-stream',
        headers={'Cache-Control': 'no-cache'}
    )


@app.route("/api/bob-zskills-non-dialog", methods=["POST"])
def bob_zskills_non_dialog_endpoint():
    data = request.get_json()
    prompt = data["prompt"]

    return Response(
        stream_with_context(
            run_bob_stream(
                prompt=prompt,
                cwd="/app/bob-zskill-workspace",
                allow_write=True
            )
        ),
        content_type='text/event-stream',
        headers={'Cache-Control': 'no-cache'}
    )


if __name__ == '__main__':
    port = int(os.getenv('PORT', 5001))
    # Bob Shell の初期化(ライセンス同意)
    initialize_bob_shell("/app/bob-simple")
    
    from waitress import serve
    serve(app, host="0.0.0.0", port=port)

---

API の実行例

基本(SSE 受信)

curl -X POST https://your-app.code.engine.appdomain.cloud/api/bob-simple-non-dialog \
    -H "Content-Type: application/json" \
    -d '{"prompt": "Hello, what is 2+2?"}'

SSE レスポンス例:

data: {"data": "<thinking>"}
data: {"data": "The user is asking a simple arithmetic question..."}
data: {"data": "</thinking>"}
data: {"data": "[using tool attempt_completion: Successfully completed | Cost: 0.01]"}
data: {"data": "---output---"}
data: {"data": ""}
data: {"data": "2 + 2 = 4"}
data: {"data": ""}
data: {"data": "---output---"}
data: {"status": "completed", "returncode": 0}

---

SSE の課題と Code Engine のセッション切れ

Bob Shell のタスクは数秒で終わるものから、分析系のスキル実行で 10 分以上かかるものまでさまざまです。ここで Code Engine 特有の問題に直面します。

何が起きているか

おそらく、Code Engine の前段にあるミドルウェア層の影響かと思われますが、長時間レスポンスを継続している HTTP 接続が途中で切られることがあります。SSE で全プロセスを最後まで送り切ろうとすると、フロントエンドで「接続が切れた」エラーが出るのに、コンテナ内では Bob のプロセスは生きているという状態になりがちです。

解決策: バックグラウンド実行 + ポーリング

長時間タスクは、HTTP リクエスト自体はすぐにレスポンスを返してバックグラウンドで Bob を走らせ、別エンドポイントでポーリングして結果を取得する設計に切り替えます。

構造

┌──────────┐  POST /api/bob-analysis
│ Frontend │ ─────────────────────────────────────► ┌─────────────┐
│          │ ◄───── { bob_session_id, "started" } ──│ Flask       │
│          │                                        │             │
│          │                                        │ Thread起動  │
│          │                                        │  └─ Bob実行 │
│          │  GET /api/bob-session-messages?id=...  │     ↓       │
│          │ ────────────────────────────────────►  │  メモリに格納 │
│          │ ◄── { status, messages: [...] } ─────  │             │
│          │  (1〜3秒ごとにポーリング)                 │  完了時に     │
│          │                                        │  DB に保存   │
└──────────┘                                        └─────────────┘

サブプロセスをスレッドで走らせる

import threading
import uuid
from datetime import datetime, timezone

# メモリ内のセッションストア(threading.Lock で保護)
bob_sessions: dict = {}
bob_sessions_lock = threading.Lock()


def generate_bob_session_id() -> str:
    timestamp = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S")
    unique_id = str(uuid.uuid4())[:8]
    return f"bob_{timestamp}_{unique_id}"


def _run_bob_subprocess(
    bob_session_id: str,
    prompt: str,
    cwd: str,
    allow_write: bool = False,
    timeout: int = 3600,
) -> None:
    """バックグラウンドで Bob Shell を実行し、出力をメモリに溜める。"""
    cmd = ["bob", "-p", prompt]
    if allow_write:
        cmd.append("--yolo")

    process = subprocess.Popen(
        cmd,
        cwd=cwd,
        stdout=subprocess.PIPE,
        stderr=subprocess.STDOUT,
        text=True,
        bufsize=1,
        env=os.environ.copy(),
    )

    start_time = time.time()
    if process.stdout:
        for line in iter(process.stdout.readline, ''):
            if time.time() - start_time > timeout:
                update_bob_session_status(bob_session_id, "failed", "Timeout")
                break
            if line:
                with bob_sessions_lock:
                    bob_sessions[bob_session_id]["messages"].append(line.rstrip())

    returncode = process.wait(timeout=5)
    status = "completed" if returncode == 0 else "failed"
    update_bob_session_status(bob_session_id, status)
    # 完了時に DB に永続化、メモリから削除
    _save_session_to_cloudant(bob_session_id)


def run_bob_background(
    bob_session_id: str,
    prompt: str,
    cwd: str,
    allow_write: bool = False,
    timeout: int = 3600,
) -> None:
    """daemon スレッドで Bob Shell を実行。"""
    thread = threading.Thread(
        target=_run_bob_subprocess,
        args=(bob_session_id, prompt, cwd, allow_write, timeout),
        daemon=True,
    )
    thread.start()

開始エンドポイント(即レスポンス)

@app.route("/api/bob-analysis", methods=["POST"])
def bob_analysis_endpoint():
    data = request.get_json()
    
    # bob_session_id を即発行
    bob_session_id = generate_bob_session_id()
    
    # メモリにセッション作成
    with bob_sessions_lock:
        bob_sessions[bob_session_id] = {
            "status": "running",
            "messages": [],
            "created_at": datetime.now(timezone.utc).isoformat(),
            "metadata": data,
        }
    
    # バックグラウンドで Bob 実行を開始
    run_bob_background(
        bob_session_id=bob_session_id,
        prompt=build_prompt_from(data),
        cwd="/app/bob-zskill-workspace",
        allow_write=True,
    )
    
    # 即レスポンスを返す(セッション ID だけ)
    return jsonify({
        "bob_session_id": bob_session_id,
        "status": "started",
    })

ポーリングエンドポイント

@app.route("/api/bob-session-messages", methods=["GET"])
def bob_session_messages_endpoint():
    bob_session_id = request.args.get('bob_session_id')
    
    with bob_sessions_lock:
        session = bob_sessions.get(bob_session_id)
    
    if not session:
        # メモリに無ければ Cloudant から取得
        session = load_bob_session_from_cloudant(bob_session_id)
    
    if not session:
        return jsonify({"error": "Session not found"}), 404
    
    return jsonify(session)

フロントエンドからの呼び出し例

// 1. タスク開始
const { bob_session_id } = await fetch('/api/bob-session-messages', {
  method: 'POST',
  headers: { 'Content-Type': 'application/json' },
  body: JSON.stringify({ save_process_id, no, env_id, type: 'z/OS' }),
}).then(r => r.json());

// 2. ポーリング(2秒間隔)
const pollInterval = setInterval(async () => {
  const session = await fetch(
    `/api/bob-session-messages?bob_session_id=${bob_session_id}`
  ).then(r => r.json());
  
  // メッセージを画面に追加(差分のみ反映)
  updateMessages(session.messages);
  
  if (session.status !== 'running') {
    clearInterval(pollInterval);
    showResult(session);
  }
}, 2000);

---

Bob プロジェクトを Object Storage で外出しする

Code Engine では IBM Cloud Object Storage の bucket をマウントすることができ、ICOS を通して Bob Shell のプロジェクトフォルダにアクセス (read / write) することが可能になります。

Code Engine でのマウント設定

Code Engine 上で Persistent data stores を設定したのちに、アプリケーションのデプロイ構成のvolume mount で data stores を選択しマウントパスを指定します。

---

MCP ツール(zskills など)を wheel 経由で持ち込む

以下のような正規でのインストールコマンドを Dockerfile に記載する方法が失敗しました。

RUN uv tool install git+https://github.ibm.com/griffin/zskills

回避策として、ローカル(認証済みの開発マシン)で wheel をビルドして、リポジトリの vendor/ または wheels/ に置いておきます。

wheel のビルド(ローカル作業)

# 一度クローン
git clone https://github.ibm.com/griffin/zskills.git /tmp/zskills
cd /tmp/zskills

# wheel をビルド
pip install build
python -m build --wheel

# プロジェクトに配置
mkdir -p ~/your-flask-app/wheels
cp dist/zskills-*.whl ~/your-flask-app/wheels/

Dockerfile でコピーしてインストール

RUN curl -LsSf https://astral.sh/uv/install.sh | bash
ENV PATH="/root/.local/bin:${PATH}"

COPY wheels/zskills-*.whl /tmp/
RUN uv tool install /tmp/zskills-*.whl && rm /tmp/zskills-*.whl

コンテナ内での配置

uv tool install でインストールされた zskills は、コンテナ内で以下に配置されます:

• 実行バイナリ: /root/.local/bin/zskills
• 実体 (venv): /root/.local/share/uv/tools/zskills/

/root/.local/bin が PATH に通っているので、Bob Shell 内から zskills として呼べます。

.bob/mcp.json の設定

したがって、.bob/mcp.json は以下のように登録しておきます。

{
  "mcpServers": {
    "zskills": {
      "command": "/root/.local/bin/zskills"
    }
  }
}

これで Bob Shell から zskills のスキルやツールが MCP 経由で呼べるようになります。

---

まとめ

Bob Shell は単なる CLI ではなく、「AI エージェントエンジン」を任意の環境に埋め込める汎用パーツとして応用範囲が広いように感じました。

---

参考リンク

• IBM Bob 公式: https://bob.ibm.com/
• Bob Shell ドキュメント: https://bob.ibm.com/docs/shell
• 非対話モードの使い方: https://bob.ibm.com/docs/shell/getting-started/start-bobshell-non-interactive
• カスタムモード: https://bob.ibm.com/docs/shell/configuration/custom-modes-bobshell
• IBM Cloud Code Engine: https://www.ibm.com/products/code-engine
• IBM Cloud Object Storage: https://www.ibm.com/products/cloud-object-storage