ИИ-ассистент для мониторинга n8n с Claude и MCP

В производственной среде рабочих процессов n8n знакомо напряжение от уведомления о сбое и необходимости разбираться в логах для выявления причины.

Пользователь: Автоматизация больше не функционирует, уведомление не пришло!

Первоначальный этап предполагает открытие интерфейса n8n и просмотр недавних выполнений для определения проблем.

Через несколько минут анализ переходит к переключению между выполнениями, сопоставлению временных меток и изучению ошибок в формате JSON для понимания места поломки.

Представьте, если агент способен объяснить причину сбоя рабочего процесса в 3 часа ночи, избавив от необходимости копаться в логах?

Это реально достижимо!

В рамках эксперимента API n8n, обеспечивающее доступ к логам выполнений экземпляра, было интегрировано с сервером MCP на базе Claude.

Итогом стал ИИ-ассистент, способный отслеживать рабочие процессы, анализировать сбои и описывать проблемы естественным языком.

В этой статье описывается пошаговый процесс создания такой системы.

Первая часть демонстрирует реальный случай из экземпляра n8n, где несколько рабочих процессов завершились сбоем ночью.

Этот пример иллюстрирует, как агент выявляет проблемы и объясняет их коренные причины.

Далее подробно рассказывается о подключении API экземпляра n8n к серверу MCP через веб-хук, позволяющий Claude Desktop получать данные о выполнениях для отладки на естественном языке.

Веб-хук охватывает три функции:

• Получить активные рабочие процессы: предоставляет перечень всех активных рабочих процессов
• Получить последние выполнения: содержит данные о последних n выполнениях
• Получить детали выполнений (статус = Ошибка): сведения о неудачных выполнениях, оформленные для поддержки анализа корневых причин

Демонстрация: Использование ИИ для анализа неудачных выполнений n8n

Рассмотрим один из экземпляров n8n, где запущено несколько рабочих процессов для получения информации о событиях из различных городов мира.

Эти рабочие процессы помогают бизнес- и сетевым сообществам находить интересные мероприятия для посещения и обучения.

Для проверки решения начнем с запроса к агенту о перечне активных рабочих процессов.

Шаг 1: Сколько активных рабочих процессов?

На основе поставленного вопроса Claude распознал необходимость взаимодействия с инструментом n8n-monitor, созданным на базе сервера MCP.

Далее модель самостоятельно выбрала подходящую функцию Получить активные рабочие процессы для извлечения списка активных автоматизаций из экземпляра n8n.

Здесь проявляется мощь модели.

Она самостоятельно классифицировала рабочие процессы по их названиям:

• 8 рабочих процессов для подключения к API событий и их обработки
• 3 других рабочих процесса, находящихся в разработке, включая тот, что используется для получения логов

Это начало анализа; все эти выводы будут применены в исследовании корневых причин.

Шаг 2: Анализ последних n выполнений

На этом этапе можно запросить у Claude извлечение недавних выполнений для разбора.

Благодаря контексту в описаниях функций, которые будут рассмотрены в следующей части, Claude понял, что требуется вызов функции получения выполнений рабочих процессов.

В ответ поступит сводка выполнений с процентом сбоев и количеством затронутых рабочих процессов.

{ "summary": { "totalExecutions": 25, "successfulExecutions": 22, "failedExecutions": 3, "failureRate": "12.00%", "successRate": "88.00%", "totalWorkflowsExecuted": 7, "workflowsWithFailures": 1 }, "executionModes": { "webhook": 7, "trigger": 18 }, "timing": { "averageExecutionTime": "15.75 seconds", "maxExecutionTime": "107.18 seconds", "minExecutionTime": "0.08 seconds", "timeRange": { "from": "2025-10-24T06:14:23.127Z", "to": "2025-10-24T11:11:49.890Z" } }, [...]}

Это первое, что будет представлено; оно дает четкий обзор ситуации.

