# Directory Structure
```
├── .github
│ └── workflows
│ ├── publish.yml
│ └── test.yml
├── .gitignore
├── CHANGELOG.md
├── Dockerfile
├── LICENSE
├── pyproject.toml
├── pytest.ini
├── README.md
├── requirements-dev.txt
├── requirements.txt
├── SECURITY.md
├── smithery.yaml
├── src
│ └── mysql_mcp_server
│ ├── __init__.py
│ └── server.py
└── tests
├── conftest.py
└── test_server.py
```
# Files
--------------------------------------------------------------------------------
/.gitignore:
--------------------------------------------------------------------------------
```
# Python-generated files
__pycache__/
*.py[oc]
build/
dist/
wheels/
*.egg-info
# Virtual environments
.venv
Lib/
Scripts/
# Development environment
.python-version
uv.lock
# IDE settings (optional)
.vscode/
.idea/
# Distribution directories
*.dist-info/
pypi_publishing_notes.md
```
--------------------------------------------------------------------------------
/README.md:
--------------------------------------------------------------------------------
```markdown
[](https://smithery.ai/server/mysql-mcp-server)
[](https://mseep.ai/app/designcomputer-mysql-mcp-server)
# MySQL MCP Server
A Model Context Protocol (MCP) implementation that enables secure interaction with MySQL databases. This server component facilitates communication between AI applications (hosts/clients) and MySQL databases, making database exploration and analysis safer and more structured through a controlled interface.
> **Note**: MySQL MCP Server is not designed to be used as a standalone server, but rather as a communication protocol implementation between AI applications and MySQL databases.
## Features
- List available MySQL tables as resources
- Read table contents
- Execute SQL queries with proper error handling
- Secure database access through environment variables
- Comprehensive logging
## Installation
### Manual Installation
```bash
pip install mysql-mcp-server
```
### Installing via Smithery
To install MySQL MCP Server for Claude Desktop automatically via [Smithery](https://smithery.ai/server/mysql-mcp-server):
```bash
npx -y @smithery/cli install mysql-mcp-server --client claude
```
## Configuration
Set the following environment variables:
```bash
MYSQL_HOST=localhost # Database host
MYSQL_PORT=3306 # Optional: Database port (defaults to 3306 if not specified)
MYSQL_USER=your_username
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=your_database
```
## Usage
### With Claude Desktop
Add this to your `claude_desktop_config.json`:
```json
{
"mcpServers": {
"mysql": {
"command": "uv",
"args": [
"--directory",
"path/to/mysql_mcp_server",
"run",
"mysql_mcp_server"
],
"env": {
"MYSQL_HOST": "localhost",
"MYSQL_PORT": "3306",
"MYSQL_USER": "your_username",
"MYSQL_PASSWORD": "your_password",
"MYSQL_DATABASE": "your_database"
}
}
}
}
```
### With Visual Studio Code
Add this to your `mcp.json`:
```json
{
"servers": {
"mysql": {
"type": "stdio",
"command": "uvx",
"args": [
"--from",
"mysql-mcp-server",
"mysql_mcp_server"
],
"env": {
"MYSQL_HOST": "localhost",
"MYSQL_PORT": "3306",
"MYSQL_USER": "your_username",
"MYSQL_PASSWORD": "your_password",
"MYSQL_DATABASE": "your_database"
}
}
}
}
```
Note: Will need to install uv for this to work
### Debugging with MCP Inspector
While MySQL MCP Server isn't intended to be run standalone or directly from the command line with Python, you can use the MCP Inspector to debug it.
The MCP Inspector provides a convenient way to test and debug your MCP implementation:
```bash
# Install dependencies
pip install -r requirements.txt
# Use the MCP Inspector for debugging (do not run directly with Python)
```
The MySQL MCP Server is designed to be integrated with AI applications like Claude Desktop and should not be run directly as a standalone Python program.
## Development
```bash
# Clone the repository
git clone https://github.com/designcomputer/mysql_mcp_server.git
cd mysql_mcp_server
# Create virtual environment
python -m venv venv
source venv/bin/activate # or `venv\Scripts\activate` on Windows
# Install development dependencies
pip install -r requirements-dev.txt
# Run tests
pytest
```
## Security Considerations
- Never commit environment variables or credentials
- Use a database user with minimal required permissions
- Consider implementing query whitelisting for production use
- Monitor and log all database operations
## Security Best Practices
This MCP implementation requires database access to function. For security:
1. **Create a dedicated MySQL user** with minimal permissions
2. **Never use root credentials** or administrative accounts
3. **Restrict database access** to only necessary operations
4. **Enable logging** for audit purposes
5. **Regular security reviews** of database access
See [MySQL Security Configuration Guide](https://github.com/designcomputer/mysql_mcp_server/blob/main/SECURITY.md) for detailed instructions on:
- Creating a restricted MySQL user
- Setting appropriate permissions
- Monitoring database access
- Security best practices
⚠️ IMPORTANT: Always follow the principle of least privilege when configuring database access.
## License
MIT License - see LICENSE file for details.
## Contributing
1. Fork the repository
2. Create your feature branch (`git checkout -b feature/amazing-feature`)
3. Commit your changes (`git commit -m 'Add some amazing feature'`)
4. Push to the branch (`git push origin feature/amazing-feature`)
5. Open a Pull Request
```
--------------------------------------------------------------------------------
/SECURITY.md:
--------------------------------------------------------------------------------
```markdown
## MySQL Security Configuration
### Creating a Restricted MySQL User
It's crucial to create a dedicated MySQL user with minimal permissions for the MCP server. Never use the root account or a user with full administrative privileges.
#### 1. Create a new MySQL user
```sql
-- Connect as root or administrator
CREATE USER 'mcp_user'@'localhost' IDENTIFIED BY 'your_secure_password';
```
#### 2. Grant minimal required permissions
Basic read-only access (recommended for exploration and analysis):
```sql
-- Grant SELECT permission only
GRANT SELECT ON your_database.* TO 'mcp_user'@'localhost';
```
Standard access (allows data modification but not structural changes):
```sql
-- Grant data manipulation permissions
GRANT SELECT, INSERT, UPDATE, DELETE ON your_database.* TO 'mcp_user'@'localhost';
```
Advanced access (includes ability to create temporary tables for complex queries):
```sql
-- Grant additional permissions for advanced operations
GRANT SELECT, INSERT, UPDATE, DELETE, CREATE TEMPORARY TABLES
ON your_database.* TO 'mcp_user'@'localhost';
```
#### 3. Apply the permissions
```sql
FLUSH PRIVILEGES;
```
### Additional Security Measures
1. **Network Access**
- Restrict the user to connecting only from localhost if the MCP server runs on the same machine
- If remote access is needed, specify exact IP addresses rather than using wildcards
2. **Query Restrictions**
- Consider using VIEWs to further restrict data access
- Set appropriate `max_queries_per_hour`, `max_updates_per_hour` limits:
```sql
ALTER USER 'mcp_user'@'localhost'
WITH MAX_QUERIES_PER_HOUR 1000
MAX_UPDATES_PER_HOUR 100;
```
3. **Data Access Control**
- Grant access only to specific tables when possible
- Use column-level permissions for sensitive data:
```sql
GRANT SELECT (public_column1, public_column2)
ON your_database.sensitive_table TO 'mcp_user'@'localhost';
```
4. **Regular Auditing**
- Enable MySQL audit logging for the MCP user
- Regularly review logs for unusual patterns
- Periodically review and adjust permissions
### Environment Configuration
When setting up the MCP server, use these restricted credentials in your environment:
```bash
MYSQL_USER=mcp_user
MYSQL_PASSWORD=your_secure_password
MYSQL_DATABASE=your_database
MYSQL_HOST=localhost
```
### Monitoring Usage
To monitor the MCP user's database usage:
```sql
-- Check current connections
SELECT * FROM information_schema.PROCESSLIST
WHERE user = 'mcp_user';
-- View user privileges
SHOW GRANTS FOR 'mcp_user'@'localhost';
-- Check resource limits
SELECT * FROM mysql.user
WHERE user = 'mcp_user' AND host = 'localhost';
```
### Best Practices
1. **Regular Password Rotation**
- Change the MCP user's password periodically
- Use strong, randomly generated passwords
- Update application configurations after password changes
2. **Permission Review**
- Regularly audit granted permissions
- Remove unnecessary privileges
- Keep permissions as restrictive as possible
3. **Access Patterns**
- Monitor query patterns for potential issues
- Set up alerts for unusual activity
- Maintain detailed logs of database access
4. **Data Protection**
- Consider encrypting sensitive columns
- Use SSL/TLS for database connections
- Implement data masking where appropriate
```
--------------------------------------------------------------------------------
/requirements.txt:
--------------------------------------------------------------------------------
```
mcp>=1.0.0
mysql-connector-python>=9.1.0
```
--------------------------------------------------------------------------------
/requirements-dev.txt:
--------------------------------------------------------------------------------
```
pytest>=7.0.0
pytest-asyncio>=0.23.0
pytest-cov>=4.1.0
black>=23.0.0
isort>=5.12.0
mypy>=1.0.0
```
--------------------------------------------------------------------------------
/pytest.ini:
--------------------------------------------------------------------------------
```
[pytest]
asyncio_mode = auto
asyncio_default_fixture_loop_scope = function
testpaths = tests
python_files = test_*.py
```
--------------------------------------------------------------------------------
/src/mysql_mcp_server/__init__.py:
--------------------------------------------------------------------------------
```python
from . import server
import asyncio
def main():
"""Main entry point for the package."""
asyncio.run(server.main())
# Expose important items at package level
__all__ = ['main', 'server']
```
--------------------------------------------------------------------------------
/.github/workflows/publish.yml:
--------------------------------------------------------------------------------
```yaml
name: Publish to PyPI
on:
release:
types: [published]
jobs:
deploy:
runs-on: ubuntu-latest
steps:
- uses: actions/checkout@v4
- name: Set up Python
uses: actions/setup-python@v4
with:
python-version: '3.12'
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install build
- name: Build package
run: python -m build
- name: Publish to PyPI
uses: pypa/gh-action-pypi-publish@release/v1
with:
password: ${{ secrets.PYPI_API_TOKEN }}
# Remove attestations line if it exists
```
--------------------------------------------------------------------------------
/pyproject.toml:
--------------------------------------------------------------------------------
```toml
[project]
name = "mysql_mcp_server"
version = "0.2.2"
description = "A Model Context Protocol (MCP) server that enables secure interaction with MySQL databases. This server allows AI assistants to list tables, read data, and execute SQL queries through a controlled interface, making database exploration and analysis safer and more structured."
readme = "README.md"
requires-python = ">=3.11"
dependencies = [
"mcp>=1.0.0",
"mysql-connector-python>=9.1.0",
]
[[project.authors]]
name = "Dana K. Williams"
email = "[email protected]"
[build-system]
requires = ["hatchling"]
build-backend = "hatchling.build"
[project.scripts]
mysql_mcp_server = "mysql_mcp_server:main"
```
--------------------------------------------------------------------------------
/Dockerfile:
--------------------------------------------------------------------------------
```dockerfile
# Generated by https://smithery.ai. See: https://smithery.ai/docs/config#dockerfile
# Use the official Python image from the Docker Hub
FROM python:3.11-slim
# Set the working directory in the container
WORKDIR /app
# Copy the requirements file into the container
COPY requirements.txt .
# Install the dependencies specified in the requirements file
RUN pip install --no-cache-dir -r requirements.txt
# Copy the current directory contents into the container at /app
COPY src/ /app/src
# Set environment variables for MySQL (these can be overwritten with `docker run -e`)
ENV MYSQL_HOST=host.docker.internal
ENV MYSQL_PORT=3306
ENV MYSQL_USER=your_username
ENV MYSQL_PASSWORD=your_password
ENV MYSQL_DATABASE=your_database
ENV PYTHONPATH=/app/src
# Command to run the server
CMD ["python", "-m", "mysql_mcp_server.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.2.2] - 2025-04-18
### Fixed
- Fixed handling of SQL commands that return result sets, including `SHOW INDEX`, `SHOW CREATE TABLE`, and `DESCRIBE`
- Added improved error handling for result fetching operations
- Added additional debug output to aid in troubleshooting
## [0.2.1] - 2025-02-15
### Added
- Support for MYSQL_PORT configuration through environment variables
- Documentation for PORT configuration in README
### Changed
- Updated tests to use handler functions directly
- Refactored database configuration to runtime
## [0.2.0] - 2025-01-20
### Added
- Initial release with MCP server implementation
- Support for SQL queries through MCP interface
- Ability to list tables and read data
```
--------------------------------------------------------------------------------
/tests/conftest.py:
--------------------------------------------------------------------------------
```python
# tests/conftest.py
import pytest
import os
import mysql.connector
from mysql.connector import Error
@pytest.fixture(scope="session")
def mysql_connection():
"""Create a test database connection."""
try:
connection = mysql.connector.connect(
host=os.getenv("MYSQL_HOST", "127.0.0.1"),
user=os.getenv("MYSQL_USER", "root"),
password=os.getenv("MYSQL_PASSWORD", "testpassword"),
database=os.getenv("MYSQL_DATABASE", "test_db")
)
if connection.is_connected():
# Create a test table
cursor = connection.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS test_table (
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255),
value INT
)
""")
connection.commit()
yield connection
# Cleanup
cursor.execute("DROP TABLE IF EXISTS test_table")
connection.commit()
cursor.close()
connection.close()
except Error as e:
pytest.fail(f"Failed to connect to MySQL: {e}")
@pytest.fixture(scope="session")
def mysql_cursor(mysql_connection):
"""Create a test cursor."""
cursor = mysql_connection.cursor()
yield cursor
cursor.close()
```
--------------------------------------------------------------------------------
/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
required:
- mysqlUser
- mysqlPassword
- mysqlDatabase
properties:
mysqlHost:
type: string
description: "The hostname of the MySQL server. Use localhost for local connections or a specific address for remote databases. For Docker, host.docker.internal allows accessing the host machine."
default: "host.docker.internal"
mysqlPort:
type: number
description: "The port of the MySQL server (default: 3306)."
default: 3306
mysqlUser:
type: string
description: "The username for MySQL authentication."
mysqlPassword:
type: string
description: "The password for MySQL authentication."
mysqlDatabase:
type: string
description: "The database to connect to."
commandFunction:
# A function that produces the CLI command to start the MCP on stdio.
|-
(config) => ({
command: 'docker',
args: [
'run',
'-i',
'--rm',
'-e', `MYSQL_HOST=${config.mysqlHost}`,
'-e', `MYSQL_PORT=${config.mysqlPort}`,
'-e', `MYSQL_USER=${config.mysqlUser}`,
'-e', `MYSQL_PASSWORD=${config.mysqlPassword}`,
'-e', `MYSQL_DATABASE=${config.mysqlDatabase}`,
'smithery/mysql-mcp-server:latest'
]
})
```
--------------------------------------------------------------------------------
/.github/workflows/test.yml:
--------------------------------------------------------------------------------
```yaml
# .github/workflows/test.yml
name: Tests
on:
push:
branches: [ main ]
pull_request:
branches: [ main ]
jobs:
test:
runs-on: ubuntu-latest
services:
mysql:
image: mysql:8.0
env:
MYSQL_ROOT_PASSWORD: testpassword
MYSQL_DATABASE: test_db
ports:
- 3306:3306
options: --health-cmd="mysqladmin ping" --health-interval=10s --health-timeout=5s --health-retries=3
strategy:
matrix:
python-version: ["3.11", "3.12"]
steps:
- uses: actions/checkout@v4
- name: Set up Python ${{ matrix.python-version }}
uses: actions/setup-python@v4
with:
python-version: ${{ matrix.python-version }}
- name: Install dependencies
run: |
python -m pip install --upgrade pip
pip install -r requirements.txt
pip install -r requirements-dev.txt
pip install -e .
- name: Verify installation
run: |
# Show Python path
python -c "import sys; print('\n'.join(sys.path))"
# Try importing the module
python -c "import mysql_mcp_server; print('Module found at:', mysql_mcp_server.__file__)"
# List installed packages
pip list | grep mysql
# Show the content of site-packages
ls -R $(python -c "import site; print(site.getsitepackages()[0])")
- name: Run tests
env:
MYSQL_HOST: 127.0.0.1
MYSQL_USER: root
MYSQL_PASSWORD: testpassword
MYSQL_DATABASE: test_db
run: |
python -m pytest -v
```
--------------------------------------------------------------------------------
/tests/test_server.py:
--------------------------------------------------------------------------------
```python
import pytest
from mysql_mcp_server.server import app, list_tools, list_resources, read_resource, call_tool
from pydantic import AnyUrl
def test_server_initialization():
"""Test that the server initializes correctly."""
assert app.name == "mysql_mcp_server"
@pytest.mark.asyncio
async def test_list_tools():
"""Test that list_tools returns expected tools."""
tools = await list_tools()
assert len(tools) == 1
assert tools[0].name == "execute_sql"
assert "query" in tools[0].inputSchema["properties"]
@pytest.mark.asyncio
async def test_call_tool_invalid_name():
"""Test calling a tool with an invalid name."""
with pytest.raises(ValueError, match="Unknown tool"):
await call_tool("invalid_tool", {})
@pytest.mark.asyncio
async def test_call_tool_missing_query():
"""Test calling execute_sql without a query."""
with pytest.raises(ValueError, match="Query is required"):
await call_tool("execute_sql", {})
# Skip database-dependent tests if no database connection
@pytest.mark.asyncio
@pytest.mark.skipif(
not all([
pytest.importorskip("mysql.connector"),
pytest.importorskip("mysql_mcp_server")
]),
reason="MySQL connection not available"
)
async def test_list_resources():
"""Test listing resources (requires database connection)."""
try:
resources = await list_resources()
assert isinstance(resources, list)
except ValueError as e:
if "Missing required database configuration" in str(e):
pytest.skip("Database configuration not available")
raise
```
--------------------------------------------------------------------------------
/src/mysql_mcp_server/server.py:
--------------------------------------------------------------------------------
```python
import asyncio
import logging
import os
import sys
from mysql.connector import connect, Error
from mcp.server import Server
from mcp.types import Resource, Tool, TextContent
from pydantic import AnyUrl
# Configure logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger("mysql_mcp_server")
def get_db_config():
"""Get database configuration from environment variables."""
config = {
"host": os.getenv("MYSQL_HOST", "localhost"),
"port": int(os.getenv("MYSQL_PORT", "3306")),
"user": os.getenv("MYSQL_USER"),
"password": os.getenv("MYSQL_PASSWORD"),
"database": os.getenv("MYSQL_DATABASE"),
# Add charset and collation to avoid utf8mb4_0900_ai_ci issues with older MySQL versions
# These can be overridden via environment variables for specific MySQL versions
"charset": os.getenv("MYSQL_CHARSET", "utf8mb4"),
"collation": os.getenv("MYSQL_COLLATION", "utf8mb4_unicode_ci"),
# Disable autocommit for better transaction control
"autocommit": True,
# Set SQL mode for better compatibility - can be overridden
"sql_mode": os.getenv("MYSQL_SQL_MODE", "TRADITIONAL")
}
# Remove None values to let MySQL connector use defaults if not specified
config = {k: v for k, v in config.items() if v is not None}
if not all([config.get("user"), config.get("password"), config.get("database")]):
logger.error("Missing required database configuration. Please check environment variables:")
logger.error("MYSQL_USER, MYSQL_PASSWORD, and MYSQL_DATABASE are required")
raise ValueError("Missing required database configuration")
return config
# Initialize server
app = Server("mysql_mcp_server")
@app.list_resources()
async def list_resources() -> list[Resource]:
"""List MySQL tables as resources."""
config = get_db_config()
try:
logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
with connect(**config) as conn:
logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
with conn.cursor() as cursor:
cursor.execute("SHOW TABLES")
tables = cursor.fetchall()
logger.info(f"Found tables: {tables}")
resources = []
for table in tables:
resources.append(
Resource(
uri=f"mysql://{table[0]}/data",
name=f"Table: {table[0]}",
mimeType="text/plain",
description=f"Data in table: {table[0]}"
)
)
return resources
except Error as e:
logger.error(f"Failed to list resources: {str(e)}")
logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
return []
@app.read_resource()
async def read_resource(uri: AnyUrl) -> str:
"""Read table contents."""
config = get_db_config()
uri_str = str(uri)
logger.info(f"Reading resource: {uri_str}")
if not uri_str.startswith("mysql://"):
raise ValueError(f"Invalid URI scheme: {uri_str}")
parts = uri_str[8:].split('/')
table = parts[0]
try:
logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
with connect(**config) as conn:
logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
with conn.cursor() as cursor:
cursor.execute(f"SELECT * FROM {table} LIMIT 100")
columns = [desc[0] for desc in cursor.description]
rows = cursor.fetchall()
result = [",".join(map(str, row)) for row in rows]
return "\n".join([",".join(columns)] + result)
except Error as e:
logger.error(f"Database error reading resource {uri}: {str(e)}")
logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
raise RuntimeError(f"Database error: {str(e)}")
@app.list_tools()
async def list_tools() -> list[Tool]:
"""List available MySQL tools."""
logger.info("Listing tools...")
return [
Tool(
name="execute_sql",
description="Execute an SQL query on the MySQL server",
inputSchema={
"type": "object",
"properties": {
"query": {
"type": "string",
"description": "The SQL query to execute"
}
},
"required": ["query"]
}
)
]
@app.call_tool()
async def call_tool(name: str, arguments: dict) -> list[TextContent]:
"""Execute SQL commands."""
config = get_db_config()
logger.info(f"Calling tool: {name} with arguments: {arguments}")
if name != "execute_sql":
raise ValueError(f"Unknown tool: {name}")
query = arguments.get("query")
if not query:
raise ValueError("Query is required")
try:
logger.info(f"Connecting to MySQL with charset: {config.get('charset')}, collation: {config.get('collation')}")
with connect(**config) as conn:
logger.info(f"Successfully connected to MySQL server version: {conn.get_server_info()}")
with conn.cursor() as cursor:
cursor.execute(query)
# Special handling for SHOW TABLES
if query.strip().upper().startswith("SHOW TABLES"):
tables = cursor.fetchall()
result = ["Tables_in_" + config["database"]] # Header
result.extend([table[0] for table in tables])
return [TextContent(type="text", text="\n".join(result))]
# Handle all other queries that return result sets (SELECT, SHOW, DESCRIBE etc.)
elif cursor.description is not None:
columns = [desc[0] for desc in cursor.description]
try:
rows = cursor.fetchall()
result = [",".join(map(str, row)) for row in rows]
return [TextContent(type="text", text="\n".join([",".join(columns)] + result))]
except Error as e:
logger.warning(f"Error fetching results: {str(e)}")
return [TextContent(type="text", text=f"Query executed but error fetching results: {str(e)}")]
# Non-SELECT queries
else:
conn.commit()
return [TextContent(type="text", text=f"Query executed successfully. Rows affected: {cursor.rowcount}")]
except Error as e:
logger.error(f"Error executing SQL '{query}': {e}")
logger.error(f"Error code: {e.errno}, SQL state: {e.sqlstate}")
return [TextContent(type="text", text=f"Error executing query: {str(e)}")]
async def main():
"""Main entry point to run the MCP server."""
from mcp.server.stdio import stdio_server
# Add additional debug output
print("Starting MySQL MCP server with config:", file=sys.stderr)
config = get_db_config()
print(f"Host: {config['host']}", file=sys.stderr)
print(f"Port: {config['port']}", file=sys.stderr)
print(f"User: {config['user']}", file=sys.stderr)
print(f"Database: {config['database']}", file=sys.stderr)
logger.info("Starting MySQL MCP server...")
logger.info(f"Database config: {config['host']}/{config['database']} as {config['user']}")
async with stdio_server() as (read_stream, write_stream):
try:
await app.run(
read_stream,
write_stream,
app.create_initialization_options()
)
except Exception as e:
logger.error(f"Server error: {str(e)}", exc_info=True)
raise
if __name__ == "__main__":
asyncio.run(main())
```