# Directory Structure
```
├── .env.example
├── .gitignore
├── .python-version
├── CHANGELOG.md
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── README.md
├── requirements.txt
├── smithery.yaml
├── src
│ └── mysqldb_mcp_server
│ ├── __init__.py
│ └── server.py
└── uv.lock
```
# Files
--------------------------------------------------------------------------------
/.python-version:
--------------------------------------------------------------------------------
```
3.11
```
--------------------------------------------------------------------------------
/.env.example:
--------------------------------------------------------------------------------
```
# MySQL Connection Settings
MYSQL_HOST=localhost
MYSQL_USER=root
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=optional_database_name
MYSQL_READONLY=true
```
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
# MacOS
.DS_Store
# Environment variables
.env
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
# mysqldb-mcp-server MCP server
[](https://smithery.ai/server/@burakdirin/mysqldb-mcp-server)
A MySQL database MCP server project.
## Installation
You can install the package using `uv`:
```bash
uv pip install mysqldb-mcp-server
```
Or using `pip`:
```bash
pip install mysqldb-mcp-server
```
## Components
### Tools
The server provides two tools:
- `connect_database`: Connects to a specific MySQL database
- `database` parameter: Name of the database to connect to (string)
- Returns a confirmation message when connection is successful
- `execute_query`: Executes MySQL queries
- `query` parameter: SQL query/queries to execute (string)
- Returns query results in JSON format
- Multiple queries can be sent separated by semicolons
## Configuration
The server uses the following environment variables:
- `MYSQL_HOST`: MySQL server address (default: "localhost")
- `MYSQL_USER`: MySQL username (default: "root")
- `MYSQL_PASSWORD`: MySQL password (default: "")
- `MYSQL_DATABASE`: Initial database (optional)
- `MYSQL_READONLY`: Read-only mode (set to 1/true to enable, default: false)
## Quickstart
### Installation
#### Claude Desktop
MacOS: `~/Library/Application\ Support/Claude/claude_desktop_config.json`
Windows: `%APPDATA%/Claude/claude_desktop_config.json`
<details>
<summary>Development/Unpublished Server Configuration</summary>
```json
{
"mcpServers": {
"mysqldb-mcp-server": {
"command": "uv",
"args": [
"--directory",
"/Users/burakdirin/Projects/mysqldb-mcp-server",
"run",
"mysqldb-mcp-server"
],
"env": {
"MYSQL_HOST": "localhost",
"MYSQL_USER": "root",
"MYSQL_PASSWORD": "password",
"MYSQL_DATABASE": "[optional]",
"MYSQL_READONLY": "true"
}
}
}
}
```
</details>
<details>
<summary>Published Server Configuration</summary>
```json
{
"mcpServers": {
"mysqldb-mcp-server": {
"command": "uvx",
"args": [
"mysqldb-mcp-server"
],
"env": {
"MYSQL_HOST": "localhost",
"MYSQL_USER": "root",
"MYSQL_PASSWORD": "password",
"MYSQL_DATABASE": "[optional]",
"MYSQL_READONLY": "true"
}
}
}
}
```
</details>
### Installing via Smithery
To install MySQL Database Integration Server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/@burakdirin/mysqldb-mcp-server):
```bash
npx -y @smithery/cli install @burakdirin/mysqldb-mcp-server --client claude
```
## Development
### Building and Publishing
To prepare the package for distribution:
1. Sync dependencies and update lockfile:
```bash
uv sync
```
2. Build package distributions:
```bash
uv build
```
This will create source and wheel distributions in the `dist/` directory.
3. Publish to PyPI:
```bash
uv publish
```
Note: You'll need to set PyPI credentials via environment variables or command flags:
- Token: `--token` or `UV_PUBLISH_TOKEN`
- Or username/password: `--username`/`UV_PUBLISH_USERNAME` and `--password`/`UV_PUBLISH_PASSWORD`
### Debugging
Since MCP servers run over stdio, debugging can be challenging. For the best debugging
experience, we strongly recommend using the [MCP Inspector](https://github.com/modelcontextprotocol/inspector).
You can launch the MCP Inspector via [`npm`](https://docs.npmjs.com/downloading-and-installing-node-js-and-npm) with this command:
```bash
npx @modelcontextprotocol/inspector uv --directory /Users/burakdirin/Projects/mysqldb-mcp-server run mysqldb-mcp-server
```
Upon launching, the Inspector will display a URL that you can access in your browser to begin debugging.
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp>=1.4.1
python-dotenv>=1.0.1
mysql-connector-python>=9.2.0
```
--------------------------------------------------------------------------------
/src/mysqldb_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
"""
MySQL MCP Server
A server for interacting with MySQL databases through MCP.
"""
from .server import mcp
def main() -> None:
"""Run the MySQL MCP server"""
mcp.run()
__all__ = ['mcp', 'main']
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
FROM python:3.11-slim
# Set working directory
WORKDIR /app
# Copy necessary project files
COPY pyproject.toml ./
COPY README.md ./
COPY src/ ./src/
# Upgrade pip and install build dependencies
RUN pip install --upgrade pip \
&& pip install hatchling
# Install the package (skip prepare scripts if any)
RUN pip install --ignore-installed --no-cache-dir .
# MCP server runs over stdio so no ports are exposed
# Start the MCP server using the entrypoint installed by setup
CMD ["mysqldb-mcp-server"]
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mysqldb-mcp-server"
version = "0.1.2"
description = "An MCP server implementation for MySQL database integration"
readme = "README.md"
requires-python = ">=3.11"
license = "MIT"
dependencies = [
"mcp>=1.4.1",
"mysql-connector-python>=9.2.0",
"python-dotenv>=1.0.1",
]
[project.urls]
Homepage = "https://github.com/burakdirin/mysqldb-mcp-server"
Repository = "https://github.com/burakdirin/mysqldb-mcp-server"
"Bug Tracker" = "https://github.com/burakdirin/mysqldb-mcp-server/issues"
[[project.authors]]
name = "Burak Dirin"
email = "[email protected]"
[build-system]
requires = [ "hatchling",]
build-backend = "hatchling.build"
[project.scripts]
mysqldb-mcp-server = "mysqldb_mcp_server:main"
[tool.hatch.build.targets.wheel]
packages = ["src/mysqldb_mcp_server"]
```
--------------------------------------------------------------------------------
/CHANGELOG.md:
--------------------------------------------------------------------------------
```markdown
# Changelog
All notable changes to this project will be documented in this file.
The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
## [0.1.2] - 2025-03-17
### Changed
- Implemented lazy database initialization to improve deployment reliability
- Modified startup behavior to prevent immediate database connection attempts
## [0.1.1] - 2025-03-17
### Added
- GitHub repository information to package metadata
## [0.1.0] - 2025-03-17
### Added
- Initial MCP server implementation for MySQL database integration
- Python 3.11+ support
- MySQL Connector integration
[0.1.2]: https://github.com/burakdirin/mysqldb-mcp-server/releases/tag/v0.1.2
[0.1.1]: https://github.com/burakdirin/mysqldb-mcp-server/releases/tag/v0.1.1
[0.1.0]: https://github.com/burakdirin/mysqldb-mcp-server/releases/tag/v0.1.0
```
--------------------------------------------------------------------------------
/smithery.yaml:
--------------------------------------------------------------------------------
```yaml
# Smithery configuration file: https://smithery.ai/docs/config#smitheryyaml
startCommand:
type: stdio
configSchema:
# JSON Schema defining the configuration options for the MCP.
type: object
properties:
MYSQL_HOST:
type: string
default: localhost
description: MySQL server address
MYSQL_USER:
type: string
default: root
description: MySQL username
MYSQL_PASSWORD:
type: string
default: ""
description: MySQL password
MYSQL_DATABASE:
type: string
default: ""
description: Initial database (optional)
MYSQL_READONLY:
type: boolean
default: false
description: Read-only mode, set true to enable
commandFunction:
# A JS function that produces the CLI command based on the given config to start the MCP on stdio.
|-
(config) => ({
command: 'mysqldb-mcp-server',
args: [],
env: {
MYSQL_HOST: config.MYSQL_HOST,
MYSQL_USER: config.MYSQL_USER,
MYSQL_PASSWORD: config.MYSQL_PASSWORD,
MYSQL_DATABASE: config.MYSQL_DATABASE,
MYSQL_READONLY: config.MYSQL_READONLY.toString()
}
})
exampleConfig:
MYSQL_HOST: localhost
MYSQL_USER: root
MYSQL_PASSWORD: password
MYSQL_DATABASE: ""
MYSQL_READONLY: false
```
--------------------------------------------------------------------------------
/src/mysqldb_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
import json
import os
from contextlib import asynccontextmanager
from dataclasses import dataclass
from enum import Enum
from typing import Any, AsyncIterator, Dict, List, Optional, Union
import mysql.connector
from dotenv import load_dotenv
from mcp.server.fastmcp import Context, FastMCP
from mysql.connector import Error as MySQLError
from mysql.connector.cursor import MySQLCursor
class MySQLServerError(Exception):
"""Base exception for MySQL server errors"""
pass
class ConnectionError(MySQLServerError):
"""Raised when there's an issue with the database connection"""
pass
class QueryError(MySQLServerError):
"""Raised when there's an issue executing a query"""
pass
class QueryType(Enum):
"""Enum for different types of SQL queries"""
SELECT = "SELECT"
INSERT = "INSERT"
UPDATE = "UPDATE"
DELETE = "DELETE"
CREATE = "CREATE"
DROP = "DROP"
ALTER = "ALTER"
TRUNCATE = "TRUNCATE"
USE = "USE"
SHOW = "SHOW"
DESCRIBE = "DESCRIBE"
@classmethod
def is_write_operation(cls, query_type: str) -> bool:
"""Check if the query type is a write operation"""
write_operations = {
cls.INSERT, cls.UPDATE, cls.DELETE,
cls.CREATE, cls.DROP, cls.ALTER, cls.TRUNCATE
}
try:
return cls(query_type.upper()) in write_operations
except ValueError:
return False
@dataclass
class MySQLContext:
"""Context for MySQL connection"""
host: str
user: str
password: str
database: Optional[str]
readonly: bool
connection: Optional[mysql.connector.MySQLConnection] = None
def ensure_connected(self) -> None:
"""Ensure database connection is available, connecting lazily if needed"""
if not self.connection or not self.connection.is_connected():
config = {
"host": self.host,
"user": self.user,
"password": self.password,
}
if self.database:
config["database"] = self.database
try:
self.connection = mysql.connector.connect(**config)
except MySQLError as e:
raise ConnectionError(
f"Failed to connect to database: {str(e)}")
class QueryExecutor:
"""Handles MySQL query execution and result processing"""
def __init__(self, context: MySQLContext):
self.context = context
def _format_datetime(self, value: Any) -> Any:
"""Format datetime values to string"""
return value.strftime('%Y-%m-%d %H:%M:%S') if hasattr(value, 'strftime') else value
def _process_row(self, row: Dict[str, Any]) -> Dict[str, Any]:
"""Process a single row of results"""
return {key: self._format_datetime(value) for key, value in row.items()}
def _process_results(self, cursor: MySQLCursor) -> Union[List[Dict[str, Any]], Dict[str, int]]:
"""Process query results"""
if cursor.with_rows:
results = cursor.fetchall()
return [self._process_row(row) for row in results]
return {"affected_rows": cursor.rowcount}
def execute_single_query(self, query: str) -> Dict[str, Any]:
"""Execute a single query and return results"""
self.context.ensure_connected()
cursor = None
try:
cursor = self.context.connection.cursor(dictionary=True)
query_type = QueryType(query.strip().upper().split()[0])
# Handle readonly mode
if self.context.readonly and QueryType.is_write_operation(query_type.value):
raise QueryError(
"Server is in read-only mode. Write operations are not allowed.")
# Handle USE statements
if query_type == QueryType.USE:
db_name = query.strip().split()[-1].strip('`').strip()
self.context.database = db_name
cursor.execute(query)
return {"message": f"Switched to database: {db_name}"}
# Execute query
cursor.execute(query)
results = self._process_results(cursor)
if not self.context.readonly:
self.context.connection.commit()
return results
except MySQLError as e:
raise QueryError(f"Error executing query: {str(e)}")
finally:
if cursor:
cursor.close()
def execute_multiple_queries(self, query: str) -> List[Dict[str, Any]]:
"""Execute multiple queries and return results"""
queries = [q.strip() for q in query.split(';') if q.strip()]
results = []
for single_query in queries:
try:
result = self.execute_single_query(single_query)
results.append(result)
except QueryError as e:
results.append({"error": str(e)})
return results
def get_env_vars() -> tuple[str, str, str, Optional[str], bool]:
"""Get MySQL connection settings from environment variables
Returns:
Tuple of (host, user, password, database, readonly)
"""
load_dotenv()
host = os.getenv("MYSQL_HOST", "localhost")
user = os.getenv("MYSQL_USER", "root")
password = os.getenv("MYSQL_PASSWORD", "")
database = os.getenv("MYSQL_DATABASE") # Optional
readonly = os.getenv("MYSQL_READONLY", "0") in ("1", "true", "True")
return host, user, password, database, readonly
@asynccontextmanager
async def mysql_lifespan(server: FastMCP) -> AsyncIterator[MySQLContext]:
"""MySQL connection lifecycle manager"""
# Get connection settings from environment variables
host, user, password, database, readonly = get_env_vars()
# Initialize context without connecting
ctx = MySQLContext(
host=host,
user=user,
password=password,
database=database,
readonly=readonly,
connection=None # Don't connect immediately
)
try:
yield ctx
finally:
if ctx.connection and ctx.connection.is_connected():
ctx.connection.close()
# Create MCP server instance
mcp = FastMCP("MySQL Explorer", lifespan=mysql_lifespan)
def _get_executor(ctx: Context) -> QueryExecutor:
"""Helper function to get QueryExecutor from context"""
mysql_ctx = ctx.request_context.lifespan_context
return QueryExecutor(mysql_ctx)
@mcp.tool()
def connect_database(database: str, ctx: Context) -> str:
"""Connect to a specific MySQL database"""
try:
executor = _get_executor(ctx)
result = executor.execute_single_query(f"USE `{database}`")
return json.dumps(result, indent=2)
except (ConnectionError, QueryError) as e:
return str(e)
@mcp.tool()
def execute_query(query: str, ctx: Context) -> str:
"""Execute MySQL queries"""
try:
executor = _get_executor(ctx)
results = executor.execute_multiple_queries(query)
if len(results) == 1:
return json.dumps(results[0], indent=2)
return json.dumps(results, indent=2)
except (ConnectionError, QueryError) as e:
return str(e)
if __name__ == "__main__":
mcp.run()
```