OceanBase MCP Server 与 Cline 集成

MCP（Model Context Protocol） 是 Anthropic 公司于 2024 年 11 月推出并开源，旨在实现大语言模型与外部工具或数据源交互的协议。通过 MCP，用户不需要将大模型的输出手动复制执行，大模型可以直接指挥工具执行相应的动作（Action）。

OceanBase MCP Server 通过 MCP 协议提供了大模型与 OceanBase 数据库交互的能力，可以执行 SQL 语句。通过合适的客户端可以快速搭建项目原型，已在 github 上开源。

Cline 是一个开源的支持 MCP 协议的 AI 编码助手。

本文使用 Cline，展示如何通过 OceanBase MCP Server 快速构建一个后端应用。

前提条件

• 您已完成部署 OceanBase 数据库并且创建了 MySQL 模式用户租户。创建用户租户的详细信息，参见 创建租户。

• 安装 Python 3.11 及以上版本 和相应 pip。如果您的机器上 Python 版本较低，可以使用 Miniconda 来创建新的 Python 3.11 及以上的环境，具体可参考 Miniconda 安装指南。

• 根据所用的操作系统，安装 Git。

• 安装 Python 包管理器 uv。安装完成后，可使用 uv --version 命令验证安装是否成功：

  pip install uv
  uv --version
  

• 安装 Cline：

  • 如使用 Visual Studio Code IDE，可在 扩展 栏中搜索安装 Cline 插件，插件名称为 Cline。安装完成后，点击 设置图标，为 Cline 配置大模型 API：

    

  • 如无 IDE，则参考 Cline 安装指南，下载地址为 Cline。

步骤一：获取数据库连接信息

联系 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 租户。

步骤二：配置 OceanBase MCP Server

这里以 Visual Studio Code 为例，展示如何配置 OceanBase MCP Server。

克隆 OceanBase MCP Server 仓库

执行下面的命令将源代码下载到本地：

git clone https://github.com/oceanbase/mcp-oceanbase.git

进入源代码目录：

cd mcp-oceanbase

安装依赖

在 mcp-oceanbase 目录下执行下面的命令创建虚拟环境，并安装依赖：

uv venv
source .venv/bin/activate
uv pip install .

创建 Visual Studio Code 的工作目录

手动在本地创建一个 Visual Studio Code 的工作目录，并用 Visual Studio Code 打开，后面 Cline 生成的文件将放在这个目录下，示例的目录名为 cline-generate。

在 Cline 插件中配置 OceanBase MCP Server

单击侧边栏中的 Cline 图标，打开 Cline 对话框。

添加并配置 MCP Servers

1. 点击下图所示的 MCP Servers 图标。

   

2. 按照下图序号指引，手动配置 OceanBase MCP Server。

   

3. 填写配置文件。

   在上一个步骤中打开的 cline_mcp_settings.json 文件中，输入下面的配置信息，然后保存该文件。/path/to/your/mcp-oceanbase/src/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/mcp-oceanbase/src/oceanbase_mcp_server",
           "run",
           "oceanbase_mcp_server"
       ],
       "env": {
           "OB_HOST": "***",
           "OB_PORT": "***",
           "OB_USER": "***",
           "OB_PASSWORD": "***",
           "OB_DATABASE": "***"
       }
       }
   }
   }
   

4. 如配置成功，将显示 可使用 状态，并且会显示出 MCP tools 和 resources 信息，如下所示：

   

5. 点击下图所示的开关按钮，使其可以被 Cline 使用：

   

测试 MCP Server

如下打开 Cline 会话框，输入提示“test库中有多少张表”，Cline 会展示即将执行的 SQL 语句，如果无问题，点击运行。

Cline 会展示当前 test 库中的表名称，说明可以正常连接 OceanBase 数据库。

