MCP（Model Context Protocol）は、Anthropic社が2024年11月に発表し、オープンソース化したプロトコルです。大規模言語モデルと外部ツールやデータソースとの連携を実現することを目的としています。MCPを利用することで、ユーザーは大規模言語モデルの出力を手動でコピーして実行する必要がなく、モデルが直接ツールに指示を出して対応するアクションを実行させることができます。
OceanBase MCP Serverは、MCPプロトコルを通じて大規模言語モデルとOceanBaseの連携を提供し、SQL文を実行できるようにします。適切なクライアントを利用することでプロジェクトのプロトタイプを迅速に構築でき、GitHub上でオープンソースとして公開されています。
Traeは、MCP Serverと連携可能なIDEで、公式サイトから最新版をダウンロードできます。
本記事では、Trae IDEとOceanBase MCP Serverを連携させ、バックエンドアプリケーションを迅速に構築する方法を紹介します。
前提条件
OceanBaseデータベースのデプロイが完了し、MySQLモードのユーザーテナントが作成されていること。テナント作成の詳細については、テナントの作成を参照してください。
Python 3.11以降のバージョンと対応するpipをインストールしていること。お使いのマシンのPythonバージョンが低い場合は、Minicondaを使用して新しいPython 3.11以降の環境を作成できます。詳細については、Minicondaインストールガイドを参照してください。
使用しているOSに応じて、Gitをインストールしていること。
Pythonパッケージマネージャーuvをインストールしていること。インストール完了後、
uv --versionコマンドを実行してインストールが成功したかどうか確認できます：pip install uv uv --versionTrae IDEをダウンロードし、ご自身のOSに適したバージョンを選択してインストールします。
ステップ１：データベース接続情報を取得する
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リポジトリのクローン
以下のコマンドを実行して、ソースコードをローカルにダウンロードします：
```shell
git clone https://github.com/oceanbase/mcp-oceanbase.git
```
ソースコードディレクトリに移動します：
```shell
cd mcp-oceanbase
```
依存関係のインストール
mcp-oceanbase ディレクトリで以下のコマンドを実行して仮想環境を作成し、依存関係をインストールします：
```shell
uv venv
source .venv/bin/activate
uv pip install .
```
Traeクライアントの作業ディレクトリの作成
Traeの作業ディレクトリを手動で作成し、Traeで開きます。今後Traeが生成するファイルはこのディレクトリに配置されます。例としてのディレクトリ名は trae-generate です。
TraeクライアントでOceanBase MCP Serverの設定
ショートカットキー Ctrl + U (Windows) または Command + U (MacOS) を使用してチャットダイアログを開き、右上の歯車アイコンをクリックして MCP を選択します。
MCP Serversの追加と設定
MCP Serversの追加をクリックし、手動で設定を選択します。設定ファイルに入力し、編集ボックス内のサンプル内容を削除します。
次に以下の内容を入力します：
```json
{
mcpServers: {
oceanbase: {
command: "uv",
args: [
--directory,
// oceanbase_mcp_serverフォルダの絶対パスに置き換える必要があります
/path/to/your/mcp-oceanbase/src/oceanbase_mcp_server,
run,
oceanbase_mcp_server
],
env: {
// OceanBaseデータベースの接続情報に置き換える必要があります
OB_HOST: "***",
OB_PORT: "***",
OB_USER: "***",
OB_PASSWORD: "***",
OB_DATABASE: "***"
}
}
}
}
```
設定が成功すると、
使用可能状態が表示されます。
MCP Serverのテスト
Builder with MCPインテリジェントを選択します。ダイアログにプロンプト
test 庫中有多少張表？を入力します。Traeクライアントは実行されるSQLステートメントを表示します。確認後、実行ボタンをクリックしてクエリを実行します。Traeクライアントは
testデータベース内のすべてのテーブル名を表示します。これにより、OceanBaseデータベースへの接続が正常に確立されたことが確認できます。
FastAPIを使用してRESTful APIスタイルのプロジェクトを迅速に作成する
FastAPIを使用して、RESTful APIスタイルのプロジェクトを迅速に作成できます。FastAPIは、PythonでRESTful APIを迅速に構築できるWebフレームワークです。
- customerテーブルを作成する
ダイアログボックスに次のプロンプトを入力します：customerテーブルを作成してください。主キーはIDで、name、age、telephone、locationの各フィールドを含めてください。SQL文を確認した後、実行ボタンをクリックしてクエリを実行します。
<!--  -->
- テストデータを挿入する
ダイアログボックスに次のプロンプトを入力します：customerテーブルに10件のデータを挿入してください。SQL文を確認した後、実行ボタンをクリックしてクエリを実行します。
<!--  -->
挿入が成功すると、実行結果が表示されます：
<!--  -->
- FastAPIプロジェクトを作成する
ダイアログボックスに次のプロンプトを入力します：FastAPIプロジェクトを作成し、customerテーブルに基づいたRESTful APIを生成してください。SQL文を確認した後、実行ボタンをクリックしてクエリを実行します。
<!--  -->
このステップでは、3つのファイルが自動生成されます。AIが生成するファイルの内容は不確定な場合があるため、初回使用時は「すべて受け入れる」を選択し、後から実際のニーズに合わせて調整することをお勧めします。
- 仮想環境を作成し、依存関係をインストールする
以下のコマンドを実行し、現在のディレクトリでuvパッケージ管理ツールを使用して仮想環境を作成し、依存パッケージをインストールします：
```shell
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
実行結果は次のとおりです：
```json
[{"ID":1,"name":"張三","age":25,"telephone":"13800138000","location":"北京"},{"ID":2,"name":"李四","age":30,"telephone":"13900139000","location":"上海"},{"ID":3,"name":"王五","age":35,"telephone":"13700137000","location":"廣州"},{"ID":4,"name":"趙六","age":22,"telephone":"13600136000","location":"深圳"},{"ID":5,"name":"孫七","age":40,"telephone":"13500135000","location":"成都"},{"ID":6,"name":"周八","age":28,"telephone":"13400134000","location":"杭州"},{"ID":7,"name":"吳九","age":33,"telephone":"13300133000","location":"南京"},{"ID":8,"name":"鄭十","age":27,"telephone":"13200132000","location":"武漢"},{"ID":9,"name":"陳十一","age":31,"telephone":"13100131000","location":"西安"},{"ID":10,"name":"林十二","age":24,"telephone":"13000130000","location":"重慶"}]
```
これにより、追加、削除、更新、検索のためのRESTful APIが正常に生成されたことが確認できます：
from fastapi import FastAPI
from pydantic import BaseModel
import mysql.connector
app = FastAPI()
# データベース接続設定
config = {
user': '*******',
password': '******',
host': 'xx.xxx.xxx.xx',
database': 'test',
port':xxxx,
raise_on_warnings': True
}
class Customer(BaseModel):
id: int
name: str
age: int
telephone: str
location: str
@app.get('/customers')
async def get_customers():
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor(dictionary=True)
query = 'SELECT * FROM customer'
cursor.execute(query)
results = cursor.fetchall()
cursor.close()
cnx.close()
return results
@app.get('/customers/{customer_id}')
async def get_customer(customer_id: int):
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor(dictionary=True)
query = 'SELECT * FROM customer WHERE ID = %s'
cursor.execute(query, (customer_id,))
result = cursor.fetchone()
cursor.close()
cnx.close()
return result
@app.post('/customers')
async def create_customer(customer: Customer):
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
query = 'INSERT INTO customer (ID, name, age, telephone, location) VALUES (%s, %s, %s, %s, %s)'
data = (customer.id, customer.name, customer.age, customer.telephone, customer.location)
cursor.execute(query, data)
cnx.commit()
cursor.close()
cnx.close()
return {'message': 'Customer created successfully'}
@app.put('/customers/{customer_id}')
async def update_customer(customer_id: int, customer: Customer):
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
query = 'UPDATE customer SET name = %s, age = %s, telephone = %s, location = %s WHERE ID = %s'
data = (customer.name, customer.age, customer.telephone, customer.location, customer_id)
cursor.execute(query, data)
cnx.commit()
cursor.close()
cnx.close()
return {'message': 'Customer updated successfully'}
@app.delete('/customers/{customer_id}')
async def delete_customer(customer_id: int):
cnx = mysql.connector.connect(**config)
cursor = cnx.cursor()
query = 'DELETE FROM customer WHERE ID = %s'
cursor.execute(query, (customer_id,))
cnx.commit()
cursor.close()
cnx.close()
return {'message': 'Customer deleted successfully'}