MCP（Model Context Protocol）は、Anthropic社が2024年11月に発表し、オープンソース化したプロトコルです。大規模言語モデルと外部ツールやデータソースとの連携を実現することを目的としています。MCPを利用することで、ユーザーは大規模言語モデルの出力を手動でコピーして実行する必要がなく、モデルが直接ツールに指示を出して対応するアクションを実行させることができます。
OceanBase MCP Serverは、MCPプロトコルを通じて大規模言語モデルとOceanBaseデータベースの連携を提供し、SQL文を実行できるようにします。適切なクライアントを利用することでプロジェクトのプロトタイプを迅速に構築でき、GitHub上でオープンソースとして公開されています。
Augment Codeは、開発者向けのAIプラットフォームです。コードの理解、問題のデバッグ、そしてコードベースを理解しているためより迅速なリリースを支援します。チャット、次の編集内容の提案、コードの自動補完を活用して、より多くの作業を効率的に進めることができます。
本記事では、Augment Codeを使用し、OceanBase MCP Serverを通じてバックエンドアプリケーションを迅速に構築する方法を紹介します。
前提条件
OceanBaseデータベースのデプロイが完了し、MySQLモードのユーザーテナントが作成されていること。テナント作成の詳細については、テナントの作成を参照してください。
Python 3.11以降のバージョンと対応するpipをインストールしていること。お使いのマシンのPythonバージョンが低い場合は、Minicondaを使用して新しいPython 3.11以降の環境を作成できます。詳細については、Minicondaインストールガイドを参照してください。
オペレーティングシステムに応じて、Gitをインストールします。お使いのOSに合わせて、ダウンロードしてインストールしてください：
Windows：https://git-scm.com/downloads/win。
macOS：https://git-scm.com/download/mac。
Linux：https://git-scm.com/downloads/linux。
Pythonパッケージマネージャーuvをインストールします。
macOS/Linuxでは、独立したスクリプトを使用してインストールできます：
curl -LsSf https://astral.sh/uv/install.sh | shWindowsプラットフォームでは、以下のスクリプトを使用します：
irm https://astral.sh/uv/install.ps1 | iexまたは、プラットフォームに依存しないpipインストール方法を使用することもできます：
pip install uv
インストールが完了したら、
uv --versionコマンドを実行してインストールが成功したかどうか確認します：pip install uv uv --versionAugment Codeのインストール：
VS Codeを開きます(なければ、公式ドキュメントを参照してインストールしてください)。
サイドバーの拡張機能アイコンをクリックします。
“Augment Code”を検索してインストールをクリックします。
Augment Codeアカウントに登録してログインします。
ステップ１：データベース接続情報を取得する
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/awesome-oceanbase-mcp.git
ソースコードディレクトリに移動します：
cd awesome-oceanbase-mcp
依存関係のインストール
oceanbase_mcp_server ディレクトリで以下のコマンドを実行して仮想環境を作成し、依存関係をインストールします：
uv venv
source .venv/bin/activate
uv pip install .
MCP Serverの追加と設定
Augment CodeでOceanBase MCP Serverを設定します。
Import from jsonをクリックし、MCP設定ファイルを入力します。
設定ファイルに情報を入力し、確認をクリックします。
/path/to/your/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/oceanbase_mcp_server/src/oceanbase_mcp_server", "run", "oceanbase_mcp_server" ], "env": { "OB_HOST": "***", "OB_PORT": "***", "OB_USER": "***", "OB_PASSWORD": "***", "OB_DATABASE": "***" } } } }データベースへの接続が可能か確認します。
「customerテーブルを作成してください。主キーはIDで、name、age、telephone、locationの各フィールドを含めてください」というプロンプトが表示されます。
Augment Codeは現在のtestデータベース内のテーブル数を表示します。これにより、OceanBaseデータベースへの接続が正常に行われていることが確認できます。
ステップ3：FastAPIを使用してRESTful APIスタイルのプロジェクトを迅速に作成する
FastAPIは、PythonでRESTful APIを迅速に構築できるWebフレームワークです。
テーブルを作成します。
「customerテーブルを作成してください。主キーはIDで、name、age、telephone、locationの各フィールドを含めてください」というプロンプトを入力します：
テストデータを挿入します。
「10件のテストデータを挿入してください」というプロンプトを入力します：
FastAPIプロジェクトを作成します。
「FastAPIプロジェクトを作成し、customerテーブルに基づいたRESTful APIを生成してください」というプロンプトを入力すると、複数のファイルが自動生成され、コードが修正され、FastAPIプロジェクトが起動します：
テーブル内のデータを確認します。
コマンドラインで
curl http://127.0.0.1:8000/customersを実行するか、他のリクエストツールを使用して、テーブル内のデータを確認します：curl http://127.0.0.1:8000/customers [{"name":"張三","age":28,"telephone":"13812345678","location":"北京市朝陽区","id":1},{"name":"李四","age":35,"telephone":"13987654321","location":"上海市浦東新区","id":2},{"name":"王五","age":42,"telephone":"15612345678","location":"廣州市天河區","id":3},{"name":"趙六","age":29,"telephone":"18712345678","location":"深圳市南山区","id":4},{"name":"錢七","age":33,"telephone":"13512345678","location":"杭州市西湖區","id":5},{"name":"孫八","age":26,"telephone":"15987654321","location":"成都市錦江區","id":6},{"name":"周九","age":38,"telephone":"18612345678","location":"武漢市江漢區","id":7},{"name":"吳十","age":31,"telephone":"13712345678","location":"南京市鼓樓區","id":8},{"name":"鄭十一","age":45,"telephone":"15812345678","location":"西安市雁塔區","id":9},{"name":"王十二","age":27,"telephone":"18512345678","location":"重慶市渝中區","id":10},{"name":"テストユーザー","age":25,"telephone":"13900000000","location":"テストアドレス","id":11}]追加、削除、変更、検索のコードがすべて生成されています。
from sqlalchemy.orm import Session from database import Customer from schemas import CustomerCreate, CustomerUpdate from typing import List, Optional def get_customer(db: Session, customer_id: int) -> Optional[Customer]: """IDに基づいて単一の顧客を取得します""" return db.query(Customer).filter(Customer.id == customer_id).first() def get_customers(db: Session, skip: int = 0, limit: int = 100) -> List[Customer]: """顧客リストを取得します""" return db.query(Customer).offset(skip).limit(limit).all() def create_customer(db: Session, customer: CustomerCreate) -> Customer: """新しい顧客を作成します""" db_customer = Customer(**customer.dict()) db.add(db_customer) db.commit() db.refresh(db_customer) return db_customer def update_customer(db: Session, customer_id: int, customer: CustomerUpdate) -> Optional[Customer]: """顧客情報を更新します""" db_customer = db.query(Customer).filter(Customer.id == customer_id).first() if db_customer: update_data = customer.dict(exclude_unset=True) for field, value in update_data.items(): setattr(db_customer, field, value) db.commit() db.refresh(db_customer) return db_customer def delete_customer(db: Session, customer_id: int) -> bool: """顧客を削除します""" db_customer = db.query(Customer).filter(Customer.id == customer_id).first() if db_customer: db.delete(db_customer) db.commit() return True return False def search_customers_by_name(db: Session, name: str) -> List[Customer]: """氏名に基づいて顧客を検索します""" return db.query(Customer).filter(Customer.name.contains(name)).all() def get_customers_by_location(db: Session, location: str) -> List[Customer]: """住所に基づいて顧客を取得します""" return db.query(Customer).filter(Customer.location.contains(location)).all()