MCP（Model Context Protocol）は、Anthropic社が2024年11月に発表し、オープンソース化したプロトコルです。大規模言語モデル（LLM）が外部のツールやデータソースと直接やり取りできるよう設計されています。MCPを使えば、ユーザーがLLMの出力をコピー＆ペーストして手動で実行する必要はなく、LLMからツールへ直接指示を送り、そのままアクションを実行させることが可能になります。
OceanBase MCP Serverは、MCPプロトコルを介して大規模言語モデルとOceanBaseデータベースを連携させ、SQL文の実行を可能にするサーバーです。GitHubでオープンソースとして公開されており、対応するクライアントを利用すれば、プロジェクトのプロトタイプを素早く構築できます。
Cursorは、AI技術を統合したコードエディターで、Windows、macOS、Linuxを含む複数のオペレーティングシステムに対応しています。
この記事では、CursorとOceanBase MCP Serverを連携させて、バックエンドアプリケーションを迅速に構築する方法を紹介します。
前提条件
OceanBaseデータベースのデプロイが完了し、MySQLモードのユーザーテナントが作成されていること。テナント作成の詳細については、テナントの作成を参照してください。
Python 3.11以降のバージョンと対応するpipをインストールします。お使いのマシンにインストールされているPythonのバージョンが古い場合は、Minicondaを使用してPython 3.11以降の新しい環境を作成することができます。詳細はMinicondaインストールガイドを参照してください。
お使いのオペレーティングシステムに応じて、Gitをインストールします。
Pythonパッケージマネージャーのuvをインストールします。インストール完了後、
uv --versionコマンドでインストールが成功したかを確認できます。pip install uv uv --versionCursorをダウンロードし、ご利用のオペレーティングシステムに対応するバージョンをインストールします。初めてCursorを利用する場合は、新規アカウントを作成か、既存のアカウントでログインしてください。ログイン後は、新しいプロジェクトを作成することも、既存のプロジェクトを開くこともできます。
ステップ1：データベース接続情報を取得する
OceanBaseデータベースのデプロイ担当者または管理者に連絡し、対応するデータベース接続文字列を取得します。例：
obclient -h$host -P$port -u$user_name -p$password -D$database_name
パラメータの説明：
$host：OceanBaseデータベースへの接続IPアドレスです。OceanBaseデータベースプロキシ(OceanBase Database Proxy、ODP)経由で接続する場合はODPのアドレスを、直接接続する場合はOBServerノードのIPアドレスを使用します。$port：OceanBaseデータベースの接続ポートです。ODP経由の接続ではデフォルトで2883が使用されますが、ODPのデプロイ時にカスタマイズ可能です。直接接続の場合はデフォルトで2881が使用され、OceanBaseデータベースのデプロイ時にカスタマイズできます。$database_name：アクセスするデータベースの名前です。注意
テナントに接続するユーザーは、データベースに対する
CREATE、INSERT、DROP、およびSELECT権限が付与されていなければなりません。ユーザー権限の詳細については、MySQLモードの権限分類を参照してください。$user_name：テナントへの接続アカウントです。ODP経由で接続する場合の一般的な形式はユーザー名@テナント名#クラスタ名またはクラスタ名:テナント名:ユーザー名。直接接続の場合はユーザー名@テナント名となります。$password：アカウントのパスワードです。
接続文字列の詳細については、OBClientを使用してOceanBaseテナントに接続するを参照してください。
ステップ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 + U (Windows)またはCommand + U (MacOS)を使用してチャットダイアログを開き、右上の歯車アイコンをクリックしてMCP Toolsを選択します。
MCP Serversの追加と設定
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文を表示します。内容を確認し、問題がなければ実行ボタンをクリックします。実行結果として、Cursorクライアント内でtestデータベース内のすべてのテーブル名が表示されます。これにより、OceanBaseデータベースへの接続が正常に確立されたことを確認できます。
FastAPIを使用してRESTful APIスタイルのプロジェクトを迅速に作成する
FastAPIは、PythonでRESTful APIを迅速に構築できるWebフレームワークです。ここでは、customerテーブルを作成し、テストデータを挿入したうえで、FastAPIプロジェクトを自動生成します。
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ボタンをクリックします。このステップでは3つのファイルが自動生成されます。AIが生成する内容は完全ではない場合があるため、初めて使用する際は
すべて受け入れるを選択し、後からニーズに合わせて修正することを推奨します。仮想環境の作成と依存関係のインストール
以下のコマンドを実行し、現在のディレクトリで
uvパッケージ管理ツールを使って仮想環境を作成し、依存パッケージをインストールします。uv venv source .venv/bin/activate uv pip install -r requirements.txtFastAPIプロジェクトの起動
以下のコマンドを実行して、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}