Во второй части вывода содержится детальный разбор сбоев для каждого затронутого рабочего процесса.

 "failureAnalysis": { "workflowsImpactedByFailures": [ "7uvA2XQPMB5l4kI5" ], "failedExecutionsByWorkflow": { "7uvA2XQPMB5l4kI5": { "workflowId": "7uvA2XQPMB5l4kI5", "failures": [ { "id": "13691", "startedAt": "2025-10-24T11:00:15.072Z", "stoppedAt": "2025-10-24T11:00:15.508Z", "mode": "trigger" }, { "id": "13683", "startedAt": "2025-10-24T09:00:57.274Z", "stoppedAt": "2025-10-24T09:00:57.979Z", "mode": "trigger" }, { "id": "13677", "startedAt": "2025-10-24T07:00:57.167Z", "stoppedAt": "2025-10-24T07:00:57.685Z", "mode": "trigger" } ], "failureCount": 3 } }, "recentFailures": [ { "id": "13691", "workflowId": "7uvA2XQPMB5l4kI5", "startedAt": "2025-10-24T11:00:15.072Z", "mode": "trigger" }, { "id": "13683", "workflowId": "7uvA2XQPMB5l4kI5", "startedAt": "2025-10-24T09:00:57.274Z", "mode": "trigger" }, { "id": "13677", "workflowId": "7uvA2XQPMB5l4kI5", "startedAt": "2025-10-24T07:00:57.167Z", "mode": "trigger" } ] },

Теперь пользователь получает обзор затронутых рабочих процессов вместе с деталями случаев сбоев.

В этом конкретном случае рабочий процесс “Bangkok Meetup” запускается ежечасно.

Из анализа видно, что проблемы возникли трижды (из пяти) за последние пять часов.

Примечание: Последнее предложение можно игнорировать, поскольку агент пока не имеет доступа к деталям выполнений.

Заключительная часть вывода включает анализ общей производительности рабочих процессов.

 "workflowPerformance": { "allWorkflowMetrics": { "CGvCrnUyGHgB7fi8": { "workflowId": "CGvCrnUyGHgB7fi8", "totalExecutions": 7, "successfulExecutions": 7, "failedExecutions": 0, "successRate": "100.00%", "failureRate": "0.00%", "lastExecution": "2025-10-24T11:11:49.890Z", "executionModes": { "webhook": 7 } }, [... другие рабочие процессы ...] , "topProblematicWorkflows": [ { "workflowId": "7uvA2XQPMB5l4kI5", "totalExecutions": 5, "successfulExecutions": 2, "failedExecutions": 3, "successRate": "40.00%", "failureRate": "60.00%", "lastExecution": "2025-10-24T11:00:15.072Z", "executionModes": { "trigger": 5 } }, { "workflowId": "CGvCrnUyGHgB7fi8", "totalExecutions": 7, "successfulExecutions": 7, "failedExecutions": 0, "successRate": "100.00%", "failureRate": "0.00%", "lastExecution": "2025-10-24T11:11:49.890Z", "executionModes": { "webhook": 7 } }, [... другие рабочие процессы ...] } ] }

Такой детальный разбор помогает расставить приоритеты в обслуживании, если сбои затрагивают несколько рабочих процессов.

В данном примере единственный сбоивший рабочий процесс – Ⓜ️ Bangkok Meetup.

Что, если требуется узнать, когда начались проблемы?

Не волнуйтесь, добавлен раздел с деталями выполнений по часам.

 "timeSeriesData": { "2025-10-24T11:00": { "total": 5, "success": 4, "error": 1 }, "2025-10-24T10:00": { "total": 6, "success": 6, "error": 0 }, "2025-10-24T09:00": { "total": 3, "success": 2, "error": 1 }, "2025-10-24T08:00": { "total": 3, "success": 3, "error": 0 }, "2025-10-24T07:00": { "total": 3, "success": 2, "error": 1 }, "2025-10-24T06:00": { "total": 5, "success": 5, "error": 0 } }

Достаточно позволить Claude сгенерировать удобную визуализацию, подобную приведенной ниже.

Важно отметить, что никаких указаний по представлению результатов не предоставлялось Claude; все это инициатива модели!

Впечатляюще, не так ли?

Шаг 3: Анализ корневой причины

Теперь, зная, какие рабочие процессы имеют проблемы, следует искать коренные причины.

Claude обычно вызывает функцию Получить неудачные выполнения для извлечения деталей сбоев.

Для справки, сбой в этом рабочем процессе вызван ошибкой в узле JSON Tech, обрабатывающем вывод вызова API.

• Meetup Tech отправляет HTTP-запрос к API Meetup
• Обработка в узле Result Tech
• JSON Tech преобразует вывод в структурированный JSON

Вот что происходит при нормальном выполнении.

Однако иногда вызов API завершается неудачей, и узел JavaScript получает ошибку из-за несоответствия формата входных данных.

Примечание: Эта проблема была исправлена в производстве (узел кода стал более устойчивым), но сохранена для демонстрации.

Проверим, сможет ли Claude выявить коренную причину.

Вот вывод функции Получить неудачные выполнения.

