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

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

OceanBase MCP Serverは、MCPプロトコルを通じて大規模言語モデルとOceanBaseの連携を提供し、SQL文を実行できるようにします。適切なクライアントを利用することでプロジェクトのプロトタイプを迅速に構築でき、GitHub上でオープンソースとして公開されています。

Cursorは、AI技術を統合したコードエディタであり、Windows、macOS、Linuxを含む様々なOSをサポートしています。

本記事では、CursorとOceanBase MCP Serverを連携させ、バックエンドアプリケーションを迅速に構築する方法を紹介します。

前提条件

環境に利用可能なトランザクション型(MySQL)クラスタインスタンスがあります。

テナントの作成を参照してテナントの作成を完了した後、以下の手順を実行してください。

環境には既に使用可能なMySQL互換モードのテナントとMySQLデータベースおよびアカウントが存在し、データベースアカウントに読み書き権限が付与されています。作成が必要な場合は、詳細についてはアカウントの作成およびデータベースの作成(MySQL専用)をご参照ください。
プロジェクト管理者またはインスタンス管理者ロールを持ち、プロジェクト内のインスタンスに対する読み書き操作を実行できます。権限がない場合は、組織管理者に連絡して権限の追加を依頼してください。
Python 3.11以降のバージョンと対応するpipをインストールしてください。マシンのPythonバージョンが古い場合は、Minicondaを使用して新しいPython 3.11以降の環境を作成できます。詳細についてはMinicondaインストールガイドをご参照ください。
使用しているOSに応じて、Gitをインストールします。
Pythonパッケージマネージャuvをインストールします。インストール完了後、uv --versionコマンドを使用してインストールが成功したかどうか確認できます：
```
pip install uv
uv --version
```
Cursorをダウンロードし、ご自身のOSに適したバージョンを選択してインストールします。Cursorを初めて使用する際は、新規アカウントを登録するか、既存のアカウントでログインする必要があります。ログイン後、新しいプロジェクトを作成するか、既存のプロジェクトを開くことができます。

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

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

ステップ2：OceanBase MCP Serverの設定

OceanBase MCP Serverリポジトリのクローン

以下のコマンドを実行して、ソースコードをローカルにダウンロードします：

git clone https://github.com/oceanbase/mcp-oceanbase.git

ソースコードディレクトリに移動します：

cd mcp-oceanbase

依存関係のインストール

mcp-oceanbase ディレクトリで以下のコマンドを実行して仮想環境を作成し、依存関係をインストールします：

uv venv
source .venv/bin/activate
uv pip install .

Cursorクライアントの作業ディレクトリの作成とOceanBase MCP Serverの設定

Cursorの作業ディレクトリを手動で作成し、Cursorで開きます。今後Cursorが生成するファイルはこのディレクトリに配置されます。例示のディレクトリ名は cursor です。

ショートカットキー Ctrl + L (Windows) または Command + L (MacOS) を使用してチャットダイアログを開き、右上の歯車アイコンをクリックして MCP Tools を選択します。

MCP Serverの追加と設定

Add Custom MCP をクリックして設定ファイルを入力します。

設定ファイルに入力し、確認をクリックします。

/path/to/your/mcp-oceanbase/src/oceanbase_mcp_server は oceanbase_mcp_server フォルダの絶対パスに、OB_HOST、OB_PORT、OB_USER、OB_PASSWORD、OB_DATABASE はご自身のデータベースの対応する情報に置き換える必要があります：

{
"mcpServers": {
    "oceanbase": {
    "command": "uv",
    "args": [
        "--directory", 
        "/path/to/your/mcp-oceanbase/src/oceanbase_mcp_server",
        "run",
        "oceanbase_mcp_server"
    ],
    "env": {
        "OB_HOST": "***",
        "OB_PORT": "***",
        "OB_USER": "***",
        "OB_PASSWORD": "***",
        "OB_DATABASE": "***"
    }
    }
}
}

設定が成功すると、利用可能 状態が表示されます。

MCP Serverのテスト

ダイアログに「test データベースにはテーブルがいくつありますか」と入力します。Cursorクライアントは実行予定のSQL文を表示します。内容を確認し、問題がなければ Run tool ボタンをクリックしてクエリを実行します。Cursorクライアントは test データベース内のすべてのテーブル名を表示します。これにより、OceanBaseデータベースへの接続が正常に確立されたことが確認できます。

FastAPIを使用してRESTful APIスタイルのプロジェクトを迅速に作成する

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

customerテーブルを作成する

ダイアログボックスに次のプロンプトを入力します：customerテーブルを作成してください。主キーはIDで、name、age、telephone、locationの各フィールドを含めてください。SQL文を確認した後、Run toolボタンをクリックしてクエリを実行します。
テストデータを挿入する

