# Directory Structure
```
├── .env.sample
├── .gitignore
├── .python-version
├── LICENSE
├── pyproject.toml
├── README.md
├── src
│ └── mcp_server_mariadb
│ ├── __init__.py
│ └── server.py
├── tests
│ ├── __init__.py
│ └── test_db.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.10
```
--------------------------------------------------------------------------------
/.env.sample:
--------------------------------------------------------------------------------
```
MARIADB_HOST=localhost
MARIADB_USER=test_user
MARIADB_PASSWORD=test_password
MARIADB_DATABASE=test_db
MARIADB_PORT=3306
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Byte-compiled / optimized / DLL files
__pycache__/
*.py[cod]
*$py.class
# C extensions
*.so
# Distribution / packaging
.Python
build/
develop-eggs/
dist/
downloads/
eggs/
.eggs/
lib/
lib64/
parts/
sdist/
var/
wheels/
share/python-wheels/
*.egg-info/
.installed.cfg
*.egg
MANIFEST
# PyInstaller
# Usually these files are written by a python script from a template
# before PyInstaller builds the exe, so as to inject date/other infos into it.
*.manifest
*.spec
# Installer logs
pip-log.txt
pip-delete-this-directory.txt
# Unit test / coverage reports
htmlcov/
.tox/
.nox/
.coverage
.coverage.*
.cache
nosetests.xml
coverage.xml
*.cover
*.py,cover
.hypothesis/
.pytest_cache/
cover/
# Translations
*.mo
*.pot
# Django stuff:
*.log
local_settings.py
db.sqlite3
db.sqlite3-journal
# Flask stuff:
instance/
.webassets-cache
# Scrapy stuff:
.scrapy
# Sphinx documentation
docs/_build/
# PyBuilder
.pybuilder/
target/
# Jupyter Notebook
.ipynb_checkpoints
# IPython
profile_default/
ipython_config.py
# pyenv
# For a library or package, you might want to ignore these files since the code is
# intended to run in multiple environments; otherwise, check them in:
# .python-version
# pipenv
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
# However, in case of collaboration, if having platform-specific dependencies or dependencies
# having no cross-platform support, pipenv may install dependencies that don't work, or not
# install all needed dependencies.
#Pipfile.lock
# UV
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
#uv.lock
# poetry
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
# This is especially recommended for binary packages to ensure reproducibility, and is more
# commonly ignored for libraries.
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
#poetry.lock
# pdm
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
#pdm.lock
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
# in version control.
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
.pdm.toml
.pdm-python
.pdm-build/
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
__pypackages__/
# Celery stuff
celerybeat-schedule
celerybeat.pid
# SageMath parsed files
*.sage.py
# Environments
.env
.venv
env/
venv/
ENV/
env.bak/
venv.bak/
# Spyder project settings
.spyderproject
.spyproject
# Rope project settings
.ropeproject
# mkdocs documentation
/site
# mypy
.mypy_cache/
.dmypy.json
dmypy.json
# Pyre type checker
.pyre/
# pytype static type analyzer
.pytype/
# Cython debug symbols
cython_debug/
# PyCharm
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
# and can be added to the global gitignore or merged into this file. For a more nuclear
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
#.idea/
# PyPI configuration file
.pypirc
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# mcp-server-mariadb
An MCP server implementation for retrieving data from mariadb
## Features
### Resources
Expose schema list in database
### Tools
- query_database
- Execute read-only operations against MariDB
## dependency
### install mariadb
- mac
- when install mariadb,
maybe raise os error below.
you can resolve by installing mariadb-connector-c.
```bash
OSError: mariadb_config not found.
This error typically indicates that MariaDB Connector/C, a dependency which
must be preinstalled, is not found.
If MariaDB Connector/C is not installed, see installation instructions
If MariaDB Connector/C is installed, either set the environment variable
MARIADB_CONFIG or edit the configuration file 'site.cfg' to set the
'mariadb_config' option to the file location of the mariadb_config utility.
```
1. execute `brew install mariadb-connector-c`
2. execute `echo 'export PATH="/opt/homebrew/opt/mariadb-connector-c/bin:$PATH"' >> ~/.bashrc`
3. set environment variable `export MARIADB_CONFIG=$(brew --prefix mariadb-connector-c)/bin/mariadb_config`
4. execute `uv add mariadb` again.
## Usage with Claude Desktop
### Configuration File
Paths to Claude Desktop config file:
- **MacOS**: `~/Library/Application Support/Claude/claude_desktop_config.json`
- **Windows**: `%APPDATA%\Claude\claude_desktop_config.json`
<!-- markdownlint-disable MD033 -->
<details>
<summary>Add this configuration to enable published servers</summary>
```json
{
"mcpServers": {
"mcp_server_mariadb": {
"command": "/PATH/TO/uvx"
"args": [
"mcp-server-mariadb",
"--host",
"${DB_HOST}",
"--port",
"${DB_PORT}",
"--user",
"${DB_USER}",
"--password",
"${DB_PASSWORD}",
"--database",
"${DB_NAME}"
]
}
}
}
```
**Note**: Replace these placeholders with actual paths:
- `/PATH/TO/uvx`: Full path to uvx executable
</details>
<details>
<summary>Add this configuration to enable development/unpublished servers</summary>
```json
{
"mcpServers": {
"mcp_server_mariadb": {
"command": "/PATH/TO/uv",
"args": [
"--directory",
"/YOUR/SOURCE/PATH/mcp-server-mariadb/src/mcp_server_mariadb",
"run",
"server.py"
],
"env": {
"MARIADB_HOST": "127.0.0.1",
"MARIADB_USER": "USER",
"MARIADB_PASSWORD": "PASSWORD",
"MARIADB_DATABASE": "DATABASE",
"MARIADB_PORT": "3306"
}
}
}
}
```
**Note**: Replace these placeholders with actual paths:
- `/PATH/TO/uv`: Full path to UV executable
- `/YOUR/SOURCE/PATH/mcp-server-mariadb/src/mcp_server_mariadb`: Path to server source code
</details>
## License
This mcp server is licensed under the MIT license. please see the LICENSE file in the repository.
```
--------------------------------------------------------------------------------
/tests/__init__.py:
--------------------------------------------------------------------------------
```python
```
--------------------------------------------------------------------------------
/src/mcp_server_mariadb/__init__.py:
--------------------------------------------------------------------------------
```python
from . import server
def main():
"""Entry point for the MCP server package."""
server.main()
__all__ = ["main"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mcp-server-mariadb"
version = "0.1.2"
description = "MCP Server for MariaDB"
readme = "README.md"
authors = [
{ name = "HeejunShin", email = "[email protected]" }
]
requires-python = ">=3.10"
dependencies = [
"mariadb>=1.1.12",
"mcp[cli]>=1.3.0",
"mysql-connector-python>=9.2.0",
"pytest>=8.3.4",
"python-dotenv>=1.0.1",
]
[project.scripts]
mcp-server-mariadb = "mcp_server_mariadb:main"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
```
--------------------------------------------------------------------------------
/tests/test_db.py:
--------------------------------------------------------------------------------
```python
from dotenv import load_dotenv
from src.mcp_server_mariadb.server import get_connection, is_read_only_query
load_dotenv()
def test_db_connection():
connection = get_connection()
assert connection is not None
def test_is_read_only_query():
assert is_read_only_query("SELECT * FROM users")
assert is_read_only_query("SHOW TABLES")
assert is_read_only_query("DESCRIBE users")
assert is_read_only_query("DESC users")
assert is_read_only_query("EXPLAIN SELECT * FROM users")
def test_is_not_read_only_query():
assert not is_read_only_query(
"INSERT INTO users (name, email) VALUES ('John', '[email protected]')"
)
assert not is_read_only_query(
"UPDATE users SET email = '[email protected]' WHERE id = 1"
)
assert not is_read_only_query("DELETE FROM users WHERE id = 1")
assert not is_read_only_query(
"CREATE TABLE users (id INT AUTO_INCREMENT PRIMARY KEY, name VARCHAR(255), email VARCHAR(255))"
)
assert not is_read_only_query("DROP TABLE users")
```
--------------------------------------------------------------------------------
/src/mcp_server_mariadb/server.py:
--------------------------------------------------------------------------------
```python
import argparse
import os
from contextlib import closing
from dataclasses import dataclass, field
import mariadb
from dotenv import load_dotenv
from mcp.server.fastmcp import FastMCP
load_dotenv()
mcp = FastMCP(
"MariaDB Explorer", dependencies=["mysql-connector-python", "python-dotenv"]
)
READ_ONLY_KEYWORDS = ("SELECT", "SHOW", "DESCRIBE", "DESC", "EXPLAIN")
READ_ONLY_KEYWORD_NAMES = ", ".join(READ_ONLY_KEYWORDS)
def get_arguments() -> dict:
"""Parse command-line arguments and return as a dictionary."""
parser = argparse.ArgumentParser(description="MariaDB Configuration")
parser.add_argument("--host", help="MariaDB host")
parser.add_argument("--port", type=int, help="MariaDB port")
parser.add_argument("--user", help="MariaDB user")
parser.add_argument("--password", help="MariaDB password")
parser.add_argument("--database", help="MariaDB database")
args = parser.parse_args()
return {k: v for k, v in vars(args).items() if v is not None}
@dataclass
class DBconfig:
"""Database configuration"""
host: str = field(default_factory=lambda: os.getenv("MARIADB_HOST", "localhost"))
port: int = field(default_factory=lambda: int(os.getenv("MARIADB_PORT", "3306")))
user: str = field(default_factory=lambda: os.getenv("MARIADB_USER", ""))
password: str = field(default_factory=lambda: os.getenv("MARIADB_PASSWORD", ""))
database: str = field(default_factory=lambda: os.getenv("MARIADB_DATABASE", ""))
@classmethod
def from_args(cls) -> "DBconfig":
"""Create a DBconfig instance from command-line arguments and environment variables."""
cli_args = get_arguments()
return cls(**{**cls().__dict__, **cli_args})
def get_connection():
"""Create a connection to the database connection"""
config = DBconfig.from_args()
try:
conn = mariadb.connect(
user=config.user,
password=config.password,
host=config.host,
port=config.port,
database=config.database,
)
return conn
except mariadb.Error as e:
print(f"Error connecting to MariaDB Platform: {e}")
def is_read_only_query(query: str) -> bool:
"""check if a query is read-only by examining its first word"""
first_word = query.strip().split()[0].upper()
return first_word in READ_ONLY_KEYWORDS
@mcp.resource("schema://tables")
def list_tables() -> str:
"""Get the schema for a specific table"""
try:
with closing(get_connection()) as conn:
cursor = conn.cursor()
cursor.execute("SHOW TABLES")
tables = cursor.fetchall()
return "\n".join(table[0] for table in tables)
except Exception as e:
return f"Error retrieving tables: {str(e)}"
@mcp.tool()
def query_database(query: str) -> str:
"""
Execute a read-only SQL query on the database
Args:
query: SQL query to execute (must be SELECT, SHOW, DESCRIBE, DESC, EXPLAIN)
"""
if not is_read_only_query(query):
return "Error: Only read-only queries (SELECT, SHOW, DESCRIBE, DESC, EXPLAIN) are allowed"
try:
with closing(get_connection()) as conn:
cursor = conn.cursor()
cursor.execute(query)
results = cursor.fetchall()
# Get column names
columns = [desc[0] for desc in cursor.description]
# Format results as a table
output = []
output.append(" | ".join(columns))
output.append(
"-" * (sum(len(col) for col in columns) + 3 * (len(columns) - 1))
)
for row in results:
output.append(" | ".join(str(val) for val in row))
return "\n".join(output)
except Exception as e:
return f"Error executing query {str(e)}"
def main():
mcp.run(transport="stdio")
if __name__ == "__main__":
main()
```