OceanBase MCP ServerとClaude Codeの連携

claude mcp add ob-mcp-server --transport stdio \
  -e OB_HOST=*** \
  -e OB_PORT=*** \
  -e OB_USER=*** \
  -e OB_PASSWORD=*** \
  -e OB_DATABASE=*** \
  -- uvx --from oceanbase-mcp oceanbase_mcp_server

claude

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

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

customerテーブルに10件のデータを挿入してください。

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

uv venv
source .venv/bin/activate
uv pip install -r requirements.txt

uvicorn main:app --reload

curl "http://127.0.0.1:8000/customers"

from fastapi import FastAPI, HTTPException
from typing import List
from database import get_connection
from models import Customer, CustomerCreate, CustomerUpdate

app = FastAPI(
    title="Customer API",
    description="customerテーブルに基づくRESTful API",
    version="1.0.0"
)

@app.get("/")
def root():
    """APIのルートパス"""
    return {"message": "Welcome to Customer API", "docs": "/docs"}

@app.get("/customers", response_model=List[Customer], summary="すべての顧客を取得する")
def get_all_customers():
    """すべての顧客リストを取得する"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute("SELECT id, name, age, telephone, location FROM customer ORDER BY id")
        customers = cursor.fetchall()
        return customers
    finally:
        cursor.close()
        conn.close()

@app.get("/customers/{customer_id}", response_model=Customer, summary="単一顧客の取得")
def get_customer(customer_id: int):
    """IDに基づいて単一顧客を取得します"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(
            "SELECT id, name, age, telephone, location FROM customer WHERE id = %s",
            (customer_id,)
        )
        customer = cursor.fetchone()
        if not customer:
            raise HTTPException(status_code=404, detail="顧客が存在しません")
        return customer
    finally:
        cursor.close()
        conn.close()

@app.post("/customers", response_model=Customer, summary="顧客の作成", status_code=201)
def create_customer(customer: CustomerCreate):
    """新規顧客を作成します"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        cursor.execute(
            """INSERT INTO customer (name, age, telephone, location)
               VALUES (%s, %s, %s, %s)""",
            (customer.name, customer.age, customer.telephone, customer.location)
        )
        conn.commit()
        new_id = cursor.lastrowid
        # 新規作成された顧客を取得します
        cursor.execute(
            "SELECT id, name, age, telephone, location FROM customer WHERE id = %s",
            (new_id,)
        )
        return cursor.fetchone()
    finally:
        cursor.close()
        conn.close()

@app.put("/customers/{customer_id}", response_model=Customer, summary="顧客情報を更新する")
def update_customer(customer_id: int, customer: CustomerUpdate):
    """顧客情報を更新します"""
    conn = get_connection()
    cursor = conn.cursor(dictionary=True)
    try:
        # 顧客が存在するか確認します
        cursor.execute("SELECT id FROM customer WHERE id = %s", (customer_id,))
        if not cursor.fetchone():
            raise HTTPException(status_code=404, detail="顧客が存在しません")

        # 動的更新ステートメントを構築します
        update_fields = []
        values = []
        if customer.name is not None:
            update_fields.append("name = %s")
            values.append(customer.name)
        if customer.age is not None:
            update_fields.append("age = %s")
            values.append(customer.age)
        if customer.telephone is not None:
            update_fields.append("telephone = %s")
            values.append(customer.telephone)
        if customer.location is not None:
            update_fields.append("location = %s")
            values.append(customer.location)

        if update_fields:
            values.append(customer_id)
            sql = f"UPDATE customer SET {', '.join(update_fields)} WHERE id = %s"
            cursor.execute(sql, values)
            conn.commit()

        # 更新後の顧客を返します
        cursor.execute(
            "SELECT id, name, age, telephone, location FROM customer WHERE id = %s",
            (customer_id,)
        )
        return cursor.fetchone()
    finally:
        cursor.close()
        conn.close()

@app.delete("/customers/{customer_id}", summary="顧客を削除する", status_code=204)
def delete_customer(customer_id: int):
    """顧客を削除します"""
    conn = get_connection()
    cursor = conn.cursor()
    try:
        cursor.execute("DELETE FROM customer WHERE id = %s", (customer_id,))
        if cursor.rowcount == 0:
            raise HTTPException(status_code=404, detail="顧客は存在しません")
        conn.commit()
        return None
    finally:
        cursor.close()
        conn.close()

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