MCP SQL 서버

대화형 AI 인터페이스를 통해 SQL 데이터베이스 상호작용 도구를 제공하는 FastMCP 서버입니다.

개요

이 프로젝트는 대화형 AI 인터페이스를 통해 MS SQL Server 작업을 제공하는 서버를 구축합니다. FastMCP 프레임워크를 사용하여 SQL 데이터 쿼리 및 조작 도구를 제공하여 사용자가 자연어를 사용하여 데이터베이스와 상호 작용할 수 있도록 지원합니다.

특징

SQL 쿼리를 실행하고 결과를 봅니다.
데이터베이스에서 사용 가능한 테이블 나열
열 정보를 사용하여 테이블 구조를 설명합니다.
비쿼리 작업(INSERT, UPDATE, DELETE) 실행
시스템에서 사용 가능한 ODBC 드라이버 나열
데이터베이스 정보 및 서버 세부 정보 보기

요구 사항

파이썬 3.7 이상
파이오디씨
비동기
FastMCP 프레임워크
마이크로소프트 SQL 서버
SQL Server용 ODBC 드라이버 17

설치

Python 종속성 설치:

지엑스피1

Microsoft SQL Server가 설치되어 있고 SQL Server용 ODBC 드라이버 17이 설치되어 있는지 확인하세요.
스크립트에서 연결 설정을 구성합니다.

# Connection parameters
SERVER = "server\\instance"  # Change to your SQL Server instance
DATABASE = "db_name"              # Change to your database name

용법

서버를 실행합니다:

python mcp_sql_server.py

서버가 초기화되고 지정된 SQL Server 데이터베이스에 대한 연결이 설정됩니다.

사용 가능한 도구

쿼리_SQL

SQL 쿼리를 실행하고 결과를 반환합니다.

query_sql(query: str = None) -> str

쿼리가 제공되지 않으면 기본적으로 SELECT * FROM [dbo].[Table_1] 이 사용됩니다.
쿼리 결과를 형식화된 문자열로 반환합니다.

리스트_테이블

데이터베이스에서 사용 가능한 모든 테이블을 나열합니다.

list_tables() -> str

테이블 이름 목록을 문자열로 반환합니다.

설명_테이블

특정 테이블의 구조를 가져옵니다.

describe_table(table_name: str) -> str

table_name : 설명할 테이블의 이름
이름과 데이터 유형을 포함한 열 정보를 반환합니다.

실행_비쿼리

INSERT, UPDATE, DELETE 또는 기타 쿼리가 아닌 SQL 문을 실행합니다.

execute_nonquery(sql: str) -> str

sql : 실행할 SQL 문
영향을 받은 행 수를 포함한 작업 결과를 반환합니다.
자동으로 트랜잭션(커밋/롤백)을 처리합니다.

list_odbc_drivers

시스템에서 사용 가능한 모든 ODBC 드라이버를 나열합니다.

list_odbc_drivers() -> str

설치된 ODBC 드라이버의 쉼표로 구분된 목록을 반환합니다.

데이터베이스 정보

연결된 데이터베이스에 대한 일반 정보를 가져옵니다.

database_info() -> str

서버 이름, 데이터베이스 이름, SQL Server 버전, 현재 서버 시간 및 테이블 수를 반환합니다.

건축학

서버는 작업 차단을 방지하기 위해 비동기 아키텍처를 사용합니다.

수명 주기 관리 : app_lifespan 컨텍스트 관리자는 데이터베이스 연결 설정 및 해제를 처리합니다.
비차단 작업 : 데이터베이스 작업은 asyncio.get_event_loop().run_in_executor() 사용하여 별도의 스레드에서 실행되어 메인 이벤트 루프가 차단되는 것을 방지합니다.
오류 처리 : 모든 작업에는 유용한 오류 메시지를 통한 포괄적인 오류 처리가 포함됩니다.

오류 처리

서버는 다양한 오류 조건을 처리합니다.

데이터베이스 연결 실패
SQL 쿼리 구문 오류
테이블을 찾을 수 없음 오류
권한 관련 문제

모든 오류는 기록되고 해당 오류 메시지가 클라이언트에게 반환됩니다.

사용자 정의

새로운 데이터베이스 도구를 추가하거나 기존 도구를 수정하려면 기존 도구에서 사용하는 패턴을 따르세요.

@mcp.tool()
async def your_new_tool(ctx: Context, param1: str) -> str:
    """Documentation for your tool"""
    try:
        conn = ctx.request_context.lifespan_context["conn"]
        
        if conn is None:
            return "Database connection is not available."
            
        def your_db_operation():
            # Your database operations here
            pass
            
        loop = asyncio.get_event_loop()
        result = await loop.run_in_executor(None, your_db_operation)
        
        # Process and return results
        return "Your result"
    except Exception as e:
        return f"Error: {str(e)}"

보안 고려 사항

서버는 Windows 인증을 사용합니다("Trusted_Connection=yes")
SQL 주입을 방지하기 위해 SQL 쿼리에 대한 입력 검증을 구현하는 것을 고려하세요.
최소 권한 원칙에 따라 데이터베이스 사용자 권한을 제한합니다.

문제 해결

일반적인 문제:

연결 오류 : SQL Server 인스턴스 이름을 확인하고 실행 중인지 확인하세요.
ODBC 드라이버 오류 : SQL Server용 ODBC 드라이버 17이 설치되어 있는지 확인하세요.
권한 오류 : 애플리케이션을 실행하는 Windows 사용자에게 적절한 SQL Server 권한이 있는지 확인하세요.

특허

[귀하의 라이센스 정보]

연락하다

[귀하의 연락처]

MCP SQL Server

Integrations