데이터베이스 MCP 서버
다양한 데이터베이스 시스템에 연결하고 상호작용하기 위한 도구를 제공하는 MCP(Model Context Protocol) 서버입니다.
특징
다중 데이터베이스 지원 : SQLite, PostgreSQL, MySQL/MariaDB 및 SQL Server 데이터베이스에 연결
통합 인터페이스 : 지원되는 모든 데이터베이스 유형에 대한 데이터베이스 작업을 위한 공통 도구
데이터베이스별 확장 : 필요한 경우 데이터베이스별 기능을 위한 특정 도구
스키마 관리 : 테이블 및 인덱스 생성, 변경 및 삭제
쿼리 실행 : 원시 SQL 쿼리를 실행하거나 구조화된 쿼리 도구를 사용합니다.
트랜잭션 지원 : 트랜잭션 시작, 커밋 및 롤백
Related MCP server: SQLite MCP Server
설치
필수 조건
Python 3.8 이상
필수 Python 패키지(pip로 자동 설치됨):
SQLAlchemy
사용하려는 데이터베이스에 따라 다양한 데이터베이스 드라이버가 있습니다.
SQLite(Python에 포함됨)
PostgreSQL:
psycopg2-binaryMySQL/MariaDB:
mysql-connector-pythonSQL 서버:
pyodbc
소스에서 설치
지엑스피1
구성
서버는 환경 변수, 구성 파일을 사용하거나 런타임에 연결 세부 정보를 제공하여 구성할 수 있습니다.
환경 변수
DB_CONFIG_PATH: JSON 구성 파일의 경로DB_CONNECTIONS: 연결 ID의 쉼표로 구분된 목록 또는 연결 세부 정보가 포함된 JSON 문자열
구성 파일 형식
{
"connections": {
"sqlite_conn": {
"type": "sqlite",
"db_path": "/path/to/database.db"
},
"postgres_conn": {
"type": "postgres",
"host": "localhost",
"port": 5432,
"database": "mydatabase",
"user": "myuser",
"password": "mypassword"
}
}
}용법
서버 실행
Claude의 MCP 서버로서
# Run with default settings
python -m db_mcp_server
# Specify a configuration file
python -m db_mcp_server --config /path/to/config.json
# Set logging level
python -m db_mcp_server --log-level DEBUG독립형 웹 서버(모든 LLM용)
# Run as a web server
python -m db_mcp_server.web_server
# Specify host and port
python -m db_mcp_server.web_server --host 0.0.0.0 --port 8000
# Specify configuration file and logging level
python -m db_mcp_server.web_server --config /path/to/config.json --log-level DEBUG사용 가능한 MCP 도구
연결 관리
add_connection: 새로운 데이터베이스 연결을 추가합니다test_connection: 데이터베이스 연결 테스트list_connections: 모든 데이터베이스 연결을 나열합니다.remove_connection: 데이터베이스 연결 제거
쿼리 실행
execute_query: SQL 쿼리를 실행합니다.get_records: 테이블에서 레코드를 가져옵니다insert_record: 테이블에 레코드를 삽입합니다update_record: 테이블의 레코드를 업데이트합니다.delete_record: 테이블에서 레코드 삭제
스키마 관리
list_tables: 데이터베이스의 모든 테이블을 나열합니다.get_table_schema: 테이블의 스키마를 가져옵니다.create_table: 새로운 테이블을 생성합니다drop_table: 테이블 삭제create_index: 테이블에 인덱스를 생성합니다drop_index: 인덱스 삭제alter_table: 테이블 구조 변경
거래 관리
begin_transaction: 거래 시작commit_transaction: 트랜잭션을 커밋합니다.rollback_transaction: 트랜잭션을 롤백합니다.
예시
연결 추가
{
"connection_id": "my_sqlite_db",
"type": "sqlite",
"db_path": "/path/to/database.db"
}쿼리 실행
{
"connection_id": "my_sqlite_db",
"query": "SELECT * FROM users WHERE age > ?",
"params": [21]
}테이블 만들기
{
"connection_id": "my_sqlite_db",
"table": "users",
"columns": [
{
"name": "id",
"type": "INTEGER",
"primary_key": true,
"nullable": false
},
{
"name": "name",
"type": "TEXT",
"nullable": false
},
{
"name": "email",
"type": "TEXT",
"nullable": true
}
]
}레코드 삽입
{
"connection_id": "my_sqlite_db",
"table": "users",
"data": {
"name": "John Doe",
"email": "john@example.com"
}
}개발
테스트 실행
# Run all tests
python -m unittest discover
# Run specific test file
python -m unittest tests.test_sqlite다른 LLM에서 연결하기
독립형 웹 서버로 실행할 경우, 다른 LLM(예: Llama 3)은 HTTP를 통해 데이터베이스 MCP 서버에 연결할 수 있습니다. 서버는 다음과 같은 엔드포인트를 제공합니다.
엔드포인트
/list_tools- GET 또는 POST: 사용 가능한 모든 도구의 목록과 해당 설명 및 입력 스키마를 반환합니다./call_tool- POST: 특정 데이터베이스 도구 실행
예: 다른 LLM에서 호출
이 서버를 다른 LLM과 함께 사용하려면 LLM이 서버에 HTTP 요청을 생성하도록 설정해야 합니다. Llama 3와 같은 LLM에 대한 프롬프트를 구성하는 방법의 예는 다음과 같습니다.
You can interact with a database by making HTTP requests to a database service at http://localhost:8000.
The service provides the following endpoints:
1. To get a list of available tools:
Make a POST request to: http://localhost:8000/list_tools
2. To execute a database tool:
Make a POST request to: http://localhost:8000/call_tool
with a JSON body like:
{
"name": "tool_name",
"arguments": {
"param1": "value1",
"param2": "value2"
}
}
For example, to execute a SQL query, you would make a request like:
POST http://localhost:8000/call_tool
Content-Type: application/json
{
"name": "execute_query",
"arguments": {
"connection_id": "my_db",
"query": "SELECT * FROM users"
}
}클라이언트 통합을 위한 샘플 Python 코드
import requests
import json
# Base URL of the database MCP server
BASE_URL = "http://localhost:8000"
# List available tools
def list_tools():
response = requests.post(f"{BASE_URL}/list_tools")
return response.json()
# Execute a database tool
def call_tool(tool_name, arguments):
payload = {
"name": tool_name,
"arguments": arguments
}
response = requests.post(f"{BASE_URL}/call_tool", json=payload)
return response.json()
# Example: List tables in a database
def list_tables(connection_id):
return call_tool("list_tables", {"connection_id": connection_id})
# Example: Execute a SQL query
def execute_query(connection_id, query, params=None):
return call_tool("execute_query", {
"connection_id": connection_id,
"query": query,
"params": params
})
# Example: Add a new connection
def add_connection(connection_id, db_type, **kwargs):
args = {"connection_id": connection_id, "type": db_type}
args.update(kwargs)
return call_tool("add_connection", args)