대상: Ubuntu 24.04 LTS / MySQL 8.x / Python 3.12 / Python MCP SDK v2. 조회뿐 아니라 INSERT·UPDATE·DELETE를 지원하되, 모든 변경은 서버 관리자 SSH 승인 이후에만 실행합니다. 운영 DB에 바로 적용하지 말고 테스트 DB에서 충분히 검증하세요.
1. 구성과 보안 원칙
ChatGPT → Secure MCP Tunnel(HTTPS 443 아웃바운드) → Ubuntu의 Python MCP(127.0.0.1:8000/mcp) → MySQL(127.0.0.1:3306). MySQL 포트는 외부에 공개하지 않습니다. 테이블·컬럼 허용 목록, 단일 행 변경, 5분 만료 승인, 트랜잭션, 중복 실행 방지 로그를 적용합니다.
2. Ubuntu 설치
|
1 2 3 4 5 6 7 8 9 10 11 |
sudo apt update sudo apt install -y python3 python3-venv python3-pip default-mysql-client curl ca-certificates sudo install -d -m 755 /opt/mysql-mcp sudo chown "$USER":"$USER" /opt/mysql-mcp cd /opt/mysql-mcp python3 -m venv .venv .venv/bin/python -m pip install --upgrade pip .venv/bin/pip install "mcp>=2,<3" mysql-connector-python .venv/bin/pip freeze > requirements.lock.txt |
3. 테스트 MySQL 스키마 및 권한
MySQL 관리자에서 아래 SQL을 실행합니다. 샘플 DB의 계정 비밀번호는 반드시 변경하고, 운영 테이블에는 테이블 단위 GRANT를 따로 설정합니다.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 |
CREATE DATABASE IF NOT EXISTS sampledb CHARACTER SET utf8mb4; USE sampledb; CREATE TABLE IF NOT EXISTS mcp_demo_items ( id BIGINT UNSIGNED AUTO_INCREMENT PRIMARY KEY, name VARCHAR(100) NOT NULL, price DECIMAL(12,2) NOT NULL, stock INT NOT NULL DEFAULT 0 ) ENGINE=InnoDB; CREATE TABLE IF NOT EXISTS mcp_change_log ( request_id CHAR(32) PRIMARY KEY, action VARCHAR(10) NOT NULL, table_name VARCHAR(64) NOT NULL, target_id BIGINT UNSIGNED NULL, applied_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB; INSERT INTO mcp_demo_items(name, price, stock) VALUES ('키보드',35000,10),('마우스',15000,20); CREATE USER 'mcp_app'@'127.0.0.1' IDENTIFIED BY 'CHANGE_TO_STRONG_PASSWORD'; GRANT SELECT,INSERT,UPDATE,DELETE ON sampledb.mcp_demo_items TO 'mcp_app'@'127.0.0.1'; GRANT SELECT,INSERT ON sampledb.mcp_change_log TO 'mcp_app'@'127.0.0.1'; SHOW GRANTS FOR 'mcp_app'@'127.0.0.1'; |
4. 허용 테이블·컬럼 정책
파일 /opt/mysql-mcp/policy.json을 생성합니다. 삭제가 불필요하다면 delete를 false로 유지합니다. 정수형 자동증가 기본키, 단일 행 작업만 지원하는 샘플입니다.
|
1 2 3 4 5 6 7 8 9 10 |
{ "mcp_demo_items": { "pk": "id", "read": ["id","name","price","stock"], "write": ["name","price","stock"], "delete": false } } |
5. 서버 코드: server.py
/opt/mysql-mcp/server.py에 아래 코드를 저장합니다. list_tables, describe_table, list_rows, get_row, prepare_change, apply_change의 6개 도구를 제공합니다.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 |
import os, re, json, uuid, fcntl from pathlib import Path from datetime import datetime, timezone, timedelta import mysql.connector from mcp.server import MCPServer from mcp.types import ToolAnnotations # Only localhost exposure; tunnel handles external transport. mcp = MCPServer("Ubuntu-MySQL") POLICY = json.loads(Path("/opt/mysql-mcp/policy.json").read_text()) PENDING = Path("/var/lib/mysql-mcp/pending") IDENT = re.compile(r"^[A-Za-z_][A-Za-z0-9_]{0,63}$") def db(): return mysql.connector.connect( host="127.0.0.1", port=3306, user=os.environ["MYSQL_USER"], password=os.environ["MYSQL_PASSWORD"], database=os.environ["MYSQL_DATABASE"], autocommit=False, connection_timeout=5) def quoted(name): if not IDENT.fullmatch(name): raise ValueError("Bad identifier") return "`" + name + "`" def policy(table): p = POLICY.get(table) if not p: raise ValueError("Table not allowed") for c in [table, p["pk"], *p["read"], *p["write"]]: quoted(c) if not set(p["write"]).issubset(set(p["read"]) - {p["pk"]}): raise ValueError("Invalid policy") return p def norm(obj): return json.loads(json.dumps(obj, default=str, ensure_ascii=False)) def one(conn, table, row_id, lock=False): p = policy(table) cols = ",".join(map(quoted, p["read"])) sql = (f"SELECT {cols} FROM {quoted(table)} " f"WHERE {quoted(p['pk'])}=%s LIMIT 1" + (" FOR UPDATE" if lock else "")) cur = conn.cursor(dictionary=True, buffered=True) try: cur.execute(sql, (row_id,)) return norm(cur.fetchone()) finally: cur.close() def valid_values(table, action, values): p = policy(table) if action == "delete": if values: raise ValueError("DELETE has no values") return {} if not isinstance(values, dict) or not values: raise ValueError("Nonempty values required") if not set(values).issubset(p["write"]): raise ValueError("Column not allowed") for v in values.values(): if v is not None and type(v) not in (str, int, float, bool): raise ValueError("Only scalar values") if isinstance(v, str) and len(v) > 200: raise ValueError("Value too long") return values def request_path(rid): if not re.fullmatch(r"[a-f0-9]{32}", rid): raise ValueError("Invalid request ID") return PENDING / (rid + ".json") @mcp.tool(annotations=ToolAnnotations(read_only_hint=True)) def list_tables() -> dict: """List approved tables.""" return {"tables": list(POLICY)} @mcp.tool(annotations=ToolAnnotations(read_only_hint=True)) def describe_table(table: str) -> dict: """Describe allowed fields, primary key, write policy.""" return policy(table) @mcp.tool(annotations=ToolAnnotations(read_only_hint=True)) def list_rows(table: str, limit: int = 10) -> dict: """Read up to 10 approved-column rows.""" p = policy(table) columns = ",".join(map(quoted, p["read"])) sql = (f"SELECT {columns} FROM {quoted(table)} " f"ORDER BY {quoted(p['pk'])} DESC LIMIT %s") conn = db() try: cur = conn.cursor(dictionary=True, buffered=True) try: cur.execute(sql, (max(1, min(limit, 10)),)) return {"rows": norm(cur.fetchall())} finally: cur.close() finally: conn.close() @mcp.tool(annotations=ToolAnnotations(read_only_hint=True)) def get_row(table: str, row_id: int) -> dict: """Read one row by positive integer primary key.""" if row_id < 1: raise ValueError("Invalid row_id") conn = db() try: return {"row": one(conn, table, row_id)} finally: conn.close() @mcp.tool() def prepare_change(action: str, table: str, row_id: int | None = None, values: dict | None = None) -> dict: """Preview one INSERT, UPDATE, or DELETE. No data writes yet.""" if action not in ("insert", "update", "delete"): raise ValueError("Unsupported action") p = policy(table) if action == "delete" and not p.get("delete", False): raise ValueError("DELETE disabled") if action == "insert" and row_id is not None: raise ValueError("INSERT ID must auto-increment") if action != "insert" and (row_id is None or row_id < 1): raise ValueError("Positive row_id required") values = valid_values(table, action, values) before = None if action != "insert": conn = db() try: before = one(conn, table, row_id) finally: conn.close() if before is None: raise ValueError("Row missing") rid = uuid.uuid4().hex item = dict(request_id=rid, status="pending", approved=False, action=action, table=table, row_id=row_id, values=values, before=before, expires=(datetime.now(timezone.utc) + timedelta(minutes=5)).isoformat()) fd = os.open(request_path(rid), os.O_WRONLY | os.O_CREAT | os.O_EXCL, 0o600) with os.fdopen(fd, "w") as f: json.dump(item, f, ensure_ascii=False) return {"request_id": rid, "action": action, "table": table, "row_id": row_id, "before": before, "values": values, "state": "awaiting SSH approval"} @mcp.tool() def apply_change(request_id: str) -> dict: """Execute one approved change, with audit and retry protection.""" with request_path(request_id).open("r+") as f: fcntl.flock(f, fcntl.LOCK_EX) req = json.load(f) if req["status"] == "applied": return {"status": "already_applied"} if not req["approved"] or req["status"] != "pending": raise PermissionError("Local approval required") if datetime.now(timezone.utc) >= datetime.fromisoformat(req["expires"]): raise ValueError("Request expired") action, table, row_id = req["action"], req["table"], req["row_id"] p = policy(table) if action == "delete" and not p.get("delete", False): raise ValueError("DELETE disabled") values = valid_values(table, action, req["values"]) conn = db() try: conn.start_transaction() cur = conn.cursor(dictionary=True, buffered=True) try: cur.execute("SELECT request_id FROM mcp_change_log WHERE request_id=%s", (request_id,)) if cur.fetchone(): conn.rollback() return {"status": "already_applied"} if action != "insert": current = one(conn, table, row_id, lock=True) if current != req["before"]: raise ValueError("Row changed since preview") if action == "insert": cols = list(values) sql = (f"INSERT INTO {quoted(table)} (" + ",".join(map(quoted, cols)) + ") VALUES (" + ",".join(["%s"] * len(cols)) + ")") cur.execute(sql, tuple(values[c] for c in cols)) row_id = cur.lastrowid elif action == "update": cols = list(values) sql = (f"UPDATE {quoted(table)} SET " + ",".join(f"{quoted(c)}=%s" for c in cols) + f" WHERE {quoted(p['pk'])}=%s") cur.execute(sql, (*[values[c] for c in cols], row_id)) elif action == "delete": cur.execute( f"DELETE FROM {quoted(table)} WHERE {quoted(p['pk'])}=%s", (row_id,)) if cur.rowcount != 1: raise ValueError("Exactly one changed row required") cur.execute( "INSERT INTO mcp_change_log(request_id,action,table_name,target_id) " "VALUES(%s,%s,%s,%s)", (request_id, action, table, row_id)) conn.commit() req["status"] = "applied" f.seek(0); json.dump(req, f, ensure_ascii=False) f.truncate(); f.flush(); os.fsync(f.fileno()) return {"status": "applied", "request_id": request_id, "action": action, "row_id": row_id} finally: cur.close() except Exception: conn.rollback() raise finally: conn.close() if __name__ == "__main__": mcp.run(transport="streamable-http", host="127.0.0.1", port=8000, stateless_http=True, json_response=True) |
6. Ubuntu 관리자 승인 코드: approve.py
이 파일은 MCP 도구로 제공하지 않습니다. 관리자가 SSH에서 요청 ID와 변경 전 값을 직접 확인해야 합니다.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 |
import fcntl, json, os, re, sys from datetime import datetime, timezone from pathlib import Path rid = sys.argv[1] if len(sys.argv) == 2 else "" if not re.fullmatch(r"[0-9a-f]{32}", rid): sys.exit("Usage: approve.py 32-character-request-id") path = Path("/var/lib/mysql-mcp/pending") / (rid + ".json") with path.open("r+") as f: fcntl.flock(f, fcntl.LOCK_EX) item = json.load(f) if item["status"] != "pending": sys.exit("Not pending") if datetime.now(timezone.utc) >= datetime.fromisoformat(item["expires"]): sys.exit("Expired") print(json.dumps(item, ensure_ascii=False, indent=2)) if input("Type APPROVE: ").strip() != "APPROVE": sys.exit("Cancelled") item["approved"] = True f.seek(0) json.dump(item, f, ensure_ascii=False) f.truncate() f.flush() os.fsync(f.fileno()) print("Approved") |
7. 환경 파일·권한·systemd
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 |
sudo useradd --system --home-dir /var/lib/mysql-mcp --shell /usr/sbin/nologin mcpreader sudo install -d -m 700 -o mcpreader -g mcpreader /var/lib/mysql-mcp/pending sudo install -d -m 700 /etc/mysql-mcp sudo nano /etc/mysql-mcp/mcp.env # 아래 내용을 입력 (실제 비밀번호로 변경): # MYSQL_USER=mcp_app # MYSQL_PASSWORD=CHANGE_TO_STRONG_PASSWORD # MYSQL_DATABASE=sampledb sudo chown root:root /etc/mysql-mcp/mcp.env sudo chmod 600 /etc/mysql-mcp/mcp.env sudo chown -R root:root /opt/mysql-mcp sudo chmod 755 /opt/mysql-mcp sudo nano /etc/systemd/system/mysql-mcp.service |
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 |
[Unit] Description=Python MySQL MCP After=network.target [Service] Type=simple User=mcpreader Group=mcpreader WorkingDirectory=/opt/mysql-mcp EnvironmentFile=/etc/mysql-mcp/mcp.env ExecStart=/opt/mysql-mcp/.venv/bin/python /opt/mysql-mcp/server.py Restart=on-failure RestartSec=5 NoNewPrivileges=true PrivateTmp=true ProtectHome=true ProtectSystem=strict ReadWritePaths=/var/lib/mysql-mcp/pending [Install] WantedBy=multi-user.target |
|
1 2 3 4 5 6 7 |
sudo /opt/mysql-mcp/.venv/bin/python -m py_compile /opt/mysql-mcp/server.py /opt/mysql-mcp/approve.py sudo systemctl daemon-reload sudo systemctl enable --now mysql-mcp sudo systemctl status mysql-mcp sudo journalctl -u mysql-mcp -n 50 --no-pager |
8. 로컬 조회 및 UPDATE 테스트
SDK의 세션/프로토콜 버전에 따라 헤더가 달라질 수 있으므로 MCP Inspector 사용도 권장합니다. 로컬 CLI 테스트 예시는 다음과 같습니다.
|
1 2 3 4 5 6 7 |
curl -sS http://127.0.0.1:8000/mcp \ -H 'Content-Type: application/json' \ -H 'Accept: application/json, text/event-stream' \ -H 'MCP-Protocol-Version: 2026-07-28' \ -d '{"jsonrpc":"2.0","id":1,"method":"tools/call","params":{"name":"list_rows","arguments":{"table":"mcp_demo_items","limit":5}}}' |
UPDATE 예: prepare_change에 {"action":"update","table":"mcp_demo_items","row_id":1,"values":{"price":32000}} 전달 → 반환된 request_id를 기록 → SSH에서 승인 → apply_change(request_id) 호출 → SELECT로 결과 확인.
|
1 2 3 4 |
sudo python3 /opt/mysql-mcp/approve.py 실제_32자리_request_id # APPROVE 입력 후 ChatGPT/MCP 클라이언트에서 apply_change 호출 |
9. Secure MCP Tunnel · ChatGPT 등록
OpenAI Platform에서 tunnel_id와 Runtime API Key 발급 → tunnel-client 최신 실행 파일 설치 → 다음과 같이 HTTP MCP 프로필을 초기화합니다. tunnel-client CLI 인자는 공식 문서를 최신 기준으로 확인하세요.
|
1 2 3 4 5 6 7 8 |
export CONTROL_PLANE_API_KEY="YOUR_RUNTIME_API_KEY" tunnel-client help quickstart tunnel-client init --profile local-mysql --tunnel-id YOUR_TUNNEL_ID \ --mcp-server-url http://127.0.0.1:8000/mcp tunnel-client doctor --profile local-mysql --explain tunnel-client run --profile local-mysql |
ChatGPT → Plugins → + → Add custom MCP server → Connection: Tunnel에서 터널을 선택하고 도구를 확인합니다. 조직별 연결 권한이 필요하며, 2026년 10월 기준 사용자 지정 MCP 쓰기/수정 기능은 Business·Enterprise·Edu에서 베타 제공됩니다. 개인 Plus에서는 동일한 쓰기 연결 기능이 제공된다고 보장할 수 없습니다.
10. 운영 시 주의사항
이 코드는 학습·테스트용 구현 예제이며 실제 서버에서 빌드·통합 검증하지 않았습니다. 운영 반영 전 DB 백업, 개인정보 컬럼 제거, OAuth/워크스페이스 권한 설정, 쿼리 타임아웃·요청 제한, 변경 로그 모니터링, 트리거·외래키 연쇄 작업 확인이 필요합니다. DB 사용자에게 CREATE/ALTER/DROP/GRANT 권한을 주지 마세요.