ダイアログボックスに次のプロンプトを入力します：customerテーブルに10件のデータを挿入してください。SQL文を確認した後、Run toolボタンをクリックしてクエリを実行します。挿入が成功すると、customerテーブルに10件のテストデータを正常に挿入しました…というメッセージが表示されます。
FastAPIプロジェクトを作成する

ダイアログボックスに次のプロンプトを入力します：FastAPIプロジェクトを作成し、customerテーブルに基づいたRESTful APIを生成してください。SQL文を確認した後、Run toolボタンをクリックしてクエリを実行します。

このステップでは2つのファイルが自動生成されます。AIが生成するファイルの内容は不確定な場合があるため、初回使用時はすべて受け入れるを選択し、後から実際のニーズに合わせて調整することをお勧めします。
仮想環境を作成し、依存関係をインストールする

以下のコマンドを実行して、現在のディレクトリでuvパッケージ管理ツールを使用して仮想環境を作成し、依存パッケージをインストールします：
```
uv venv
source .venv/bin/activate
uv pip install -r requirements.txt
```
FastAPIプロジェクトを起動する

以下のコマンドを実行して、FastAPIプロジェクトを起動します：
```
uvicorn main:app --reload
```

テーブル内のデータを確認する

コマンドラインで以下のコマンドを実行するか、他のリクエストツールを使用して、テーブル内のデータを確認します：

curl http://127.0.0.1:8000/customers

実行結果は次のとおりです：

[{"id":1,"name":"Alice","age":28,"telephone":"1234567890","location":"Beijing"},{"id":2,"name":"Bob","age":32,"telephone":"2345678901","location":"Shanghai"},{"id":3,"name":"Charlie","age":25,"telephone":"3456789012","location":"Guangzhou"},{"id":4,"name":"David","age":40,"telephone":"4567890123","location":"Shenzhen"},{"id":5,"name":"Eve","age":22,"telephone":"5678901234","location":"Chengdu"},{"id":6,"name":"Frank","age":35,"telephone":"6789012345","location":"Wuhan"},{"id":7,"name":"Grace","age":30,"telephone":"7890123456","location":"Hangzhou"},{"id":8,"name":"Heidi","age":27,"telephone":"8901234567","location":"Nanjing"},{"id":9,"name":"Ivan","age":29,"telephone":"9012345678","location":"Tianjin"},{"id":10,"name":"Judy","age":31,"telephone":"0123456789","location":"Chongqing"}]

これにより、追加、削除、更新、検索のためのRESTful APIが正常に生成されたことが確認できます：

from fastapi import FastAPI, HTTPException, Depends
from pydantic import BaseModel
from typing import List
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session

# OceanBase接続設定（実際の状況に応じて変更してください）
DATABASE_URL = "mysql://***:***@***:***/***"

engine = create_engine(DATABASE_URL, echo=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

class Customer(Base):
    __tablename__ = "customer"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(100))
    age = Column(Integer)
    telephone = Column(String(20))
    location = Column(String(100))

class CustomerCreate(BaseModel):
    id: int
    name: str
    age: int
    telephone: str
    location: str

class CustomerUpdate(BaseModel):
    name: str = None
    age: int = None
    telephone: str = None
    location: str = None

class CustomerOut(BaseModel):
    id: int
    name: str
    age: int
    telephone: str
    location: str
    class Config:
        orm_mode = True

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

app = FastAPI()

@app.post("/customers/", response_model=CustomerOut)
def create_customer(customer: CustomerCreate, db: Session = Depends(get_db)):
    db_customer = Customer(**customer.dict())
    db.add(db_customer)
    try:
        db.commit()
        db.refresh(db_customer)
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    return db_customer

@app.get("/customers/", response_model=List[CustomerOut])
def read_customers(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
    return db.query(Customer).offset(skip).limit(limit).all()

@app.get("/customers/{customer_id}", response_model=CustomerOut)
def read_customer(customer_id: int, db: Session = Depends(get_db)):
    customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")
    return customer

@app.put("/customers/{customer_id}", response_model=CustomerOut)
def update_customer(customer_id: int, customer: CustomerUpdate, db: Session = Depends(get_db)):
    db_customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if db_customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")
    for var, value in vars(customer).items():
        if value is not None:
            setattr(db_customer, var, value)
    db.commit()
    db.refresh(db_customer)
    return db_customer

@app.delete("/customers/{customer_id}")
def delete_customer(customer_id: int, db: Session = Depends(get_db)):
    db_customer = db.query(Customer).filter(Customer.id == customer_id).first()
    if db_customer is None:
        raise HTTPException(status_code=404, detail="Customer not found")
    db.delete(db_customer)
    db.commit()
    return {"ok": True}