{ "workflow_id": "7uvA2XQPMB5l4kI5", "workflow_name": "Ⓜ️ Bangkok Meetup", "error_count": 5, "errors": [ { "id": "13691", "workflow_name": "Ⓜ️ Bangkok Meetup", "status": "error", "mode": "trigger", "started_at": "2025-10-24T11:00:15.072Z", "stopped_at": "2025-10-24T11:00:15.508Z", "duration_seconds": 0.436, "finished": false, "retry_of": null, "retry_success_id": null, "error": { "message": "A 'json' property isn't an object [item 0]", "description": "In the returned data, every key named 'json' must point to an object.", "http_code": null, "level": "error", "timestamp": null }, "failed_node": { "name": "JSON Tech", "type": "n8n-nodes-base.code", "id": "dc46a767-55c8-48a1-a078-3d401ea6f43e", "position": [ -768, -1232 ] }, "trigger": {} }, [... 4 другие ошибки ...] ], "summary": { "total_errors": 5, "error_patterns": { "A 'json' property isn't an object [item 0]": { "count": 5, "executions": [ "13691", "13683", "13677", "13660", "13654" ] } }, "failed_nodes": { "JSON Tech": 5 }, "time_range": { "oldest": "2025-10-24T05:00:57.105Z", "newest": "2025-10-24T11:00:15.072Z" } } }

Claude теперь имеет доступ к деталям выполнений с сообщениями об ошибках и затронутыми узлами.

В приведенном ответе видно, что Claude обобщил выводы нескольких выполнений в единый анализ.

Теперь известно, что:

• Ошибки возникали ежечасно, кроме 08:00
• Каждый раз затрагивался один и тот же узел “JSON Tech”
• Ошибка проявляется вскоре после запуска рабочего процесса

Этот описательный анализ дополняется началом диагностики.

Это утверждение верно, как подтверждает сообщение об ошибке в интерфейсе n8n.

Однако из-за ограниченного контекста Claude предлагает рекомендации по исправлению, которые не полностью точны.

Помимо корректировки кода, предоставляется план действий.

Поскольку проблема не только в узле кода, было направлено указание Claude углубить анализ корневой причины.

В итоге модель пересмотрела первоначальное предложение и начала формулировать предположения о коренных причинах.

Это приближается к реальной коренной причине, давая достаточно информации для начала исследования рабочего процесса.

Пересмотренное исправление теперь лучше, поскольку учитывает возможность проблемы в входных данных узла.

Это лучшее, чего можно ожидать от Claude при имеющейся ограниченной информации.

Заключение: Ценность этого инструмента

Этот простой эксперимент показывает, как ИИ-агент на базе Claude выходит за рамки базового мониторинга, предоставляя реальную операционную пользу.

До ручной проверки выполнений и логов можно сначала диалогизировать с системой автоматизации, чтобы узнать о сбоях, их причинах и получить контекстные объяснения за секунды.

Это не заменит специалиста полностью, но ускорит процесс анализа корневых причин.

В следующей части кратко вводится настройка сервера MCP для подключения Claude Desktop к экземпляру.

Создание локального сервера MCP для подключения Claude Desktop к микросервису FastAPI

Для обеспечения Claude тремя функциями веб-хука (Получить активные рабочие процессы, Получить выполнения рабочих процессов и Получить неудачные выполнения) реализован сервер MCP.

В этой части кратко описывается реализация, с акцентом на функции Получить активные рабочие процессы и Получить выполнения рабочих процессов, чтобы показать объяснение их использования Claude.

Создание класса для запросов к рабочему процессу

Прежде чем перейти к настройке трех инструментов, представлен вспомогательный класс, содержащий все необходимые функции для взаимодействия с веб-хуком.

Он находится в файле Python: ./utils/n8n_monitory_sync.py

import logging import os from datetime import datetime, timedelta from typing import Any, Dict, Optional import requests import traceback logger = logging.getLogger(__name__) class N8nMonitor: """Handler for n8n monitoring operations - synchronous version""" def __init__(self): self.webhook_url = os.getenv("N8N_WEBHOOK_URL", "") self.timeout = 30

Суть в получении URL веб-хука из переменной окружения и установке таймаута запроса в 30 секунд.

Первая функция get_active_workflows запрашивает веб-хук с параметром: "action": get_active_workflows".

