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

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

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

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

前提条件

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

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

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

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

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

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

作業ディレクトリの作成

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

MCP設定ファイルの作成

OpenCodeは、プロジェクトのルートディレクトリにある opencode.json ファイルを通じてMCP Serverを管理します。プロジェクトのルートディレクトリでこのファイルを作成または編集し、OceanBase MCP Serverの設定を追加してください。例：

{
  "$schema": "https://opencode.ai/config.json",
  "mcp": {
    "my-local-mcp-server": {
      "type": "local",
      "command": ["uvx", "--from", "oceanbase-mcp", "oceanbase_mcp_server"],
      "enabled": true,
      "environment": {
        "OB_HOST": "***",
        "OB_PORT": "***",
        "OB_USER": "***",
        "OB_PASSWORD": "***",
        "OB_DATABASE": "***"
      },
    },
  },
}

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

OB_HOST
OB_PORT
OB_USER
OB_PASSWORD
OB_DATABASE

MCP接続のテスト

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

opencode

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

test 庫中有多少張表？

opencode-settings

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

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

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

サンプルテーブルの作成

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

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

テストデータの挿入

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

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

RESTful APIコードの生成

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

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

コード生成が完了したら、生成ファイルを受け取り、保存します。

データベース接続の設定

OpenCodeのmain.pyファイルでDATABASE_URL設定方式を使用している場合は、実際のOceanBase接続情報に更新してください。

例：

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

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

プロジェクトディレクトリで以下のコマンドを実行し、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"

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

付録：RESTful API 全コード例

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

from fastapi import FastAPI, HTTPException, status
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base
from pydantic import BaseModel
from typing import List, Optional

app = FastAPI(title="Customer API", description="RESTful API for Customer table")

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

engine = create_engine(DATABASE_URL)
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(200))


Base.metadata.create_all(bind=engine)


class CustomerBase(BaseModel):
    name: Optional[str] = None
    age: Optional[int] = None
    telephone: Optional[str] = None
    location: Optional[str] = None


class CustomerCreate(CustomerBase):
    ID: int


class CustomerResponse(CustomerBase):
    ID: int

    class Config:
        from_attributes = True


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


@app.post(
    "/customers", response_model=CustomerResponse, status_code=status.HTTP_201_CREATED
)
def create_customer(customer: CustomerCreate):
    db = SessionLocal()
    try:
        db_customer = Customer(**customer.model_dump())
        db.add(db_customer)
        db.commit()
        db.refresh(db_customer)
        return db_customer
    except Exception as e:
        db.rollback()
        raise HTTPException(status_code=400, detail=str(e))
    finally:
        db.close()


@app.get("/customers", response_model=List[CustomerResponse])
def get_customers():
    db = SessionLocal()
    try:
        customers = db.query(Customer).all()
        return customers
    finally:
        db.close()


@app.get("/customers/{customer_id}", response_model=CustomerResponse)
def get_customer(customer_id: int):
    db = SessionLocal()
    try:
        customer = db.query(Customer).filter(Customer.ID == customer_id).first()
        if not customer:
            raise HTTPException(status_code=404, detail="Customer not found")
        return customer
    finally:
        db.close()


@app.put("/customers/{customer_id}", response_model=CustomerResponse)
def update_customer(customer_id: int, customer: CustomerBase):
    db = SessionLocal()
    try:
        db_customer = db.query(Customer).filter(Customer.ID == customer_id).first()
        if not db_customer:
            raise HTTPException(status_code=404, detail="Customer not found")

        update_data = customer.model_dump(exclude_unset=True)
        for key, value in update_data.items():
            setattr(db_customer, key, value)

        db.commit()
        db.refresh(db_customer)
        return db_customer
    finally:
        db.close()


@app.delete("/customers/{customer_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_customer(customer_id: int):
    db = SessionLocal()
    try:
        db_customer = db.query(Customer).filter(Customer.ID == customer_id).first()
        if not db_customer:
            raise HTTPException(status_code=404, detail="Customer not found")

        db.delete(db_customer)
        db.commit()
    finally:
        db.close()


if __name__ == "__main__":
    import uvicorn

    uvicorn.run(app, host="0.0.0.0", port=8000)