Integrate OceanBase MCP Server with Kiro

MCP (Model Context Protocol) is a protocol developed and open-sourced by Anthropic in November 2024. It enables large language models (LLMs) to interact with external tools or data sources. With MCP, users no longer need to manually copy and execute the output of an LLM. Instead, the LLM can directly instruct a tool to perform the corresponding action (Action).

The MCP Server provides the capability for large models to interact with seekdb through the MCP protocol and execute SQL statements. With the right client, you can quickly build a project prototype. It is open-sourced on GitHub.

Kiro is an integrated development environment (IDE) designed for AI agents, launched by Amazon Web Services (AWS). It is a programming tool with AI capabilities, designed to help developers complete the entire development process from concept to production deployment.

This topic shows you how to build a backend application quickly by using OceanBase MCP Server and Kiro.

Prerequisites

• You have deployed seekdb.

• Install Python 3.11 or later and the corresponding pip. If your machine has a lower Python version, you can use Miniconda to create a new Python 3.11 or later environment. For more information, see Miniconda Installation Guide.

• Install Git based on your operating system.

• Install the uv package manager. After the installation is complete, run the uv --version command to verify the installation:

  pip install uv
  uv --version

• Kiro client:

  You can download the Kiro client from the kiro download page and install the version that matches your operating system.

Step 1: Obtain database connection information

Contact the seekdb deployment personnel or administrator to obtain the database connection string. For example:

mysql -h$host -P$port -u$user_name -p$password -D$database_name

Parameter description:

• $host: the IP address for connecting to seekdb.

• $port: the port for connecting to seekdb. The default value is 2881.

• $database_name: the name of the database to be accessed.

  tip

  The user used for the connection must have the CREATE, INSERT, DROP, and SELECT permissions on the database.

• $user_name: the database connection account.

• $password: the account password.

Step 2: Configure OceanBase MCP Server

Clone the OceanBase MCP Server repository

Run the following command to download the source code to your local machine:

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

Go to the source code directory:

cd awesome-oceanbase-mcp

Install dependencies

Run the following command in the oceanbase_mcp_server directory to create a virtual environment and install the dependencies:

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

Configure OceanBase MCP Server in Kiro

1. Press Command + L (MacOS) to open the chat dialog box, click the gear icon in the lower-left corner, and select Kiro Settings.

   

2. Click Open User MCP Config (JSON)/Open Workspace MCP Config (JSON) and fill in the mcp configuration file.

   

   

   Fill in the following configuration file and click OK. Replace /path/to/your/oceanbase_mcp_server with the absolute path of the oceanbase_mcp_server folder, and replace OB_HOST, OB_PORT, OB_USER, OB_PASSWORD, and OB_DATABASE with the corresponding information of your 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": "***"
      }
     }
    }
   }

3. Verify whether you can connect to the database.

   Enter "How many tables are there in the test database?" in the prompt. Kiro will display the SQL statement to be executed and output the query result:

   

   Kiro will display the number of tables in the test database, indicating that you can connect to seekdb.

Step 3: Use FastAPI to create a RESTful API project

FastAPI is a Python web framework that allows you to quickly build RESTful APIs.

1. Create a table.

Enter the prompt "Create a customer table with ID as the primary key and name, age, telephone, and location as fields":

2. Insert test data.

Enter the prompt "Insert 10 test records":

3. Create a FastAPI project.

Enter the prompt "Create a FastAPI project and generate a RESTful API based on the customer table". The system automatically generates multiple files. Click Accept All to accept them. If there are issues, you can modify them later.

Since the files generated by AI may vary each time, the modification process is not shown here.

4. Run the following command to create a virtual environment using the uv package manager and install the dependencies in the current directory.

   pip install -r requirements.txt

5. Start the FastAPI project.

   python3 main.py

6. View the data in the table.

Run the curl http://127.0.0.1:8000/customers command in the command line or use other request tools to view the data in the table:

   curl http://127.0.0.1:8000/customers
   [{"name":"Zhang San","age":28,"telephone":"13812345678","location":"Chaoyang District, Beijing","id":1},{"name":"Li Si","age":35,"telephone":"13987654321","location":"Pudong New Area, Shanghai","id":2},{"name":"Wang Wu","age":42,"telephone":"15612345678","location":"Tianhe District, Guangzhou","id":3},{"name":"Zhao Liu","age":29,"telephone":"18712345678","location":"Nanshan District, Shenzhen","id":4},{"name":"Qian Qi","age":33,"telephone":"13512345678","location":"West Lake District, Hangzhou","id":5},{"name":"Sun Ba","age":26,"telephone":"15987654321","location":"Jinjiang District, Chengdu","id":6},{"name":"Zhou Jiu","age":38,"telephone":"18612345678","location":"Jianghan District, Wuhan","id":7},{"name":"Wu Shi","age":31,"telephone":"13712345678","location":"Gulou District, Nanjing","id":8},{"name":"Zheng Shiyi","age":27,"telephone":"15812345678","location":"Yanta District, Xi'an","id":9},{"name":"Wang Shier","age":45,"telephone":"18512345678","location":"Yuzhong District, Chongqing","id":10}]

7. The code for adding, deleting, modifying, and querying data has been generated.

   from database import db
   from models import CustomerCreate, CustomerUpdate
   from typing import List, Optional

class CustomerCRUD:

@staticmethod
    def get_all_customers() -> List[dict]:
        query = "SELECT * FROM customer ORDER BY id"
        return db.execute_query(query)

@staticmethod
    def get_customer_by_id(customer_id: int) -> Optional[dict]:
        query = "SELECT * FROM customer WHERE id = %s"
        result = db.execute_query(query, (customer_id,))
        return result[0] if result else None

@staticmethod
    def create_customer(customer: CustomerCreate) -> dict:
        query = """
        INSERT INTO customer (name, age, telephone, location)
        VALUES (%s, %s, %s, %s)
        """
        customer_id = db.execute_insert(
            query,
            (customer.name, customer.age, customer.telephone, customer.location)
        )
        return CustomerCRUD.get_customer_by_id(customer_id)

@staticmethod
    def update_customer(customer_id: int, customer: CustomerUpdate) -> Optional[dict]:
        # Build dynamic update query
        update_fields = []
        params = []

if customer.name is not None:
            update_fields.append("name = %s")
            params.append(customer.name)
        if customer.age is not None:
            update_fields.append("age = %s")
            params.append(customer.age)
        if customer.telephone is not None:
            update_fields.append("telephone = %s")
            params.append(customer.telephone)
        if customer.location is not None:
            update_fields.append("location = %s")
            params.append(customer.location)

if not update_fields:
            return CustomerCRUD.get_customer_by_id(customer_id)

query = f"UPDATE customer SET {', '.join(update_fields)} WHERE id = %s"
        params.append(customer_id)

db.execute_query(query, tuple(params))
        return CustomerCRUD.get_customer_by_id(customer_id)

@staticmethod
    def delete_customer(customer_id: int) -> bool:
        query = "DELETE FROM customer WHERE id = %s"
        db.execute_query(query, (customer_id,))
        return True

customer_crud = CustomerCRUD()