На заводе показания датчиков не ждут, пока до них дойдут руки. Они копятся круглосуточно. Если нужно понять, почему в три часа ночи скакнула влажность, мой обычный маршрут выглядит так: SQL-запрос, CSV, короткий скрипт на Python, график в Excel. И только потом начинается поиск аномалии. На один вопрос инженер легко отдаст полдня.
Данные у завода есть. У LLM хватает аналитики. Спотыкается всё об одно место: доступ к данным. Тащить терабайты сырых логов в контекст нейросети дорого и бессмысленно, если нужен ответ на один конкретный вопрос.
В этой схеме я использую Model Context Protocol (MCP), открытый стандарт Anthropic. Клиент получает через него доступ к внешнему инструменту, а сервер решает, какие данные отдавать модели. В нашем случае инструмент один: SQL-запрос к TimescaleDB.
Для стенда нужна база, которая принимает показания почти в реальном времени и не заставляет вручную воевать с временными рядами. Я собрал связку из двух датчиков, MQTT, Telegraf, TimescaleDB и небольшого MCP-сервера.
Что уже работает на стенде
На стенде всего два датчика: температура и влажность. Они отправляют показания через MQTT, Telegraf складывает их в TimescaleDB. CSV здесь не нужен.
Представим ночную смену. В три часа ночи влажность резко выросла. Дежурный не открывает pgAdmin. Он запускает OpenCode (подойдут и Cursor, Claude Desktop, вообще любой клиент с MCP) и пишет по-русски:
Инженер: «Посмотри показания датчиков с температурой и влажностью и покажи максимальные и минимальные значения за 3 дня».
Модель вызывает query_sensor_database, отправляет SELECT в TimescaleDB и возвращает min/max по каждому топику. Цифры я всё равно сверяю с pgAdmin. Однажды модель взяла окно в одни сутки вместо трёх. Вот и весь аргумент против слепой веры в чат.
С температурой почти то же самое. Запрос короче:
Инженер: «Нарисуй гистограмму изменений температуры».
Без агрегации график влажности превращается в плотную линию из точек. Добавляю шаг в 10 минут:
Инженер: «Нарисуй гистограмму изменений влажности с шагом в 10 минут».
Раньше корреляцию температуры и влажности я ковырял в Grafana. Здесь хватает одной фразы в чате:
Инженер: «Проанализируй корреляцию при изменении значений температуры и влажности по времени».
На стенде модель быстро обрабатывает несколько тысяч точек. Но цифры я проверяю сам. В окне времени или в запросе легко промахнуться, а чат всё равно ответит уверенно.
Ниже соберём эту цепочку с нуля.
В цепочке шесть звеньев:
Датчик → Mosquitto (MQTT) → Telegraf → TimescaleDB → MCP-сервер → LLM.
Все команды ниже выполняются на Ubuntu. Начнём с хранилища.
Шаг 1. Собираем хранилище временных рядов
Обычный Postgres справится с небольшим потоком. Но когда показания пишутся постоянно, запросы по времени приходится продумывать отдельно. Я выбрал TimescaleDB из-за гипертаблиц: временной ряд хранится в форме, с которой удобно работать в таких запросах.
Базу и pgAdmin я запускаю в Docker. Стенд поднимается одной командой, а зависимости не расползаются по хосту. Порт 5433 оставляю специально, чтобы не конфликтовать с локальным Postgres:
version: '3.8' services: timescaledb: image: timescale/timescaledb:latest-pg16 container_name: timescaledb restart: unless-stopped environment: POSTGRES_USER: ${DB_USER:-postgres} POSTGRES_PASSWORD: ${DB_PASSWORD:-changeme} POSTGRES_DB: ${DB_NAME:-sensor_data} TZ: UTC ports: - "${DB_PORT:-5433}:5432" # Маппинг нестандартного порта на хост volumes: - timescale_data:/var/lib/postgresql/data - ./backups:/backups - ./init:/docker-entrypoint-initdb.d networks: - iot_network healthcheck: test: ["CMD-SHELL", "pg_isready -U ${DB_USER:-postgres} -d ${DB_NAME:-sensor_data}"] interval: 30s timeout: 10s retries: 5 logging: driver: "json-file" options: max-size: "10m" max-file: "3" deploy: resources: limits: memory: 1G reservations: memory: 512M pgadmin: # Веб-интерфейс для удобного управления БД image: dpage/pgadmin4:latest container_name: pgadmin restart: unless-stopped environment: PGADMIN_DEFAULT_EMAIL: admin@example.com PGADMIN_DEFAULT_PASSWORD: ${PGADMIN_PASSWORD:-changeme} PGADMIN_CONFIG_SERVER_MODE: 'False' ports: - "5050:80" volumes: - pgadmin_data:/var/lib/pgadmin networks: - iot_network depends_on: - timescaledb volumes: timescale_data: name: timescale_production_data pgadmin_data: name: pgadmin_storage networks: iot_network: name: iot_sensor_network driver: bridge
Чтобы база сразу поднялась с нужной схемой, кладём инициализацию в ./init/01-timescale.sql:
-- Подключение к нашей целевой базе \c sensor_data; -- Включаем расширение TimescaleDB CREATE EXTENSION IF NOT EXISTS timescaledb CASCADE; -- Создаем отдельную схему для сенсоров CREATE SCHEMA IF NOT EXISTS sensors; -- Устанавливаем права GRANT ALL PRIVILEGES ON SCHEMA sensors TO postgres; -- Создаем базовую таблицу для данных, которую мы потом сделаем гипертаблицей CREATE TABLE IF NOT EXISTS sensors.mqtt_consumer ( time TIMESTAMPTZ NOT NULL, host TEXT, topic TEXT NOT NULL, value DOUBLE PRECISION NOT NULL ); -- Превращаем таблицу в гипертаблицу TimescaleDB по полю time SELECT create_hypertable('sensors.mqtt_consumer', 'time', if_not_exists => TRUE);
Рядом создадим файл переменных окружения .env:
# Настройки БД DB_USER=postgres DB_PASSWORD=super_password_123 # Обязательно поменяйте на свой в продакшене! DB_NAME=sensor_data DB_PORT=5433 PGADMIN_PASSWORD=super_password_pgadmin
Запускаем нашу базу:
docker-compose up -d
Проверяем, что всё поднялось и работает:
docker compose ps docker exec -it timescaledb psql -U postgres -d sensor_data -c "SELECT extversion FROM pg_extension WHERE extname = 'timescaledb';"
Шаг 2. Настраиваем MQTT-брокер
Показания идут по MQTT, значит нужен брокер, который примет сообщения и раздаст их подписчикам. Для стенда беру Mosquitto: лёгкий, контейнеризованный, с авторизацией через ACL. Публичный брокер без пароля в производстве я бы не оставлял.
Создаём структуру папок на хосте:
mkdir ~/mosquitto && cd ~/mosquitto mkdir config data
Пишем конфигурационный файл ./config/mosquitto.conf:
# Запрещаем анонимный доступ allow_anonymous false # Указываем пути к файлам авторизации внутри контейнера password_file /mosquitto/config/passwd acl_file /mosquitto/config/acl # Стандартный MQTT порт listener 1883 0.0.0.0 # Порт для веб-сокетов (если захотите выводить данные на веб-панель) listener 9001 protocol websockets # Включаем персистентность (сохранение сессий при перезапуске) persistence true persistence_location /mosquitto/data/
Права доступа к топикам задаём в файле ./config/acl:
user admin topic readwrite sensors/#
Пароли в открытом виде не храним. Файл passwd собираем через mosquitto_passwd в контейнере:
# Запускаем временный контейнер для генерации пароля (пользователь admin, пароль secret) docker run --rm -it -v $(pwd)/config:/mosquitto/config eclipse-mosquitto mosquitto_passwd -c -b /mosquitto/config/passwd admin secret
Совет: Если нужно добавить еще одного пользователя без перезаписи первого, выполните ту же команду, но уберите флаг
-c.
На Ubuntu здесь часто спотыкаются о правах. Mosquitto в контейнере работает от UID 1883. Если passwd принадлежит другому пользователю, в логе появится Unable to open passwordfile. Исправляем это на хосте:
sudo chown -R 1883:1883 ./config sudo chmod 600 ./config/passwd sudo chmod 700 ./config/acl sudo chmod 755 ./config
Брокер настроен. Осталось описать контейнер Mosquitto в docker-compose.yml:
version: '3.8' services: mosquitto: image: eclipse-mosquitto:latest container_name: mosquitto restart: unless-stopped ports: - "1883:1883" - "9001:9001" volumes: - ./config:/mosquitto/config:ro # Монтируем конфигурацию только для чтения - ./data:/mosquitto/data
Запускаем контейнер:
docker compose up -d
Проверяем, что брокер отвечает
Проверяем соединение с хоста. Анонимного клиента брокер должен отвергнуть, а авторизованный увидит тестовое сообщение:
sudo apt-get update && sudo apt-get install -y mosquitto-clients # В первом терминале запускаем подписку: mosquitto_sub -h localhost -t "sensors/#" -u "admin" -P "secret" -v # Во втором терминале отправляем тестовое значение: mosquitto_pub -h localhost -t "sensors/esp8266/temperature" -m "24.5" -u "admin" -P "secret"
Появилось сообщение в первом терминале? Значит, MQTT работает.
Шаг 3. Записываем MQTT-показания через Telegraf
Я отправлял сообщения MQTT в Postgres своим скриптом на Python. После третьего обрыва связи в тестах потерялась большая часть значений. Поэтому для этой задачи взял Telegraf от InfluxData: он читает MQTT и записывает данные в PostgreSQL. Если база ненадолго недоступна, Telegraf продолжает собирать сообщения в буфер.
Telegraf ставлю на ту же Ubuntu-машину, где работают остальные компоненты. Подключаю репозиторий и устанавливаю пакет.
# Очищаем старые ключи, если они были sudo rm -f /etc/apt/sources.list.d/influxdata.list sudo rm -f /etc/apt/trusted.gpg.d/influxdata-archive* # Добавляем официальный репозиторий wget -q https://repos.influxdata.com/influxdata-archive.key cat influxdata-archive.key | gpg --dearmor | sudo tee /etc/apt/trusted.gpg.d/influxdata-archive.gpg > /dev/null echo 'deb [signed-by=/etc/apt/trusted.gpg.d/influxdata-archive.gpg] https://repos.influxdata.com/debian stable main' | sudo tee /etc/apt/sources.list.d/influxdata.list # Устанавливаем sudo apt-get update && sudo apt-get install telegraf -y
В /etc/telegraf/telegraf.conf нужны два входа mqtt_consumer, по одному на каждый датчик, и один выход в PostgreSQL:
[agent] interval = "10s" round_interval = true flush_interval = "10s" [[inputs.mqtt_consumer]] servers = ["tcp://127.0.0.1:1883"] topics = ["sensors/esp8266/temperature"] data_format = "value" data_type = "float" username = "admin" password = "secret" client_id = "telegraf_t1" persistent_session = false [[inputs.mqtt_consumer]] servers = ["tcp://127.0.0.1:1883"] topics = ["sensors/esp8266/humidity"] data_format = "value" data_type = "float" username = "admin" password = "secret" client_id = "telegraf_t2" persistent_session = false [[outputs.postgresql]] connection = "host=localhost port=5433 user=postgres password=super_password_123 dbname=sensor_data sslmode=disable"
Проверяю конфигурацию:
sudo telegraf --config /etc/telegraf/telegraf.conf --test
Если Telegraf не нашёл ошибок, перезапускаю службу:
sudo systemctl restart telegraf
Шаг 4. Даём LLM доступ к данным
Показания уже сохраняются в базе. Следующий слой, доступ для LLM. MCP-сервер будет отдавать модели только инструмент чтения. Пишу его на Go: после компиляции на шлюз можно отправить статический бинарник без отдельного рантайма. mcp-brain будет работать рядом с Postgres.
Проект:
mkdir mcp-timescale-server && cd mcp-timescale-server go mod init mcp-timescale-server go get github.com/modelcontextprotocol/go-sdk go get github.com/jackc/pgx/v5
В main.go будет один инструмент: запрос к таблице с показаниями. Сейчас я проверяю только начало строки и пропускаю запросы, начинающиеся с SELECT. Для производства этого недостаточно. Я бы подключал отдельного пользователя Postgres с правами только на чтение sensors.mqtt_consumer. Иначе неудачно сформированный запрос даст модели слишком широкие права на базу.
package main import ( "context" "fmt" "log" "os" "strings" "github.com/jackc/pgx/v5" "github.com/modelcontextprotocol/go-sdk/mcp" ) type QueryInput struct { SqlQuery string `json:"sql_query" jsonschema:"Полный SQL запрос SELECT для чтения данных из TimescaleDB. Запрос должен быть безопасным и использовать только SELECT."` } type QueryOutput struct { Result string `json:"result" jsonschema:"Результат выполнения запроса"` } func ExecuteQuery(ctx context.Context, request *mcp.CallToolRequest, input QueryInput) (*mcp.CallToolResult, QueryOutput, error) { // Безопасность: грубая, но полезная валидация на уровне кода cleanQuery := strings.TrimSpace(strings.ToLower(input.SqlQuery)) if !strings.HasPrefix(cleanQuery, "select") { return nil, QueryOutput{}, fmt.Errorf("разрешены только запросы на чтение данных (SELECT)") } connStr := os.Getenv("TIMESCALE_URL") if connStr == "" { connStr = "postgres://postgres:super_password_123@localhost:5433/sensor_data?sslmode=disable" } conn, err := pgx.Connect(ctx, connStr) if err != nil { return nil, QueryOutput{}, fmt.Errorf("ошибка подключения к БД: %w", err) } defer conn.Close(ctx) rows, err := conn.Query(ctx, input.SqlQuery) if err != nil { return nil, QueryOutput{}, fmt.Errorf("ошибка выполнения SQL: %w", err) } defer rows.Close() var resultText string for rows.Next() { values, err := rows.Values() if err != nil { continue } resultText += fmt.Sprintf("%v\n", values) } if resultText == "" { resultText = "Запрос выполнен успешно, строк не возвращено." } return nil, QueryOutput{Result: resultText}, nil } func main() { server := mcp.NewServer(&mcp.Implementation{ Name: "timescale-mcp-brain", Version: "v1.0.0", }, nil) mcp.AddTool(server, &mcp.Tool{ Name: "query_sensor_database", Description: "Выполняет SQL SELECT запросы к базе данных TimescaleDB с данными датчиков температуры и влажности. Используй для аналитики, агрегации или получения последних значений. Таблица: sensors.mqtt_consumer (поля: time, host, topic, value).", }, ExecuteQuery) log.Println("MCP Brain server запущен на STDIO транспорте") if err := server.Run(context.Background(), &mcp.StdioTransport{}); err != nil { log.Fatal(err) } }
Собираем бинарник:
go build -o mcp-brain main.go
Шаг 5. Подключаем OpenCode к локальному серверу
Сервер общается через stdio. В OpenCode достаточно указать путь к бинарнику. Конфиг обычно лежит в ~/.config/opencode/opencode.json (на Windows C:\Users\%USER%\.config\opencode\opencode.jsonc). В него добавляю такой блок:
{ "mcpServers": { "timescale-brain": { "type": "local", "command": "/home/user/mcp-timescale-server/mcp-brain", "enabled": true, "timeout": 30000, "env": { "TIMESCALE_URL": "postgres://postgres:super_password_123@localhost:5433/sensor_data?sslmode=disable" } } } }
(Путь /home/user/... замените на путь к своему бинарнику.)
После перезапуска OpenCode у timescale-brain появляется зелёная точка. Значит, клиент увидел сервер и может отправлять ему запросы. Публичный сервер здесь не нужен: агент работает на станции инженера, Postgres остаётся на локальном шлюзе.
Что получилось и что ещё проверить
Эту схему я гонял на Ubuntu-шлюзе у заказчика, рядом с базой и без облака. MCP не отменяет проверку: модель может взять не то окно по времени или неверно прочитать корреляцию. Но вопрос «что было с датчиками за три дня?» уже не требует отдельного запроса, выгрузки и графика.
Архитектура пока простая. Следующим шагом я бы добавил интеграцию со SCADA, отдельного read-only пользователя в Postgres и аудит запросов. Тогда будут понятны три вещи: откуда брать дополнительные данные, что разрешено модели и кто запускал запрос.
MQTT здесь только для примера. У Telegraf есть inputs для Modbus и OPC UA. Транспорт можно заменить, не переделывая всю цепочку.
Если у вас похожая задача в реальном секторе, пишите. Обсудим, где в этой цепочке лучше подключить LLM.
