MCP (Model Context Protocol) is a protocol released and open-sourced by Anthropic in November 2024. It enables large language models to interact with external tools or data sources. With MCP, users no longer need to manually copy and execute the output of a large model. Instead, the large model can directly command tools to perform specific actions (Actions).
OceanBase MCP Server provides the capability for large models to interact with OceanBase Database through the MCP protocol. It can execute SQL statements and help developers quickly build prototype applications.
Claude Code is an AI programming assistant developed by Anthropic. It assists developers in code generation, refactoring, and debugging within the terminal. This topic will demonstrate how to integrate OceanBase MCP Server with Claude Code and quickly generate a FastAPI-style RESTful API project based on the database.
Prerequisites
You have a transactional instance available in your environment, and a MySQL-compatible tenant under the instance.
You have a MySQL database and account available under the tenant, and you have granted read and write permissions to the database account. For more information, see Create an account and Create a database (MySQL only).
You have the project administrator or instance administrator role to read and write to instances in the project. If you do not have the required permissions, contact your organization administrator to add the permissions.
Install Python 3.11 or later and the corresponding pip. If your machine has a low Python version, you can use Miniconda to create a new Python 3.11 or later environment. For more information, see Miniconda installation guide.
Install the Python package manager uv.
Install Claude Code.
Step 1: Get an OceanBase Cloud connection string
Log in to the OceanBase Cloud console. On the instance list, expand the target transactional instance. Under the target tenant, choose Connect > Get Connection String.
In the dialog, select Public Network.
On the Connect via Public Network page, complete the following to generate a connection string:
(Optional) If the tenant does not have a public endpoint yet, under Step 1: Get Public Endpoint, click Get Public Endpoint. If one already exists, skip this step. After the endpoint is ready, click Next Step.
Under Step 2: Security Settings, click Add to configure the allowlist as needed, then click Next Step.
Under Step 3: Access Database, set options according to your tenant mode:
MySQL-compatible tenant: Select the database and account you created in the prerequisites, and choose MySQL CLI as the connection method. Then copy the connection string for the next section.
Oracle-compatible tenant: Select the account you created in the prerequisites, and choose OBClient as the connection method. Then copy the connection string for the next section.
Step 2: Configure OceanBase MCP Server in Claude Code
Create a working directory
Manually create a working directory for Claude Code and open it, for example, claudecode-generate. The project files generated by Claude Code will be saved in this directory.
Add MCP Server configuration
Run the claude mcp add-json command in the terminal to add OceanBase MCP Server configuration:
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
Replace the following parameters with your own database information:
OB_HOSTOB_PORTOB_USEROB_PASSWORDOB_DATABASE
Test MCP connection
Start Claude Code in the project root directory:
claude
In the Claude Code dialog box, enter the following prompt:
test How many tables are there in the test database?
If the correct result is returned, it indicates that Claude Code has successfully accessed the database through OceanBase MCP Server.
Step 3: Generate a FastAPI project with Claude Code
You can use FastAPI to quickly create a RESTful API project. FastAPI is a Python web framework that allows you to build RESTful APIs quickly.
Create a sample table
In the Claude Code dialog box, enter the following prompt:
Create a customer table with the ID as the primary key and the name, age, telephone, and location fields.
Insert test data
Continue to enter the following prompt:
Insert 10 rows of data into the customer table.
Generate RESTful API code
Enter the following prompt to let Claude Code generate a FastAPI project based on the customer table:
Create a FastAPI project and generate a RESTful API based on the customer table.
After the code is generated, accept and save the generated files.
Configure the database connection
Modify the connection pool configuration in database.py and replace the database connection parameters with your actual OceanBase information.
Create a virtual environment and install dependencies
In the current directory, execute the following command to create a virtual environment and install the required packages using the uv package manager:
uv venv
source .venv/bin/activate
uv pip install -r requirements.txt
Start the FastAPI service
Execute the following command to start the project (please adjust the entry file name as needed):
uvicorn main:app --reload
Verify the API
After the service is started, you can verify it as follows:
- Access
http://127.0.0.1:8000/docsto view the Swagger documentation. - Use
curlor Postman to call the CRUD interfaces and verify the results.
For example, to query all customers:
curl "http://127.0.0.1:8000/customers"
After completing all the steps above, you will have a FastAPI RESTful API project generated by Claude Code and driven by OceanBase, capable of performing common operations such as adding, deleting, modifying, and querying data in the customer table.
Appendix: Complete RESTful API code example
After the RESTful API for creating, updating, deleting, and retrieving customers is successfully generated, the code example is as follows:
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="RESTful API based on the customer table",
version="1.0.0"
)
@app.get("/")
def root():
"""API root path"""
return {"message": "Welcome to Customer API", "docs": "/docs"}
@app.get("/customers", response_model=List[Customer], summary="Get all customers")
def get_all_customers():
"""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="Get a single customer")
def get_customer(customer_id: int):
"""Get a single customer by 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="Customer not found")
return customer
finally:
cursor.close()
conn.close()
@app.post("/customers", response_model=Customer, summary="Create a customer", status_code=201)
def create_customer(customer: CustomerCreate):
"""Create a new customer"""
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
# Get the newly created customer
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="Update a customer")
def update_customer(customer_id: int, customer: CustomerUpdate):
"""Update customer information"""
conn = get_connection()
cursor = conn.cursor(dictionary=True)
try:
# Check if the customer exists
cursor.execute("SELECT id FROM customer WHERE id = %s", (customer_id,))
if not cursor.fetchone():
raise HTTPException(status_code=404, detail="Customer not found")
# Build dynamic update statement
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()
# Return the updated customer
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="Delete customer", status_code=204)
def delete_customer(customer_id: int):
"""Delete customer"""
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="Customer not found")
conn.commit()
return None
finally:
cursor.close()
conn.close()
if __name__ == "__main__":
import uvicorn
uvicorn.run(app, host="0.0.0.0", port=8000)