Свой MCP-сервер за двадцать минут - как подключить Claude и Cursor к базе данных и логам

# Устанавливаем uv, если еще не стоит curl -LsSf https://astral.sh/uv/install.sh | sh

import os import re import psycopg2 from psycopg2.extras import RealDictCursor from mcp.server.fastmcp import FastMCP # Инициализируем сервер с понятным именем mcp = FastMCP("Production-Postgres-Explorer") DATABASE_URL = os.getenv( "DATABASE_URL", "postgresql://postgres:postgres@localhost:5432/my_database" ) def get_connection(): conn = psycopg2.connect(DATABASE_URL) # Принудительно включаем режим только для чтения на уровне сессии conn.set_session(readonly=True, autocommit=True) return conn @mcp.resource("db://schema") def get_database_schema() -> str: """Возвращает список всех таблиц и их колонок в текущей базе данных.""" query = """ SELECT table_name, column_name, data_type FROM information_schema.columns WHERE table_schema = 'public' ORDER BY table_name, ordinal_position; """ with get_connection() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(query) rows = cur.fetchall() schema_text = "Схема базы данных:\n" current_table = "" for row in rows: if row["table_name"] != current_table: current_table = row["table_name"] schema_text += f"\nТаблица: {current_table}\n" schema_text += f" - {row['column_name']} ({row['data_type']})\n" return schema_text @mcp.tool() def execute_sql(query: str) -> str: """Выполняет безопасный SQL-запрос только на чтение данных (SELECT). Аргументы: query: SQL-запрос на диалекте PostgreSQL. """ # Дополнительная проверка на опасные ключевые слова dangerous_keywords = r"\b(INSERT|UPDATE|DELETE|DROP|ALTER|TRUNCATE|GRANT|REVOKE)\b" if re.search(dangerous_keywords, query, re.IGNORECASE): return "Ошибка: Разрешены только операции чтения (SELECT)." # Жесткое ограничение на количество строк, чтобы не забить контекст модели if not re.search(r"\bLIMIT\b", query, re.IGNORECASE): query = query.rstrip("; ") + " LIMIT 50;" try: with get_connection() as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute(query) results = cur.fetchall() if not results: return "Запрос выполнен успешно. Данные не найдены." # Форматируем вывод в компактный текстовый вид header = " | ".join(results[0].keys()) divider = "-" * len(header) lines = [header, divider] for r in results: lines.append(" | ".join(str(v) for v in r.values())) return "\n".join(lines) except Exception as err: return f"Ошибка выполнения SQL: {str(err)}" if __name__ == "__main__": # Запускаем сервер через стандартный поток ввода-вывода mcp.run(transport="stdio")

{ "mcpServers": { "postgres-explorer": { "command": "uv", "args": [ "run", "--with", "mcp[cli]", "--with", "psycopg2-binary", "/абсолютный/путь/к/server.py" ], "env": { "DATABASE_URL": "postgresql://postgres:password@127.0.0.1:5432/production_db" } } } }

{ "mcpServers": { "postgres-explorer": { "command": "uv", "args": [ "run", "--with", "mcp[cli]", "--with", "psycopg2-binary", "C:\\scripts\\server.py" ], "env": { "DATABASE_URL": "postgresql://postgres:password@localhost:5432/analytics" } } } }

SELECT id, status FROM users WHERE email = 'ivan@example.com';

SELECT id, amount, status, error_code FROM payments WHERE user_id = 42 ORDER BY created_at DESC LIMIT 1;