def get_active_workflows(self) -> Dict[str, Any]: """Fetch all active workflows from n8n""" if not self.webhook_url: logger.error("Environment variable N8N_WEBHOOK_URL not configured") return {"error": "N8N_WEBHOOK_URL environment variable not set"} try: logger.info("Fetching active workflows from n8n") response = requests.post( self.webhook_url, json={"action": "get_active_workflows"}, timeout=self.timeout ) response.raise_for_status() data = response.json() logger.debug(f"Response type: {type(data)}") # List of all workflows workflows = [] if isinstance(data, list): workflows = [item for item in data if isinstance(item, dict)] if not workflows and data: logger.error(f"Expected list of dictionaries, got list of {type(data[0]).__name__}") return {"error": "Webhook returned invalid data format"} elif isinstance(data, dict): if "data" in data: workflows = data["data"] else: logger.error(f"Unexpected dict response with keys: {list(data.keys())} 
 {traceback.format_exc()}") return {"error": "Unexpected response format"} else: logger.error(f"Unexpected response type: {type(data)} 
 {traceback.format_exc()}") return {"error": f"Unexpected response type: {type(data).__name__}"} logger.info(f"Successfully fetched {len(workflows)} active workflows") return { "total_active": len(workflows), "workflows": [ { "id": wf.get("id", "unknown"), "name": wf.get("name", "Unnamed"), "created": wf.get("createdAt", ""), "updated": wf.get("updatedAt", ""), "archived": wf.get("isArchived", "false") == "true" } for wf in workflows ], "summary": { "total": len(workflows), "names": [wf.get("name", "Unnamed") for wf in workflows] } } except requests.exceptions.RequestException as e: logger.error(f"Error fetching workflows: {e} 
 {traceback.format_exc()}") return {"error": f"Failed to fetch workflows: {str(e)} 
 {traceback.format_exc()}"} except Exception as e: logger.error(f"Unexpected error fetching workflows: {e} 
 {traceback.format_exc()}") return {"error": f"Unexpected error: {str(e)} 
 {traceback.format_exc()}"}

Добавлено множество проверок, поскольку API иногда возвращает неожиданный формат данных.

Это решение более надежно, предоставляя Claude полную информацию о причинах неудачи запроса.

Теперь, когда первая функция описана, перейдем к получению последних n выполнений с помощью get_workflow_executions.

def get_workflow_executions( self, limit: int = 50, includes_kpis: bool = False, ) -> Dict[str, Any]: """Fetch workflow executions of the last 'limit' executions with or without KPIs """ if not self.webhook_url: logger.error("Environment variable N8N_WEBHOOK_URL not set") return {"error": "N8N_WEBHOOK_URL environment variable not set"} try: logger.info(f"Fetching the last {limit} executions") payload = { "action": "get_workflow_executions", "limit": limit } response = requests.post( self.webhook_url, json=payload, timeout=self.timeout ) response.raise_for_status() data = response.json() if isinstance(data, list) and len(data) > 0: data = data[0] logger.info("Successfully fetched execution data") if includes_kpis and isinstance(data, dict): logger.info("Including KPIs in the execution data") if "summary" in data: summary = data["summary"] failure_rate = float(summary.get("failureRate", "0").rstrip("%")) data["insights"] = { "health_status": "🟢 Healthy" if failure_rate < 10 else "🟡 Warning" if failure_rate < 25 else "🔴 Critical", "message": f"{summary.get('totalExecutions', 0)} executions with {summary.get('failureRate', '0%')} failure rate" } return data except requests.exceptions.RequestException as e: logger.error(f"HTTP error fetching executions: {e} 
 {traceback.format_exc()}") return {"error": f"Failed to fetch executions: {str(e)}"} except Exception as e: logger.error(f"Unexpected error fetching executions: {e} 
 {traceback.format_exc()}") return {"error": f"Unexpected error: {str(e)}"}

Единственный параметр – количество n выполнений для извлечения: "limit": n.

Выводы включают сводку со статусом здоровья, генерируемым узлом кода Processing Audit.

Функция get_workflow_executions извлекает выводы для форматирования перед отправкой агенту.

Теперь, определив основные функции, можно создать инструменты для оснащения Claude через сервер MCP.

Настройка сервера MCP с инструментами

Теперь пора создать сервер MCP с инструментами и ресурсами для оснащения и обучения Claude.

from mcp.server.fastmcp import FastMCP import logging from typing import Optional, Dict, Any from utils.n8n_monitor_sync import N8nMonitor logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler("n8n_monitor.log"), logging.StreamHandler() ] ) logger = logging.getLogger(__name__) mcp = FastMCP("n8n-monitor") monitor = N8nMonitor()

Это базовая реализация с использованием FastMCP и импортом n8n_monitor_sync.py с функциями из предыдущей части.

