OceanBase MCP ServerとClaude Codeの連携|OceanBase Cloud|ドキュメント|OceanBaseマニュアル

MCP（Model Context Protocol）は、Anthropic社が2024年11月に発表し、オープンソース化したプロトコルであり、大規模言語モデルと外部ツールやデータソースとの連携を実現することを目的としています。MCPを利用することで、ユーザーは大規模言語モデルの出力を手動でコピーして実行する必要がなく、モデルが直接ツールに指示を出して対応するアクションを実行させることができます。

OceanBase MCP Serverは、MCPプロトコルを通じて大規模言語モデルとOceanBaseデータベースとの連携を提供し、SQL文を実行することで、開発者が迅速にプロトタイプアプリケーションを構築できるよう支援します。

Claude Codeは、Anthropicが提供するAIプログラミングエージェントで、端末上で開発者を支援し、コードの生成、リファクタリング、デバッグを行うことができます。本記事では、Claude CodeにOceanBase MCP Serverを連携させ、データベースに基づいてFastAPIスタイルのRESTful APIプロジェクトを迅速に生成する方法を紹介します。

前提条件

環境に利用可能なトランザクション型（MySQL）のクラスタインスタンスがあります。
環境には使用可能なMySQL互換モードのテナント、MySQLデータベース、およびデータベースアカウントが既に存在し、データベースアカウントに読み書き権限が付与されています。作成が必要な場合は、詳細についてはアカウントの作成およびデータベースの作成（MySQL専用）をご参照ください。
プロジェクト管理者またはインスタンス管理者ロールを持ち、プロジェクト内のインスタンスに対する読み書き操作を実行できます。権限がない場合は、組織管理者に連絡して権限の追加を依頼してください。
Python 3.11以降のバージョンと対応する pip をインストールしています。マシン上のPythonバージョンが低い場合は、Minicondaを使用して新しいPython 3.11以降の環境を作成できます。詳細については Minicondaインストールガイドをご参照ください。
Pythonパッケージマネージャー uv をインストールしています。
Claude Code をインストールしています。

ステップ1：データベース接続情報を取得する

ドロップダウンリストから、IDに基づいて対象のクラスタインスタンスを選択します。
概要ページに移動します。
接続をクリックし、接続文字列を取得 を選択します。
ポップアップウィンドウで、パブリックネットワークを使用 を選択します。
アクセスアドレスを取得し、現在のブラウザIPアドレスを追加 を選択します。
データベース関連情報を入力し、接続文字列をコピー します。

接続パラメータの説明は以下のとおりです：

host：OceanBaseデータベースの接続アドレス。
port：OceanBaseデータベースの接続ポート。
database：アクセスするデータベースの名前。
user：テナント接続ユーザー名（一般的な形式は ユーザー名@テナント名#クラスタ名）。
password：アカウントのパスワード。

ステップ2：Claude CodeでOceanBase MCP Serverを設定する

作業ディレクトリの作成

手動でClaude Codeの作業ディレクトリを作成して開きます。例えば、claudecode-generate とします。後でClaude Codeが生成するプロジェクトファイルは、このディレクトリに保存されます。

MCP Server設定の追加

端末で claude mcp add-json コマンドを実行し、OceanBase MCP Serverの設定を追加します：

claude mcp add ob-mcp-server --transport stdio \
  -e OB_HOST=*** \
  -e OB_PORT=*** \
  -e OB_USER=*** \
  -e OB_PASSWORD=*** \
  -e OB_DATABASE=*** \
  -- uvx --from oceanbase-mcp oceanbase_mcp_server

以下のパラメータをご自身のデータベース情報に置き換えてください：

OB_HOST
OB_PORT
OB_USER
OB_PASSWORD
OB_DATABASE

MCP接続のテスト

プロジェクトのルートディレクトリでClaude Codeを起動します：

claude

Claude Codeのダイアログボックスにプロンプトを入力します：

test データベースにはいくつのテーブルがありますか？

正しい結果が返されれば、Claude CodeがOceanBase MCP Server経由でデータベースに正常にアクセスできたことを意味します。

ステップ3：Claude Codeを使用してFastAPIプロジェクトを生成する

FastAPIを使用して、RESTful APIスタイルのプロジェクトを迅速に作成できます。FastAPIはPythonのWebフレームワークで、RESTful APIを迅速に構築できます。

サンプルテーブルの作成

Claude Codeのダイアログボックスにプロンプトを入力します：

customerテーブルを作成してください。主キーはIDで、name、age、telephone、locationの各フィールドを含めてください。

テストデータの挿入

プロンプトを続けて入力します：

customerテーブルに10件のデータを挿入してください。

RESTful APIコードの生成

プロンプトを入力して、Claude Codeにcustomerテーブルに基づくFastAPIプロジェクトの生成を指示します：

FastAPIプロジェクトを作成し、customerテーブルに基づくRESTful APIを生成します。

コード生成が完了したら、生成されたファイルを受け取り保存します。

