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

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

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

Codexは、OpenAIが提供するオープンソースのAIプログラミングエージェントで、Rustで構築され、端末で実行できます。コードの読み取り、変更、実行が可能で、開発者がアイデアを高品質なコードに素早く変換するのを支援します。CodexはCLI、IDE拡張機能、Codex Appの3つの利用方法をサポートしています。本記事では、Codex CLIを使用してOceanBase MCP Serverと連携し、バックエンドアプリケーションを迅速に構築する方法を紹介します。

前提条件

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

Codex CLIをインストールし、OpenAI APIキーを設定しています。

npm install -g @openai/codex
export OPENAI_API_KEY="your_api_key"
# インストールが成功したか確認する
codex --help

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

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

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

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

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

作業ディレクトリの作成

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

MCP Server設定の追加

Codex CLIのバージョンに応じて、コマンドラインからMCP Serverにアクセスできます。

codex mcp add ob-mcp-server \
  --env OB_HOST=*** \
  --env OB_PORT=*** \
  --env OB_USER=*** \
  --env OB_PASSWORD=*** \
  --env OB_DATABASE=*** \
  -- uvx --from oceanbase-mcp oceanbase_mcp_server

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

OB_HOST
OB_PORT
OB_USER
OB_PASSWORD
OB_DATABASE

MCP接続のテスト

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

codex

Codexのダイアログにプロンプトを入力します：

test 庫中有多少張表？

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

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

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

サンプルテーブルの作成

Codexダイアログでプロンプトを入力します：

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

テストデータの挿入

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

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

挿入が成功すると、test.customerに10件のデータを挿入しました。というメッセージが表示されます。

RESTful APIコードの生成

プロンプトを入力して、Codexにcustomerテーブルに基づいたFastAPIプロジェクトを生成させます：

FastAPIプロジェクトを作成し、customer テーブルに基づいたRESTful APIを生成してください。

データベース接続の設定

データベース接続パラメータを実際のOceanBase接続情報に更新します。

例：

DATABASE_URL=mysql+pymysql://user:password@host:3306/test

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

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

cd customer_api
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"

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

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

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

from fastapi import Depends, FastAPI, HTTPException, status
from sqlalchemy.orm import Session

from . import crud, schemas
from .db import SessionLocal

app = FastAPI(title="Customer API")


def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()


@app.get("/customers", response_model=list[schemas.CustomerOut])
def list_customers(
    skip: int = 0, limit: int = 100, db: Session = Depends(get_db)
):
    return crud.get_customers(db, skip=skip, limit=limit)


@app.get("/customers/{customer_id}", response_model=schemas.CustomerOut)
def get_customer(customer_id: int, db: Session = Depends(get_db)):
    customer = crud.get_customer(db, customer_id)
    if not customer:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
    return customer


@app.post("/customers", response_model=schemas.CustomerOut, status_code=201)
def create_customer(
    customer_in: schemas.CustomerCreate, db: Session = Depends(get_db)
):
    return crud.create_customer(db, customer_in)


@app.put("/customers/{customer_id}", response_model=schemas.CustomerOut)
def update_customer(
    customer_id: int, customer_in: schemas.CustomerUpdate, db: Session = Depends(get_db)
):
    customer = crud.get_customer(db, customer_id)
    if not customer:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
    return crud.update_customer(db, customer, customer_in)


@app.delete("/customers/{customer_id}", status_code=204)
def delete_customer(customer_id: int, db: Session = Depends(get_db)):
    customer = crud.get_customer(db, customer_id)
    if not customer:
        raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Not found")
    crud.delete_customer(db, customer)