!6](https://obbusiness-private.oss-cn-shanghai.aliyuncs.com/doc/img/cloud/integrations/AI/cline-10.png)

使用 FastAPI 快速创建 RESTful API 风格的项目

你可以使用 FastAPI 快速创建 RESTful API 风格的项目。FastAPI 是一个 Python 的 Web 框架，可以快速构建 RESTful API。

1. 创建 customer 表

   在对话框中输入提示：创建一个 customer 表，主键是 ID，包含 name，age，telephone，location 字段，确认 SQL 语句后，点击 运行 按钮执行查询。

   

2. 插入测试数据

   在对话框中输入提示：插入 10 条数据，确认 SQL 语句后，点击 运行 按钮执行查询。

   

   插入成功后，会展示执行结果：

   

3. 创建 FastAPI 项目

   在对话框中输入提示：创建一个 FastAPI 项目，生成基于 customer 表的 RESTful API，确认 SQL 语句后，点击 运行 按钮执行查询。

   

   此步骤将自动生成三个文件。建议首次使用时选择"全部接受"，因为 AI 生成的文件内容可能具有不确定性，后续可根据实际需求进行调整。

4. 创建虚拟环境并安装依赖

   执行如下命令，在当前目录下使用 uv 包管理工具创建虚拟环境，并安装依赖包：

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

5. 启动 FastAPI 项目

   执行如下命令，启动 FastAPI 项目：

   uvicorn main:app --reload
   

6. 查看表中数据

   在命令行中运行如下命令，或者使用其他请求工具，查看表中的数据：

   curl http://127.0.0.1:8000/customers
   

   返回结果如下：

   [{"telephone":"11111111101","id":1,"name":"张三","location":"北京","age":25},{"telephone":"11111111102","id":2,"name":"李四","location":"上海","age":30},{"telephone":"11111111103","id":3,"name":"王五","location":"广州","age":22},{"telephone":"11111111104","id":4,"name":"赵六","location":"深圳","age":28},{"telephone":"11111111105","id":5,"name":"钱七","location":"杭州","age":35},{"telephone":"11111111106","id":6,"name":"孙八","location":"南京","age":40},{"telephone":"11111111107","id":7,"name":"周九","location":"成都","age":27},{"telephone":"11111111108","id":8,"name":"吴十","location":"武汉","age":33},{"telephone":"11111111109","id":9,"name":"郑十一","location":"西安","age":29},{"telephone":"11111111110","id":10,"name":"王十二","location":"重庆","age":31}]
   

   可以看到增、删、改、查的 RESTful API 已经成功生成：

   from fastapi import FastAPI, Depends, HTTPException
   from sqlalchemy.orm import Session
   from models import Customer
   from database import SessionLocal, engine
   from pydantic import BaseModel
   
   app = FastAPI()
   
   # 数据库依赖
   def get_db():
       db = SessionLocal()
       try:
           yield db
       finally:
           db.close()
   
   # 请求模型
   class CustomerCreate(BaseModel):
       name: str
       age: int
       telephone: str
       location: str
   
   # 响应模型
   class CustomerResponse(CustomerCreate):
       id: int
   
       class Config:
           from_attributes = True
   
   @app.post("/customers/")
   def create_customer(customer: CustomerCreate, db: Session = Depends(get_db)):
       db_customer = Customer(**customer.model_dump())
       db.add(db_customer)
       db.commit()
       db.refresh(db_customer)
       return db_customer
   
   @app.get("/customers/{customer_id}")
   def read_customer(customer_id: int, db: Session = Depends(get_db)):
       customer = db.query(Customer).filter(Customer.id == customer_id).first()
       if customer is None:
           raise HTTPException(status_code=404, detail="Customer not found")
       return customer
   
   @app.get("/customers/")
   def read_customers(skip: int = 0, limit: int = 10, db: Session = Depends(get_db)):
       return db.query(Customer).offset(skip).limit(limit).all()
   
   @app.put("/customers/{customer_id}")
   def update_customer(customer_id: int, customer: CustomerCreate, db: Session = Depends(get_db)):
       db_customer = db.query(Customer).filter(Customer.id == customer_id).first()
       if db_customer is None:
           raise HTTPException(status_code=404, detail="Customer not found")
       for field, value in customer.model_dump().items():
           setattr(db_customer, field, value)
       db.commit()
       db.refresh(db_customer)
       return db_customer
   
   @app.delete("/customers/{customer_id}")
   def delete_customer(customer_id: int, db: Session = Depends(get_db)):
       customer = db.query(Customer).filter(Customer.id == customer_id).first()
       if customer is None:
           raise HTTPException(status_code=404, detail="Customer not found")
       db.delete(customer)
       db.commit()
       return {"message": "Customer deleted successfully"}