データベース接続の設定

database.py 内の接続プール設定を変更し、データベース接続パラメータを実際のOceanBase情報に修正します。

仮想環境の作成と依存関係のインストール

現在のディレクトリで以下のコマンドを実行し、uvパッケージ管理ツールを使用して仮想環境を作成し、依存パッケージをインストールします：

uv venv
source .venv/bin/activate
uv pip install -r requirements.txt

FastAPIサービスの起動

以下のコマンドを実行してプロジェクトを起動します（実際のエントリーファイル名に合わせて調整してください）：

uvicorn main:app --reload

APIの検証

サービスが起動した後、以下の方法で検証できます：

`http://127.0.0.1:8000/docs`` にアクセスしてSwaggerドキュメントを確認します。
curl またはPostmanを使用してCRUDインターフェースを呼び出し、結果を検証します。

例えば、すべての顧客をクエリする場合：

curl "http://127.0.0.1:8000/customers"

上記のすべての手順を完了すると、Claude Codeによって生成され、OceanBaseデータで駆動されるFastAPI RESTful APIプロジェクトを取得できます。これにより、customer テーブルに対する追加、削除、変更、検索などの一般的な操作を実行できるようになります。

付録：RESTful APIの完全なコード例

追加、削除、更新、検索のためのRESTful APIが正常に生成された後のコード例は以下のとおりです：

from fastapi import FastAPI, HTTPException
from typing import List
from database import get_connection
from models import Customer, CustomerCreate, CustomerUpdate

app = FastAPI(
    title="Customer API",
    description="customerテーブルに基づくRESTful API",
    version="1.0.0"
)

@app.get("/")
def root():
    """APIのルートパス"""
    return {"message": "Welcome to Customer API", "docs": "/docs"}

@app.get("/customers", response_model=List[Customer], summary="すべての顧客を取得する")
def get_all_customers():
    """すべての顧客リストを取得する"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("SELECT id, name, age, telephone, location FROM customer ORDER BY id")
        customers = cursor.fetchall()
        return customers
    finally:
        cursor.close()
        conn.close()

@app.get("/customers/{customer_id}", response_model=Customer, summary="単一顧客の取得")
def get_customer(customer_id: int):
    """IDに基づいて単一顧客を取得します"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(
            "SELECT id, name, age, telephone, location FROM customer WHERE id = %s",
            (customer_id,)
        )
        customer = cursor.fetchone()
        if not customer:
            raise HTTPException(status_code=404, detail="顧客が存在しません")
        return customer
    finally:
        cursor.close()
        conn.close()

@app.post("/customers", response_model=Customer, summary="顧客の作成", status_code=201)
def create_customer(customer: CustomerCreate):
    """新規顧客を作成します"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(
            """INSERT INTO customer (name, age, telephone, location)
               VALUES (%s, %s, %s, %s)""",
            (customer.name, customer.age, customer.telephone, customer.location)
        )
        conn.commit()
        new_id = cursor.lastrowid
        # 新規作成された顧客を取得します
        cursor.execute(
            "SELECT id, name, age, telephone, location FROM customer WHERE id = %s",
            (new_id,)
        )
        return cursor.fetchone()
    finally:
        cursor.close()
        conn.close()

@app.put("/customers/{customer_id}", response_model=Customer, summary="顧客情報を更新する")
def update_customer(customer_id: int, customer: CustomerUpdate):
    """顧客情報を更新します"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        # 顧客が存在するか確認します
        cursor.execute("SELECT id FROM customer WHERE id = %s", (customer_id,))
        if not cursor.fetchone():
            raise HTTPException(status_code=404, detail="顧客が存在しません")

        # 動的更新ステートメントを構築します
        update_fields = []
        values = []
        if customer.name is not None:
            update_fields.append("name = %s")
            values.append(customer.name)
        if customer.age is not None:
            update_fields.append("age = %s")
            values.append(customer.age)
        if customer.telephone is not None:
            update_fields.append("telephone = %s")
            values.append(customer.telephone)
        if customer.location is not None:
            update_fields.append("location = %s")
            values.append(customer.location)

        if update_fields:
            values.append(customer_id)
            sql = f"UPDATE customer SET {', '.join(update_fields)} WHERE id = %s"
            cursor.execute(sql, values)
            conn.commit()

        # 更新後の顧客を返します
        cursor.execute(
            "SELECT id, name, age, telephone, location FROM customer WHERE id = %s",
            (customer_id,)
        )
        return cursor.fetchone()
    finally:
        cursor.close()
        conn.close()

@app.delete("/customers/{customer_id}", summary="顧客を削除する", status_code=204)
def delete_customer(customer_id: int):
    """顧客を削除します"""
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("DELETE FROM customer WHERE id = %s", (customer_id,))
        if cursor.rowcount == 0:
            raise HTTPException(status_code=404, detail="顧客は存在しません")
        conn.commit()
        return None
    finally:
        cursor.close()
        conn.close()

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8000)