MySQL-MCP
MCP(Model Context Protocol)로 구축된 Claude용 강력한 MySQL 데이터베이스 커넥터
MySQL-MCP는 Claude가 Model Context Protocol을 통해 MySQL 데이터베이스에 직접 접근할 수 있도록 지원합니다. 간단한 자연어 명령어를 통해 데이터베이스를 쿼리, 수정 및 탐색할 수 있는 완전한 도구 세트를 제공합니다.
특징
- 쿼리 실행 : 데이터베이스에 대해 모든 SQL 쿼리를 실행합니다.
- 쿼리 설명 : 자세한 쿼리 실행 계획을 얻으세요
- 스키마 탐색 : 데이터베이스, 테이블 및 열 정의 보기
- 데이터 검사 : 샘플 행을 포함한 테이블 내용 미리 보기
- 안전 제일 : 실수로 인한 데이터 손실을 방지하기 위한 제어
- 모범 사례 : Claude가 최적의 SQL을 작성하는 데 도움이 되는 내장 프롬프트
- 전체 로깅 : 모든 작업에 대한 포괄적인 로깅
설치
MySQL-MCP를 설치하는 방법에는 여러 가지가 있습니다.
옵션 1: pip를 사용하여 소스에서 설치
지엑스피1
옵션 2: 개발 패키지로 설치
git clone https://github.com/yourusername/mysql-mcp.git
cd mysql-mcp
pip install -e .
옵션 3: FastMCP로 설치(Claude 통합용)
먼저 FastMCP를 설치하세요.
그런 다음 MySQL-MCP 패키지를 설치하세요.
# From a local copy
fastmcp install mysql_mcp.py
# Or directly from GitHub
fastmcp install https://github.com/yourusername/mysql-mcp/mysql_mcp.py
빠른 시작
# mysql_mcp.py
from fastmcp import FastMCP
import mysql.connector
from mysql.connector import Error
import os
from dotenv import load_dotenv
from pydantic import BaseModel, Field
from typing import List, Optional, Dict, Any
# Load environment variables
load_dotenv()
# Initialize FastMCP app
mcp = FastMCP(
"MySQL MCP",
description="MySQL database connector for Claude",
dependencies=["mysql-connector-python", "python-dotenv"]
)
# Database connection configuration
class DBConfig(BaseModel):
host: str = Field(default=os.getenv("MYSQL_HOST", "localhost"))
port: int = Field(default=int(os.getenv("MYSQL_PORT", "3306")))
user: str = Field(default=os.getenv("MYSQL_USER", "root"))
password: str = Field(default=os.getenv("MYSQL_PASSWORD", ""))
database: Optional[str] = Field(default=os.getenv("MYSQL_DATABASE"))
# Global connection state
current_db = os.getenv("MYSQL_DATABASE", "")
config = DBConfig()
def get_connection():
"""Create a MySQL connection using the current configuration"""
try:
conn = mysql.connector.connect(
host=config.host,
port=config.port,
user=config.user,
password=config.password,
database=config.database if config.database else None
)
return conn
except Error as e:
raise Exception(f"Database connection error: {e}")
@mcp.tool()
def query_sql(query: str) -> Dict[str, Any]:
"""Execute a SELECT query and return the results"""
conn = get_connection()
cursor = conn.cursor(dictionary=True)
try:
cursor.execute(query)
results = cursor.fetchall()
return {
"rows": results[:100], # Limit to 100 rows for safety
"row_count": cursor.rowcount,
"column_names": [desc[0] for desc in cursor.description] if cursor.description else []
}
except Error as e:
raise Exception(f"Query error: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def execute_sql(query: str) -> Dict[str, Any]:
"""Execute a non-SELECT query (INSERT, UPDATE, DELETE, etc.)"""
conn = get_connection()
cursor = conn.cursor()
try:
cursor.execute(query)
conn.commit()
return {
"affected_rows": cursor.rowcount,
"last_insert_id": cursor.lastrowid if cursor.lastrowid else None
}
except Error as e:
conn.rollback()
raise Exception(f"Query error: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def explain_sql(query: str) -> Dict[str, Any]:
"""Get the execution plan for a query"""
conn = get_connection()
cursor = conn.cursor(dictionary=True)
try:
cursor.execute(f"EXPLAIN {query}")
results = cursor.fetchall()
return {
"plan": results
}
except Error as e:
raise Exception(f"EXPLAIN error: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def show_databases() -> Dict[str, Any]:
"""List all available databases"""
conn = get_connection()
cursor = conn.cursor()
try:
cursor.execute("SHOW DATABASES")
results = cursor.fetchall()
return {
"databases": [db[0] for db in results]
}
except Error as e:
raise Exception(f"Error listing databases: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def use_database(database: str) -> Dict[str, Any]:
"""Switch to a different database"""
global config, current_db
# Verify database exists
conn = get_connection()
cursor = conn.cursor()
try:
cursor.execute("SHOW DATABASES")
dbs = [db[0] for db in cursor.fetchall()]
if database not in dbs:
raise ValueError(f"Database '{database}' does not exist")
# Update configuration
config.database = database
current_db = database
return {
"current_database": database,
"status": "success"
}
except Error as e:
raise Exception(f"Error changing database: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def show_tables() -> Dict[str, Any]:
"""List all tables in the current database"""
if not config.database:
raise ValueError("No database selected. Use 'use_database' first.")
conn = get_connection()
cursor = conn.cursor()
try:
cursor.execute("SHOW TABLES")
results = cursor.fetchall()
return {
"database": config.database,
"tables": [table[0] for table in results]
}
except Error as e:
raise Exception(f"Error listing tables: {e}")
finally:
cursor.close()
conn.close()
@mcp.tool()
def describe_table(table: str) -> Dict[str, Any]:
"""Get column definitions for a table"""
if not config.database:
raise ValueError("No database selected. Use 'use_database' first.")
conn = get_connection()
cursor = conn.cursor(dictionary=True)
try:
cursor.execute(f"DESCRIBE {table}")
columns = cursor.fetchall()
# Get index information
cursor.execute(f"SHOW INDEX FROM {table}")
indexes = cursor.fetchall()
return {
"table": table,
"columns": columns,
"indexes": indexes
}
except Error as e:
raise Exception(f"Error describing table: {e}")
finally:
cursor.close()
conn.close()
@mcp.resource(f"schema://{'{database}'}")
def get_database_schema(database: Optional[str] = None) -> str:
"""Get the full schema of a database as a resource"""
db_to_use = database or config.database
if not db_to_use:
raise ValueError("No database specified or selected")
conn = get_connection()
cursor = conn.cursor()
schema = []
try:
# Switch to the specified database
cursor.execute(f"USE {db_to_use}")
# Get all tables
cursor.execute("SHOW TABLES")
tables = [table[0] for table in cursor.fetchall()]
# Get CREATE TABLE statements for each table
for table in tables:
cursor.execute(f"SHOW CREATE TABLE {table}")
create_stmt = cursor.fetchone()[1]
schema.append(create_stmt)
return "\n\n".join(schema)
except Error as e:
raise Exception(f"Error getting schema: {e}")
finally:
cursor.close()
conn.close()
@mcp.prompt()
def write_query_for_task(task: str) -> str:
"""Help Claude write an optimal SQL query for a given task"""
return f"""Task: {task}
Please write an SQL query that accomplishes this task efficiently.
Some guidelines:
1. Use appropriate JOINs (INNER, LEFT, RIGHT) based on the data relationships
2. Filter data in the WHERE clause to minimize data processing
3. Consider using indexes for better performance
4. Use appropriate aggregation functions when needed
5. Format the query with clear indentation for readability
If you need to see the database schema first, you can access it using the schema:// resource.
"""
@mcp.prompt()
def analyze_query_performance(query: str) -> str:
"""Help Claude analyze the performance of a query"""
return f"""Query: {query}
Please analyze this query for performance issues:
1. First, use the explain_sql tool to get the execution plan
2. Look for table scans instead of index usage
3. Check if the joins are efficient
4. Identify if the query can be optimized with better indexes
5. Suggest concrete improvements to make the query more efficient
"""
if __name__ == "__main__":
# Run the server directly
mcp.run()
환경 설정
MySQL 연결 세부정보로 .env 파일을 만듭니다.
MYSQL_HOST=localhost
MYSQL_PORT=3306
MYSQL_USER=root
MYSQL_PASSWORD=your_password
MYSQL_DATABASE=initial_database
클로드와 함께 달리기
- 아직 설치하지 않았다면 Claude에 MySQL MCP 앱을 설치하세요.
fastmcp install mysql_mcp.py
- Claude에서 도구 선택기에서 MySQL MCP 도구를 선택하세요.
- 이제 Claude에게 다음을 요청할 수 있습니다.
- "사용 가능한 모든 데이터베이스를 보여주세요"
- "고객 데이터베이스에는 어떤 테이블이 있나요?"
- "지난주에 접수된 모든 주문을 조회합니다"
- "users 테이블에 대한 스키마를 보여주세요"
- "이 느린 쿼리를 최적화하는 데 도움을 주세요: SELECT * FROM orders JOIN users ON user_id WHERE status = 'pending'"
지역적으로 실행
개발을 위해 MCP 서버를 로컬로 실행할 수도 있습니다.
# Run directly
python mysql_mcp.py
# Or use FastMCP development mode
fastmcp dev mysql_mcp.py
고급 사용법
여러 데이터베이스에 연결
여러 데이터베이스에 대한 구성을 만들고 데이터베이스 간에 전환할 수 있습니다.
@mcp.tool()
def save_connection(name: str, host: str, port: int, user: str, password: str, database: Optional[str] = None) -> Dict[str, Any]:
"""Save a named database connection configuration"""
# Implementation details...
거래 지원
원자성이 필요한 작업의 경우:
@mcp.tool()
def run_transaction(queries: List[str]) -> Dict[str, Any]:
"""Run multiple queries in a single transaction"""
# Implementation details...
스키마 분석
데이터베이스 구조에 대한 더 심층적인 통찰력 제공:
@mcp.tool()
def analyze_table_relationships() -> Dict[str, Any]:
"""Analyze foreign key relationships between tables"""
# Implementation details...
보안 고려 사항
- MySQL-MCP는 데이터베이스에 대해 직접 SQL을 실행합니다.
- Claude가 액세스할 수 있어야 하는 데이터베이스에만 액세스 권한을 제공하세요.
- 안전을 위해 읽기 전용 사용자를 사용하는 것을 고려하세요
- 프로덕션 데이터로 작업할 때 실행하기 전에 모든 쿼리를 검토하세요.
- 최소 권한 원칙에 따라 연결 권한을 제한합니다.
문제 해결
문제가 발생하는 경우:
- 올바른 데이터베이스 자격 증명을 확인하려면
.env 파일을 확인하세요. - MySQL 서버가 실행 중이고 접근 가능한지 확인하세요.
- 자세한 오류 정보는 로그 파일(
mysql_mcp.log )을 확인하세요. - 모든 종속성이 설치되었는지 확인하세요:
pip install -r requirements.txt
기여하다
기여를 환영합니다! 풀 리퀘스트를 제출해 주세요.
특허
MIT
Model Context Protocol 서버를 구축하는 빠르고 Python적인 방법인 FastMCP 로 구축되었습니다.