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

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

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

GitHub Copilotは、AIプログラミングエージェントであり、開発者がアイデアを迅速にコードに変換するのを支援します。本記事では、VS CodeでGitHub CopilotをOceanBase MCP Serverに接続し、データベースに基づいてFastAPIスタイルのRESTful APIプロジェクトを迅速に生成する方法を紹介します。

前提条件

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

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

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

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

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

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

作業ディレクトリの作成

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

MCP Serverの設定の追加

コマンドパネルまたは設定ファイルの2つの方法でMCP Serverを追加できます：

コマンドパネル：Ctrl + Shift + P (Windows/Linux) または Command + Shift + P (macOS) を押し、MCP: Add Server と入力します。Command (stdio) を選択した後、プロンプトに従って入力します。
設定ファイル：プロジェクトのルートディレクトリに .vscode/mcp.json ファイルを作成し、以下の設定を記述します：

{
  "servers": {
    "ob-mcp-server": {
      "type": "stdio",
      "command": "uvx",
      "args": [
        "--from",
        "oceanbase-mcp",
        "oceanbase_mcp_server"
      ],
      "env": {
        "OB_HOST": "***",
        "OB_PORT": "***",
        "OB_USER": "***",
        "OB_PASSWORD": "***",
        "OB_DATABASE": "***"
      }
    }
  },
  "inputs": []
}

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

OB_HOST
OB_PORT
OB_USER
OB_PASSWORD
OB_DATABASE

設定完了後、コマンドパネルの MCP: List Servers でMCP Serverの接続状態を確認できます。

MCP接続のテスト

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

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

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

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

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

サンプルテーブルの作成

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

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

SQLが生成されたら、Allow をクリックして実行します。

テストデータの挿入

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

customerテーブルに10件のデータを挿入します。

SQLを確認した後、Allow をクリックして実行します。

RESTful APIコードの生成

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

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

このステップでは通常、複数のファイルが自動生成されます。Allow をクリックして実行し、生成された内容を保存してください。

データベース接続の設定

database.py または DATABASE_URL の設定を変更し、データベース接続パラメータを実際の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"

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

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

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

from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from sqlalchemy import text
from typing import List

from database import get_db
from models import Customer, CustomerCreate, CustomerUpdate

app = FastAPI(title="Customer API", version="1.0.0")

@app.get("/customers", response_model=List[Customer])
def get_all_customers(db: Session = Depends(get_db)):
    """Get all customers"""
    result = db.execute(text("SELECT id, name, age, telephone, location FROM test.customer"))
    customers = []
    for row in result:
        customers.append({
            "id": row[0],
            "name": row[1],
            "age": row[2],
            "telephone": row[3],
            "location": row[4]
        })
    return customers

@app.get("/customers/{customer_id}", response_model=Customer)
def get_customer(customer_id: int, db: Session = Depends(get_db)):
    """Get a customer by ID"""
    result = db.execute(
        text("SELECT id, name, age, telephone, location FROM test.customer WHERE id = :id"),
        {"id": customer_id}
    )
    row = result.first()
    if not row:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    return {
        "id": row[0],
        "name": row[1],
        "age": row[2],
        "telephone": row[3],
        "location": row[4]
    }

@app.post("/customers", response_model=Customer)
def create_customer(customer: CustomerCreate, db: Session = Depends(get_db)):
    """Create a new customer"""
    db.execute(
        text("INSERT INTO test.customer (id, name, age, telephone, location) VALUES (:id, :name, :age, :telephone, :location)"),
        {
            "id": customer.id,
            "name": customer.name,
            "age": customer.age,
            "telephone": customer.telephone,
            "location": customer.location
        }
    )
    db.commit()
    
    return {
        "id": customer.id,
        "name": customer.name,
        "age": customer.age,
        "telephone": customer.telephone,
        "location": customer.location
    }

@app.put("/customers/{customer_id}", response_model=Customer)
def update_customer(customer_id: int, customer: CustomerUpdate, db: Session = Depends(get_db)):
    """Update a customer"""
    # Get current customer
    result = db.execute(
        text("SELECT id, name, age, telephone, location FROM test.customer WHERE id = :id"),
        {"id": customer_id}
    )
    row = result.first()
    if not row:
        raise HTTPException(status_code=404, detail="Customer not found")
    
    # Prepare update data
    update_data = {
        "id": customer_id,
        "name": customer.name if customer.name is not None else row[1],
        "age": customer.age if customer.age is not None else row[2],
        "telephone": customer.telephone if customer.telephone is not None else row[3],
        "location": customer.location if customer.location is not None else row[4]
    }
    
    # Update customer
    db.execute(
        text("UPDATE test.customer SET name = :name, age = :age, telephone = :telephone, location = :location WHERE id = :id"),
        update_data
    )
    db.commit()
    
    return update_data

@app.delete("/customers/{customer_id}")
def delete_customer(customer_id: int, db: Session = Depends(get_db)):
    """Delete a customer"""
    result = db.execute(
        text("SELECT id FROM test.customer WHERE id = :id"),
        {"id": customer_id}
    )
    if not result.first():
        raise HTTPException(status_code=404, detail="Customer not found")
    
    db.execute(
        text("DELETE FROM test.customer WHERE id = :id"),
        {"id": customer_id}
    )
    db.commit()
    
    return {"message": "Customer deleted successfully"}

@app.get("/")
def read_root():
    """Root endpoint"""
    return {"message": "Customer API - use /customers to manage customer data"}

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