[AI] mcp 로 postgresql 을 컨트롤 해보자.

1. postgres.py

로컬 DB 접속 클래스

from typing import Any, Dict, List, Union
import psycopg2
import psycopg2.extras
from contextlib import contextmanager

class PostgresDB:
    def __init__(self, host: str):
        self.host = host

    @contextmanager
    def get_connection(self, user: str, password: str, database: str):
        conn = psycopg2.connect(
            host=self.host,
            user=user,
            password=password,
            database=database
        )
        conn.autocommit = True
        try:
            yield conn
        finally:
            conn.close()

    def execute_query(self, user: str, password: str, database: str, query: str) -> Union[Dict[str, Any], List[Dict[str, Any]]]:
        try:
            with self.get_connection(user, password, database) as conn:
                with conn.cursor(cursor_factory=psycopg2.extras.DictCursor) as cursor:
                    cursor.execute(query)
                    if cursor.description is not None:
                        result = cursor.fetchall()
                        return [dict(row) for row in result]
                    else:
                        return {"status": "success", "message": "Query executed successfully."}
        except psycopg2.OperationalError:
            return {"status": "error", "message": "DB 연결 실패: ID, 비밀번호 또는 데이터베이스 이름을 확인하세요."}
        except Exception as e:
            return {"status": "error", "message": "쿼리 실행 중 오류 발생: " + str(e)}

# 기본 DB 인스턴스 생성 (이제는 host만 고정)
default_db = PostgresDB(
    host="localhost"
)

2. server.py

그냥 단순하게 쿼리 실행만 해주는 툴을 만든다.

from mcp.server.fastmcp import FastMCP
from database.postgres import default_db

# FastMCP 서버 초기화
mcp = FastMCP("exoluse")

@mcp.tool(
    name="postgresql_execute_query",
    description="""
        당신은 훌륭한 PostgreSQL 쿼리 실행 서버입니다.

        1. 실행하기 전 DB 접속 ID, 비밀번호, 데이터베이스명을 입력해야 합니다.
        2. 일치하면 쿼리 실행 결과를 반환합니다.
        3. 일치하지 않으면 오류 메시지를 반환합니다.
    """
)
def postgresql_execute_query(user: str, password: str, database: str, query: str):
    return default_db.execute_query(user, password, database, query)

if __name__ == "__main__":
    # 서버 초기화 및 실행
    mcp.run(transport='stdio')

------ mcp server 등록 생략 ------

3. claude desktop 에서 구동

3-1. 서버에 접속 요청

3-2. 접속 정보 입력 후

3-3. 테이블 조회 요청 (t_emp)

3-4. 데이터 insert 요청

3-5. 데이터 update 요청

3-6. 데이터 delete 요청

3-7. DDL 도 먹히나?

3-8. 다 된다. db 튜닝도 될까?

Finally

일단... 너무 위험하다. 뭔가 중간 장치가 하나 정도는 있어야 한다는 생각이 든다.
아마 cursor ai 에 적용해도 에이전트가 똑같이 해주겠지?