Создаём свой MCP-сервер за 20 минут

Полный код и деплой MCP-сервера с инструментами. Работает с Claude Desktop, Cursor, Continue.

📊 Средний⏱ 20 мин

# 1. MCP ПРОТОКОЛ ИЗНУТРИ

MCP использует JSON-RPC 2.0 для обмена сообщениями. Четыре основных метода: initialize (установка соединения), tools/list (список инструментов), tools/call (вызов), resources/read (чтение ресурсов).

// Запрос: клиент запрашивает список инструментов
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/list",
  "params": {}
}

// Ответ: сервер возвращает описание инструментов
{
  "jsonrpc": "2.0",
  "id": 1,
  "result": {
    "tools": [
      {
        "name": "run_query",
        "description": "Выполняет SQL-запрос",
        "inputSchema": {
          "type": "object",
          "properties": {
            "sql": { "type": "string" }
          }
        }
      }
    ]
  }
}

# 2. MCP-СЕРВЕР С HTTP ТРАНСПОРТОМ

Для деплоя на сервер используйте Streamable HTTP вместо stdio. Это позволяет запускать MCP-сервер как веб-сервис и подключать несколько клиентов.

from mcp.server import Server
from mcp.server.http import create_http_server

server = Server("pg-mcp-server")

# ... регистрация инструментов ...

if __name__ == "__main__":
    app = create_http_server(server)
    app.run(host="0.0.0.0", port=8000)

HTTP-транспорт поддерживает server-sent events (SSE) для стриминга ответов. Клиент подключается по URL: http://your-server:8000/mcp. Claude Desktop тоже поддерживает HTTP-транспорт через поле "url" в конфиге.

# 3. TOOLS ДЛЯ POSTGRESQL

Создадим 4 инструмента для работы с PostgreSQL: list_tables, describe_table, run_query, export_csv. Claude становится администратором базы данных.

import psycopg2
import csv, io

CONN = "postgresql://user:pass@localhost:5432/mydb"

@server.tool()
def list_tables() -> str:
    """Показывает все таблицы в базе данных"""
    sql = "SELECT table_name FROM information_schema.tables WHERE table_schema='public'"
    with psycopg2.connect(CONN) as c:
        return str(c.cursor().execute(sql).fetchall())

@server.tool()
def describe_table(table: str) -> str:
    """Показывает структуру таблицы: колонки и типы"""
    sql = f"SELECT column_name, data_type FROM information_schema.columns WHERE table_name='{table}'"
    with psycopg2.connect(CONN) as c:
        return str(c.cursor().execute(sql).fetchall())

@server.tool()
def run_query(sql: str, limit: int = 100) -> str:
    """Выполняет SQL-запрос (только SELECT, лимит 100 строк)"""
    if not sql.strip().upper().startswith("SELECT"):
        return "Разрешены только SELECT-запросы"
    with psycopg2.connect(CONN) as c:
        return str(c.cursor().execute(sql).fetchmany(limit))

@server.tool()
def export_csv(sql: str) -> str:
    """Экспортирует результат запроса в CSV"""
    with psycopg2.connect(CONN) as c:
        rows = c.cursor().execute(sql).fetchall()
        buf = io.StringIO()
        csv.writer(buf).writerows(rows)
        return buf.getvalue()

# 4. ДЕПЛОЙ НА VPS

Запустите MCP-сервер как systemd-сервис за nginx reverse proxy с SSL. Сервер будет доступен по HTTPS и готов к продакшену.

# /etc/systemd/system/mcp-server.service
[Unit]
Description=MCP Server
After=network.target

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/mcp-server
ExecStart=/usr/bin/python3 server.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

# Запуск:
sudo systemctl daemon-reload && sudo systemctl enable --now mcp-server

# nginx reverse proxy + SSL
location /mcp {
    proxy_pass http://127.0.0.1:8000;
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
}

# 5. МОНИТОРИНГ И ЛОГИ

Настройте мониторинг через Prometheus метрики и просмотр логов через journalctl. Отслеживайте количество вызовов инструментов и время ответа.

# Просмотр логов MCP-сервера
journalctl -u mcp-server -f # real-time логи
journalctl -u mcp-server --since "1 hour ago" # за последний час

# Prometheus метрики в server.py:
from prometheus_client import Counter, Histogram

tool_calls = Counter('mcp_tool_calls', 'Tool calls', ['tool'])
tool_latency = Histogram('mcp_tool_latency', 'Latency', ['tool'])

# В каждом инструменте:
tool_calls.labels('run_query').inc()
with tool_latency.labels('run_query').time():
    # ... тело инструмента ...

# 6. ТЕСТИРОВАНИЕ MCP-СЕРВЕРА

Напишите тесты с pytest, мокируя stdio. Проверьте JSON-RPC ответы и корректность работы инструментов.

import pytest, json
from mcp.client import ClientSession
from mcp.testing import stdio_client

@pytest.mark.asyncio
async def test_tools_list():
    """Проверка: tools/list возвращает список инструментов"""
    async with stdio_client("python3", ["server.py"]) as (read, write):
        async with ClientSession(read, write) as session:
            tools = await session.list_tools()
            assert len(tools) >= 4
            assert "run_query" in [t.name for t in tools]

@pytest.mark.asyncio
async def test_run_query():
    """Проверка: вызов run_query возвращает корректный JSON-RPC ответ"""
    async with stdio_client("python3", ["server.py"]) as (read, write):
        async with ClientSession(read, write) as session:
            result = await session.call_tool("run_query", {"sql": "SELECT 1"})
            assert result is not None

🔗 Полезные ссылки

📋 MCP Specification 🐍 Python SDK ⚙ systemd Docs 📊 Prometheus