# Resource for the agent (Samir: update it each time you add a tool) @mcp.resource("n8n://help") def get_help() -> str: """Get help documentation for the n8n monitoring tools""" return """ 📊 N8N MONITORING TOOLS ======================= WORKFLOW MONITORING: • get_active_workflows() List all active workflows with names and IDs EXECUTION TRACKING: • get_workflow_executions(limit=50, include_kpis=True) Get execution logs with detailed KPIs - limit: Number of recent executions to retrieve (1-100) - include_kpis: Calculate performance metrics ERROR DEBUGGING: • get_error_executions(workflow_id) Retrieve detailed error information for a specific workflow - Returns last 5 errors with comprehensive debugging data - Shows error messages, failed nodes, trigger data - Identifies error patterns and problematic nodes - Includes HTTP codes, error levels, and timing info HEALTH REPORTING: • get_workflow_health_report(limit=50) Generate comprehensive health analysis based on recent executions - Identifies problematic workflows - Shows success/failure rates - Provides execution timing metrics KEY METRICS PROVIDED: • Total executions • Success/failure rates • Execution times (avg, min, max) • Workflows with failures • Execution modes (manual, trigger, integrated) • Error patterns and frequencies • Failed node identification HEALTH STATUS INDICATORS: • 🟢 Healthy: <10% failure rate • 🟡 Warning: 10-25% failure rate • 🔴 Critical: >25% failure rate USAGE EXAMPLES: - "Show me all active workflows" - "What workflows have been failing?" - "Generate a health report for my n8n instance" - "Show execution metrics for the last 48 hours" - "Debug errors in workflow CGvCrnUyGHgB7fi8" - "What's causing failures in my data processing workflow?" DEBUGGING WORKFLOW: 1. Use get_workflow_executions() to identify problematic workflows 2. Use get_error_executions() for detailed error analysis 3. Check error patterns to identify recurring issues 4. Review failed node details and trigger data 5. Use workflow_id and execution_id for targeted fixes """

Поскольку инструмент сложен для понимания, включен промпт в форме ресурса MCP, суммирующий цели и возможности подключенного через веб-хук рабочего процесса n8n.

Теперь определим первый инструмент для получения всех активных рабочих процессов.

@mcp.tool() def get_active_workflows() -> Dict[str, Any]: """ Get all active workflows in the n8n instance. Returns: Dictionary with list of active workflows and their details """ try: logger.info("Fetching active workflows") result = monitor.get_active_workflows() if "error" in result: logger.error(f"Failed to get workflows: {result['error']}") else: logger.info(f"Found {result.get('total_active', 0)} active workflows") return result except Exception as e: logger.error(f"Unexpected error: {str(e)}") return {"error": str(e)}

Описание функции, объясняющее серверу MCP способ использования инструмента, относительно краткое, поскольку для get_active_workflows() нет входных параметров.

Аналогично поступим со вторым инструментом для извлечения последних n выполнений.

@mcp.tool() def get_workflow_executions( limit: int = 50, include_kpis: bool = True ) -> Dict[str, Any]: """ Get workflow execution logs and KPIs for the last N executions. Args: limit: Number of executions to retrieve (default: 50) include_kpis: Include calculated KPIs (default: true) Returns: Dictionary with execution data and KPIs """ try: logger.info(f"Fetching the last {limit} executions") result = monitor.get_workflow_executions( limit=limit, includes_kpis=include_kpis ) if "error" in result: logger.error(f"Failed to get executions: {result['error']}") else: if "summary" in result: summary = result["summary"] logger.info(f"Executions: {summary.get('totalExecutions', 0)}, " f"Failure rate: {summary.get('failureRate', 'N/A')}") return result except Exception as e: logger.error(f"Unexpected error: {str(e)}") return {"error": str(e)}

В отличие от предыдущего инструмента, здесь необходимо указать входные данные с значениями по умолчанию.

Теперь Claude оснащен этими двумя инструментами, которые применяются как в примере из предыдущей части.

Что дальше? Разверните на своей машине!

Чтобы сохранить краткость статьи, вводятся только эти два инструмента.

Включаются пошаговые объяснения по развертыванию на машине с детальным обзором исходного кода.

Заключение

Это только начало!

Это можно считать версией 1.0 супер-агента для управления рабочими процессами n8n.

Что подразумевается под этим?

Существует огромный потенциал для улучшения решения, особенно анализа корневых причин, путем:

• Предоставления агенту дополнительного контекста с помощью заметок в рабочих процессах
• Показ хороших входных и выходных данных с узлами оценки для помощи Claude в анализе разрывов
• Использования других конечных точек API n8n для более точного анализа

Это открытое решение, доступное для сообщества.