Introduction: MCP + AI Models
The Model Context Protocol (MCP) is an emerging way to bridge external tools and AI models in a structured, secure, and developer-friendly way.
It allows you to define custom tools in your local code, annotate them with decorators like @mcp.tool()
, and have an AI model (like Claude or GPT) call them directly in conversation.
Think of MCP as an API gateway between your local logic and the AI — instead of hardcoding every answer into a prompt, you expose real functions that the model can execute.
In this article, we’ll build a Banking Database in MySQL, wrap it in a Python abstract layer, host it locally on MAMP, and then expose it via MCP tools to Claude Desktop for interactive querying.
Step 1: Designing the Banking Database
We’ll start with a simple schema containing three core tables:
- customers — stores personal details
- accounts — links customers to one or more accounts
- transactions — records credits, debits, and fund transfers
MySQL Schema
CREATE DATABASE banking_system;
USE banking_system;
CREATE TABLE customers (
customer_id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(100) NOT NULL,
email VARCHAR(100) UNIQUE NOT NULL,
phone VARCHAR(15),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);
CREATE TABLE accounts (
account_id INT AUTO_INCREMENT PRIMARY KEY,
customer_id INT NOT NULL,
account_number VARCHAR(20) UNIQUE NOT NULL,
balance DECIMAL(15, 2) DEFAULT 0.00,
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (customer_id) REFERENCES customers(customer_id) ON DELETE CASCADE
);
CREATE TABLE transactions (
transaction_id INT AUTO_INCREMENT PRIMARY KEY,
account_id INT NOT NULL,
transaction_type ENUM('credit', 'debit') NOT NULL,
amount DECIMAL(15, 2) NOT NULL,
description VARCHAR(255),
created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
FOREIGN KEY (account_id) REFERENCES accounts(account_id) ON DELETE CASCADE
);
Step 2: Hosting MySQL with MAMP
MAMP is an easy way to run Apache, PHP, and MySQL locally on macOS.
- Download MAMP from https://www.mamp.info/en/downloads/
- Install and open MAMP.app
- Click Start Servers — MySQL will run on port 8889 by default.
- Create the
banking_system
database via phpMyAdmin (http://localhost:8888/phpMyAdmin/
).
MAMP default MySQL credentials:
return mysql.connector.connect(
host="localhost",
port=8889, # important for MAMP
user="root",
password="root",
unix_socket="/Applications/MAMP/tmp/mysql/mysql.sock",
database="banking_system"
)
Step 3:Create the Project Folder
uv init ai-banking
This will:
- Create a folder named
ai-banking
- Initialize a
pyproject.toml
- Create a
README.md
- Create a
.gitignore
Step 4: Set Up a Virtual Environment
We’ll now create a virtual environment for our project and activate it:
cd ai-banking
uv venv
source .venv/bin/activate
uv add "mcp[cli]" mysql-connector-python faker
Step 4: Python Abstract Layer for Banking
We’ll write a Python API layer that:
- Handles CRUD for customers, accounts, and transactions
- Provides abstract business functions:
- Get customer summary
- Get transaction history
- Transfer funds
- Check transaction status
curd/create curd_helper.py:
import mysql.connector
from datetime import datetime
import json
# Database Connection
def get_db_connection():
"""
Establish a MySQL database connection.
Returns:
connection (mysql.connector.connection): MySQL connection object.
"""
return mysql.connector.connect(
host="localhost",
port=8889, # important for MAMP
user="root",
password="root",
unix_socket="/Applications/MAMP/tmp/mysql/mysql.sock",
database="banking_system"
)
# ---------------------------
# CRUD OPERATIONS
# ---------------------------
# ---- Customers ----
def create_customer(name, email, phone):
"""
Create a new customer in the database.
Args:
name (str): Full name of the customer.
email (str): Email address.
phone (str): Phone number.
Returns:
int: ID of the created customer.
"""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO customers (name, email, phone) VALUES (%s, %s, %s)",
(name, email, phone)
)
conn.commit()
customer_id = cursor.lastrowid
cursor.close()
conn.close()
return customer_id
def get_customer(customer_id):
"""
Retrieve customer details by ID.
Args:
customer_id (int): Customer ID.
Returns:
dict: Customer details.
"""
conn = get_db_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT * FROM customers WHERE customer_id = %s", (customer_id,))
result = cursor.fetchone()
cursor.close()
conn.close()
return result
# ---- Accounts ----
def create_account(customer_id, account_number, balance=0.0):
"""
Create a new bank account.
Args:
customer_id (int): Customer ID.
account_number (str): Unique account number.
balance (float): Initial balance.
Returns:
int: Account ID.
"""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO accounts (customer_id, account_number, balance) VALUES (%s, %s, %s)",
(customer_id, account_number, balance)
)
conn.commit()
account_id = cursor.lastrowid
cursor.close()
conn.close()
return account_id
def get_accounts_by_customer(customer_id):
"""
Get all accounts for a given customer.
Args:
customer_id (int): Customer ID.
Returns:
list: List of accounts.
"""
conn = get_db_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT * FROM accounts WHERE customer_id = %s", (customer_id,))
accounts = cursor.fetchall()
cursor.close()
conn.close()
return accounts
# ---- Transactions ----
def create_transaction(account_id, transaction_type, amount, description=""):
"""
Create a transaction for an account.
Args:
account_id (int): Account ID.
transaction_type (str): 'credit' or 'debit'.
amount (float): Transaction amount.
description (str): Transaction description.
Returns:
int: Transaction ID.
"""
conn = get_db_connection()
cursor = conn.cursor()
cursor.execute(
"INSERT INTO transactions (account_id, transaction_type, amount, description) VALUES (%s, %s, %s, %s)",
(account_id, transaction_type, amount, description)
)
# Update account balance
if transaction_type == "credit":
cursor.execute("UPDATE accounts SET balance = balance + %s WHERE account_id = %s", (amount, account_id))
elif transaction_type == "debit":
cursor.execute("UPDATE accounts SET balance = balance - %s WHERE account_id = %s", (amount, account_id))
conn.commit()
transaction_id = cursor.lastrowid
cursor.close()
conn.close()
return transaction_id
def transfer_funds(from_account_id, to_account_id, amount, description="Fund Transfer"):
"""
Transfer funds from one account to another.
Args:
from_account_id (int): Source account ID.
to_account_id (int): Destination account ID.
amount (float): Transfer amount.
description (str): Description.
Returns:
bool: True if success, False otherwise.
"""
conn = get_db_connection()
cursor = conn.cursor()
try:
cursor.execute("SELECT balance FROM accounts WHERE account_id = %s", (from_account_id,))
balance = cursor.fetchone()[0]
if balance < amount:
raise ValueError("Insufficient funds.")
create_transaction(from_account_id, "debit", amount, description)
create_transaction(to_account_id, "credit", amount, description)
conn.commit()
return True
except Exception as e:
conn.rollback()
print("Transfer failed:", e)
return False
finally:
cursor.close()
conn.close()
Step 5: Seeding Test Data
Using faker
, we can quickly populate the database:
faker.py
from faker import Faker
from curd.curd_helper import create_customer, create_account, create_transaction, transfer_funds
import random
fake = Faker()
def seed_random_data():
"""
Populate the database with random test data:
- 25 customers
- Each with 1 to 3 accounts
- 500 total transactions (including inter-account transfers)
"""
customer_ids = []
account_ids = []
# Step 1: Create 25 random customers
for _ in range(25):
name = fake.name()
email = fake.unique.email()
phone = fake.msisdn()[:10]
cust_id = create_customer(name, email, phone)
customer_ids.append(cust_id)
# Step 2: Create accounts for each customer
for cust_id in customer_ids:
for _ in range(random.randint(1, 3)): # Each customer has 1–3 accounts
account_number = str(random.randint(1000000000, 9999999999))
balance = round(random.uniform(1000, 100000), 2)
acc_id = create_account(cust_id, account_number, balance)
account_ids.append(acc_id)
# Step 3: Create 500 total transactions
for i in range(500):
if random.random() < 0.2: # 20% fund transfers
from_acc, to_acc = random.sample(account_ids, 2)
amount = round(random.uniform(10, 2000), 2)
transfer_funds(from_acc, to_acc, amount, description="Random Fund Transfer")
else:
acc_id = random.choice(account_ids)
transaction_type = random.choice(["credit", "debit"])
amount = round(random.uniform(10, 5000), 2)
description = fake.sentence(nb_words=4)
create_transaction(acc_id, transaction_type, amount, description)
print("✅ Database seeded with random data successfully!")
# Run the seeding
if __name__ == "__main__":
seed_random_data()
Execute the fake and make sure the data are filled up with database
uv run faker.py
Step 6: Exposing Functions as MCP Tools
The magic of MCP comes from the @mcp.tool()
decorator, which makes functions callable by the AI model.
tools/tool.py
import json
from curd.curd_helper import get_db_connection, get_customer, get_accounts_by_customer, create_transaction
from server import mcp
from decimal import Decimal
# ---------------------------
# ABSTRACT BANKING FUNCTIONS
# ---------------------------
# @mcp.route("/customer//summary", methods=["GET"])
# @mcp.auth_required
@mcp.tool()
def get_customer_summary(customer_id):
"""
Get logged-in customer summary in JSON.
Args:
customer_id (int): Customer ID.
Returns:
str: JSON string containing customer info, account count, and recent transactions.
"""
customer = get_customer(customer_id)
accounts = get_accounts_by_customer(customer_id)
summary = {
"customer": customer,
"number_of_accounts": len(accounts),
"accounts": accounts
}
return json.dumps(summary, default=str)
# @mcp.route("/customer//accounts", methods=["GET"])
# @mcp.auth_required
@mcp.tool()
def make_transfer_funds(from_account_id, to_account_id, amount, description="Fund Transfer"):
"""
Transfer funds from one account to another.
Args:
from_account_id (int): Source account ID.
to_account_id (int): Destination account ID.
amount (float): Transfer amount.
description (str): Description.
Returns:
bool: True if success, False otherwise.
"""
conn = get_db_connection()
cursor = conn.cursor()
try:
cursor.execute("SELECT balance FROM accounts WHERE account_id = %s", (from_account_id,))
balance = cursor.fetchone()[0]
if float(balance) < float(amount):
raise ValueError("Insufficient funds.")
create_transaction(from_account_id, "debit", amount, description)
create_transaction(to_account_id, "credit", amount, description)
conn.commit()
return(f"Transfer of {amount} from account {from_account_id} to {to_account_id} successful.")
except Exception as e:
conn.rollback()
return(f"Transfer failed:", e)
finally:
cursor.close()
conn.close()
# @mcp.route("/account//transactions", methods=["GET"])
# @mcp.auth_required
@mcp.tool()
def get_transactions(account_id, start_date=None, end_date=None):
"""
Retrieve transactions for an account.
Args:
account_id (int): Account ID.
start_date (str): Optional. Format 'YYYY-MM-DD'.
end_date (str): Optional. Format 'YYYY-MM-DD'.
Returns:
list: Transactions.
"""
conn = get_db_connection()
cursor = conn.cursor(dictionary=True)
if start_date and end_date:
cursor.execute(
"SELECT * FROM transactions WHERE account_id = %s AND DATE(created_at) BETWEEN %s AND %s ORDER BY created_at DESC",
(account_id, start_date, end_date)
)
else:
cursor.execute(
"SELECT * FROM transactions WHERE account_id = %s ORDER BY created_at DESC LIMIT 100",
(account_id,)
)
transactions = cursor.fetchall()
cursor.close()
conn.close()
return transactions
@mcp.tool()
def get_transaction_status(transaction_id):
"""
Retrieve the status and details of a given transaction.
Args:
transaction_id (int): The unique ID of the transaction.
Returns:
dict: {
"status": str, # "completed" or "not found"
"transaction": dict # Transaction details if found
}
"""
conn = get_db_connection()
cursor = conn.cursor(dictionary=True)
cursor.execute("SELECT * FROM transactions WHERE transaction_id = %s", (transaction_id,))
transaction = cursor.fetchone()
cursor.close()
conn.close()
if transaction:
return {
"status": "completed",
"transaction": transaction
}
else:
return {
"status": "not found",
"transaction": None
}
Keep server code aside
server.py
from mcp.server.fastmcp import FastMCP
mcp = FastMCP("ai_banking")
main.py
from server import mcp
import tools.tool
if __name__ == "__main__":
mcp.run()
When the MCP server runs, Claude (or any AI with MCP support) will see these tools and can invoke them in real-time.
Step 7: Configuring Claude Desktop
- Install Claude Desktop from Anthropic’s website.
- Go to Settings → Developer Tools → MCP Servers.
- Add your Python MCP server configuration:
{
"mcpServers": {
"ai_banking": {
"disabled": false,
"timeout": 60,
"type": "stdio",
"command": "uv",
"args": [
"--directory",
"/Users/apple/LLM/ai_banking",
"run",
"main.py"
]
}
}
}
- Restart Claude Desktop.
- Open a chat and type something like:
- “Get the summary for customer 5”
The model will automatically callget_customer_summary_tool()
.
Step 7: Testing the Integration
With the MCP tools exposed, Claude Desktop now has live access to your banking system.
Example conversation:
User: Get the last 5 transactions for account 102
Claude: (calls `get_transactions_tool`)
Claude: Here are the last 5 transactions...
Conclusion
This project shows how you can:
- Host a local MySQL database on MAMP
- Create an abstract Python API layer
- Use Faker to generate realistic test data
- Expose your functions via MCP tools
- Connect them to Claude Desktop for natural language interaction
The result is an AI assistant that can query, summarize, and update your database in real-time — a foundation you can adapt to many domains beyond banking.