OceanBase MCP ServerとOpenCodeの連携

{
  "$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": "***"
      },
    },
  },
}

opencode

test 庫中有多少張表？

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

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

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

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

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