OceanBase MCP ServerとGitHub Copilotの連携

{
  "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": []
}

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, 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)