┌─────────────┐      ┌─────────────┐
│   Telegram  │─────▶│     n8n     │
│     User    │      │  Workflows  │
└─────────────┘      └─────────────┘
                           │
                           ├─▶ HTTP Request (MCP Server)
                           │
                           ├─▶ Database
                           │
                           └─▶ External APIs
    
    Когда нужен AI:
    ┌─────────────┐      ┌─────────────┐      ┌─────────────┐
    │     n8n     │─────▶│    MCP      │─────▶│   AI Model  │
    │  Workflow   │      │   Server    │      │  (Claude)   │
    └─────────────┘      └─────────────┘      └─────────────┘

# Запуск n8n через Docker
docker run -it --rm \
  --name n8n \
  -p 5678:5678 \
  -v ~/.n8n:/home/node/.n8n \
  n8nio/n8n

{
  "credentials": {
    "telegramBotApi": {
      "id": "1",
      "name": "My Telegram Bot"
    }
  },
  "parameters": {
    "updates": [
      "message"
    ]
  }
}

Chat ID: {{ $json.message.chat.id }}
Message: {{ $json.message.text }}

Condition: {{ $json.message.text }} starts with "/start"

# mcp_server.py
import asyncio
import json
import sys
from typing import Any, Sequence
from mcp.server import Server
from mcp.server.stdio import stdio_server
from mcp.types import (
    Resource,
    Tool,
    Prompt,
    TextContent,
    ImageContent,
    EmbeddedResource,
)

# Создаем экземпляр сервера
app = Server("telegram-bot-mcp")

@app.list_resources()
async def list_resources() -> list[Resource]:
    """Список доступных ресурсов"""
    return [
        Resource(
            uri="telegram://messages",
            name="Telegram Messages",
            description="Доступ к сообщениям Telegram бота",
            mimeType="application/json",
        ),
        Resource(
            uri="telegram://users",
            name="Telegram Users",
            description="Информация о пользователях бота",
            mimeType="application/json",
        ),
    ]

@app.read_resource()
async def read_resource(uri: str) -> str:
    """Чтение ресурса по URI"""
    if uri == "telegram://messages":
        # Здесь можно получить сообщения из базы данных или API
        return json.dumps({
            "messages": [],
            "total": 0
        })
    elif uri == "telegram://users":
        return json.dumps({
            "users": [],
            "total": 0
        })
    else:
        raise ValueError(f"Unknown resource: {uri}")

@app.list_tools()
async def list_tools() -> list[Tool]:
    """Список доступных инструментов"""
    return [
        Tool(
            name="send_telegram_message",
            description="Отправить сообщение пользователю Telegram",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {
                        "type": "string",
                        "description": "ID пользователя Telegram"
                    },
                    "message": {
                        "type": "string",
                        "description": "Текст сообщения"
                    }
                },
                "required": ["user_id", "message"]
            }
        ),
        Tool(
            name="trigger_n8n_workflow",
            description="Запустить n8n workflow",
            inputSchema={
                "type": "object",
                "properties": {
                    "workflow_id": {
                        "type": "string",
                        "description": "ID workflow в n8n"
                    },
                    "data": {
                        "type": "object",
                        "description": "Данные для передачи в workflow"
                    }
                },
                "required": ["workflow_id"]
            }
        ),
        Tool(
            name="get_user_info",
            description="Получить информацию о пользователе",
            inputSchema={
                "type": "object",
                "properties": {
                    "user_id": {
                        "type": "string",
                        "description": "ID пользователя"
                    }
                },
                "required": ["user_id"]
            }
        ),
    ]

@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    """Выполнение инструмента"""
    if name == "send_telegram_message":
        user_id = arguments.get("user_id")
        message = arguments.get("message")
        
        # Здесь будет вызов Telegram API или очереди сообщений
        result = await send_message_to_telegram(user_id, message)
        
        return [
            TextContent(
                type="text",
                text=json.dumps({
                    "success": True,
                    "message_id": result.get("message_id"),
                    "sent_to": user_id
                })
            )
        ]
    
    elif name == "trigger_n8n_workflow":
        workflow_id = arguments.get("workflow_id")
        data = arguments.get("data", {})
        
        # Запуск n8n workflow
        result = await trigger_n8n_workflow(workflow_id, data)
        
        return [
            TextContent(
                type="text",
                text=json.dumps({
                    "success": True,
                    "workflow_id": workflow_id,
                    "execution_id": result.get("id")
                })
            )
        ]
    
    elif name == "get_user_info":
        user_id = arguments.get("user_id")
        user_info = await get_user_info(user_id)
        
        return [
            TextContent(
                type="text",
                text=json.dumps(user_info)
            )
        ]
    
    else:
        raise ValueError(f"Unknown tool: {name}")

# Вспомогательные функции
async def send_message_to_telegram(user_id: str, message: str) -> dict:
    """Отправка сообщения в Telegram"""
    # Интеграция с Telegram ботом
    # В реальном проекте здесь будет вызов API
    return {"message_id": "12345"}

async def trigger_n8n_workflow(workflow_id: str, data: dict) -> dict:
    """Запуск n8n workflow"""
    # Интеграция с n8n API
    return {"id": "exec_12345"}

async def get_user_info(user_id: str) -> dict:
    """Получение информации о пользователе"""
    return {
        "user_id": user_id,
        "username": "user_name",
        "status": "active"
    }

async def main():
    """Запуск MCP сервера через stdio"""
    async with stdio_server() as (read_stream, write_stream):
        await app.run(
            read_stream,
            write_stream,
            app.create_initialization_options()
        )

if __name__ == "__main__":
    asyncio.run(main())

# mcp_server_http.py
from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse
import uvicorn
from mcp_server import app, call_tool, list_tools, list_resources

http_app = FastAPI(title="MCP Server HTTP")

@http_app.post("/mcp")
async def handle_mcp_request(request: Request):
    """Обработка MCP запросов через HTTP"""
    data = await request.json()
    
    method = data.get("method")
    params = data.get("params", {})
    request_id = data.get("id")
    
    try:
        if method == "tools/list":
            tools = await list_tools()
            result = [tool.dict() for tool in tools]
        elif method == "tools/call":
            tool_name = params.get("name")
            arguments = params.get("arguments", {})
            result = await call_tool(tool_name, arguments)
            result = [item.dict() if hasattr(item, "dict") else item for item in result]
        elif method == "resources/list":
            resources = await list_resources()
            result = [resource.dict() for resource in resources]
        else:
            raise ValueError(f"Unknown method: {method}")
        
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": request_id,
            "result": result
        })
    except Exception as e:
        return JSONResponse({
            "jsonrpc": "2.0",
            "id": request_id,
            "error": {
                "code": -32000,
                "message": str(e)
            }
        })

if __name__ == "__main__":
    uvicorn.run(http_app, host="0.0.0.0", port=8000)

mcp-server/
├── mcp_server.py          # Основной MCP сервер
├── mcp_server_http.py     # HTTP транспорт для n8n
├── config.py              # Конфигурация
├── requirements.txt       # Зависимости
├── .env                   # Переменные окружения
└── Dockerfile            # Для развертывания

Method: POST
URL: http://your-mcp-server:8000/mcp
Headers:
  Content-Type: application/json
Body (JSON):
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "get_user_info",
    "arguments": {
      "user_id": "{{ $json.message.from.id }}"
    }
  }
}

Workflow структура:
Telegram Trigger
    ↓
IF ({{ $json.message.text }} starts with "/ask")
    ↓ (YES)
HTTP Request (MCP Server → tools/call)
    ↓
HTTP Request (AI API через MCP)
    ↓
Telegram (Send Message)

# n8n HTTP Request нода для MCP
Method: POST
URL: http://localhost:8000/mcp
Body:
{
  "jsonrpc": "2.0",
  "id": 1,
  "method": "tools/call",
  "params": {
    "name": "ask_ai",
    "arguments": {
      "question": "{{ $json.message.text }}",
      "context": {
        "user_id": "{{ $json.message.from.id }}",
        "chat_id": "{{ $json.message.chat.id }}"
      }
    }
  }
}

# telegram_bot.py
import asyncio
import os
import aiohttp
from telegram import Update
from telegram.ext import Application, CommandHandler, MessageHandler, filters, ContextTypes
from dotenv import load_dotenv

load_dotenv()

TELEGRAM_BOT_TOKEN = os.getenv("TELEGRAM_BOT_TOKEN")
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8000/mcp")

# Клиент для взаимодействия с MCP сервером
class MCPClient:
    def __init__(self, server_url: str):
        self.server_url = server_url
        self.session = None
    
    async def _get_session(self):
        if self.session is None:
            self.session = aiohttp.ClientSession()
        return self.session
    
    async def call_tool(self, tool_name: str, arguments: dict) -> dict:
        """Вызов инструмента MCP"""
        session = await self._get_session()
        
        request_data = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/call",
            "params": {
                "name": tool_name,
                "arguments": arguments
            }
        }
        
        async with session.post(self.server_url, json=request_data) as response:
            result = await response.json()
            if "error" in result:
                raise Exception(result["error"]["message"])
            return result.get("result", [])
    
    async def list_tools(self) -> list:
        """Получение списка доступных инструментов"""
        session = await self._get_session()
        
        request_data = {
            "jsonrpc": "2.0",
            "id": 1,
            "method": "tools/list",
            "params": {}
        }
        
        async with session.post(self.server_url, json=request_data) as response:
            result = await response.json()
            if "error" in result:
                raise Exception(result["error"]["message"])
            return result.get("result", [])

mcp_client = MCPClient(MCP_SERVER_URL)

async def start(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик команды /start"""
    await update.message.reply_text(
        "🤖 Привет! Я бот с поддержкой MCP.\n\n"
        "Доступные команды:\n"
        "/help - список команд\n"
        "/tools - доступные инструменты\n"
        "/send [user_id] [message] - отправить сообщение\n"
        "/n8n [workflow_id] [data] - запустить n8n workflow\n"
        "/info [user_id] - информация о пользователе"
    )

async def help_command(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Обработчик команды /help"""
    tools = await mcp_client.list_tools()
    tools_list = "\n".join([f"• {tool['name']}: {tool['description']}" for tool in tools])
    
    await update.message.reply_text(
        f"📋 Доступные инструменты MCP:\n\n{tools_list}"
    )

async def send_message(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Отправка сообщения через MCP"""
    args = context.args
    if len(args) < 2:
        await update.message.reply_text(
            "Использование: /send [user_id] [message]"
        )
        return
    
    user_id = args[0]
    message = " ".join(args[1:])
    
    try:
        result = await mcp_client.call_tool(
            "send_telegram_message",
            {"user_id": user_id, "message": message}
        )
        
        await update.message.reply_text(
            f"✅ Сообщение отправлено!\n{result[0]['text']}"
        )
    except Exception as e:
        await update.message.reply_text(f"❌ Ошибка: {str(e)}")

async def trigger_n8n(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Запуск n8n workflow через MCP"""
    args = context.args
    if len(args) < 1:
        await update.message.reply_text(
            "Использование: /n8n [workflow_id] [json_data]"
        )
        return
    
    workflow_id = args[0]
    data = {}
    
    if len(args) > 1:
        import json
        try:
            data = json.loads(" ".join(args[1:]))
        except:
            data = {"message": " ".join(args[1:])}
    
    try:
        result = await mcp_client.call_tool(
            "trigger_n8n_workflow",
            {"workflow_id": workflow_id, "data": data}
        )
        
        await update.message.reply_text(
            f"✅ Workflow запущен!\n{result[0]['text']}"
        )
    except Exception as e:
        await update.message.reply_text(f"❌ Ошибка: {str(e)}")

async def get_info(update: Update, context: ContextTypes.DEFAULT_TYPE):
    """Получение информации о пользователе"""
    args = context.args
    if len(args) < 1:
        user_id = str(update.message.from_user.id)
    else:
        user_id = args[0]
    
    try:
        result = await mcp_client.call_tool(
            "get_user_info",
            {"user_id": user_id}
        )
        
        import json
        user_data = json.loads(result[0]['text'])
        
        await update.message.reply_text(
            f"👤 Информация о пользователе:\n\n"
            f"ID: {user_data.get('user_id')}\n"
            f"Username: {user_data.get('username')}\n"
            f"Status: {user_data.get('status')}"
        )
    except Exception as e:
        await update.message.reply_text(f"❌ Ошибка: {str(e)}")

def main():
    """Запуск Telegram бота"""
    application = Application.builder().token(TELEGRAM_BOT_TOKEN).build()
    
    # Регистрация обработчиков
    application.add_handler(CommandHandler("start", start))
    application.add_handler(CommandHandler("help", help_command))
    application.add_handler(CommandHandler("tools", help_command))
    application.add_handler(CommandHandler("send", send_message))
    application.add_handler(CommandHandler("n8n", trigger_n8n))
    application.add_handler(CommandHandler("info", get_info))
    
    # Запуск бота
    application.run_polling(allowed_updates=Update.ALL_TYPES)

if __name__ == "__main__":
    main()

Telegram Trigger
    ↓
IF ({{ $json.message.text }} = "/start")
    ↓
Telegram (Send Message: "Привет! Доступные команды: /help")
    
IF ({{ $json.message.text }} = "/help")
    ↓
Telegram (Send Message: "Команды: /start, /help, /info")

Telegram Trigger
    ↓
IF ({{ $json.message.text }} starts with "/ask")
    ↓
HTTP Request (MCP Server)
    Body: {
      "method": "tools/call",
      "params": {
        "name": "ask_ai",
        "arguments": {
          "question": "{{ $json.message.text }}"
        }
      }
    }
    ↓
Telegram (Send Message: {{ $json.result }})

Telegram Trigger
    ↓
Switch Node
    ├─ "погода" → HTTP Request (Weather API)
    ├─ "/ask" → HTTP Request (MCP → AI)
    ├─ "заказ" → Database (Save order)
    └─ default → HTTP Request (MCP → AI для общего ответа)
    ↓
Telegram (Send Message)

# n8n_integration.py
import aiohttp
import os
from typing import Dict, Any

N8N_API_URL = os.getenv("N8N_API_URL", "http://localhost:5678/api/v1")
N8N_API_KEY = os.getenv("N8N_API_KEY", "")

class N8NIntegration:
    def __init__(self, api_url: str, api_key: str = ""):
        self.api_url = api_url
        self.api_key = api_key
        self.session = None
    
    async def _get_session(self):
        if self.session is None:
            headers = {}
            if self.api_key:
                headers["X-N8N-API-KEY"] = self.api_key
            self.session = aiohttp.ClientSession(headers=headers)
        return self.session
    
    async def trigger_workflow(self, workflow_id: str, data: Dict[str, Any]) -> Dict:
        """Запуск n8n workflow"""
        session = await self._get_session()
        
        url = f"{self.api_url}/workflows/{workflow_id}/execute"
        
        payload = {
            "data": data
        }
        
        async with session.post(url, json=payload) as response:
            if response.status == 200:
                return await response.json()
            else:
                error_text = await response.text()
                raise Exception(f"n8n API error: {error_text}")
    
    async def get_workflow_status(self, execution_id: str) -> Dict:
        """Получение статуса выполнения workflow"""
        session = await self._get_session()
        
        url = f"{self.api_url}/executions/{execution_id}"
        
        async with session.get(url) as response:
            if response.status == 200:
                return await response.json()
            else:
                raise Exception(f"Failed to get execution status: {response.status}")
    
    async def list_workflows(self) -> list:
        """Получение списка доступных workflows"""
        session = await self._get_session()
        
        url = f"{self.api_url}/workflows"
        
        async with session.get(url) as response:
            if response.status == 200:
                result = await response.json()
                return result.get("data", [])
            else:
                raise Exception(f"Failed to list workflows: {response.status}")

# Использование в mcp_server.py
n8n_client = N8NIntegration(N8N_API_URL, N8N_API_KEY)

async def trigger_n8n_workflow(workflow_id: str, data: dict) -> dict:
    """Обновленная функция для запуска n8n workflow"""
    result = await n8n_client.trigger_workflow(workflow_id, data)
    return {
        "id": result.get("id"),
        "status": result.get("status"),
        "started_at": result.get("startedAt")
    }

{
  "name": "Telegram Bot Notification Workflow",
  "nodes": [
    {
      "parameters": {
        "httpMethod": "POST",
        "path": "telegram-notification",
        "responseMode": "responseNode",
        "options": {}
      },
      "name": "Webhook",
      "type": "n8n-nodes-base.webhook",
      "position": [250, 300]
    },
    {
      "parameters": {
        "operation": "sendMessage",
        "chatId": "={{ $json.data.chat_id }}",
        "text": "={{ $json.data.message }}",
        "additionalFields": {}
      },
      "name": "Send Telegram Message",
      "type": "n8n-nodes-base.telegram",
      "position": [450, 300]
    },
    {
      "parameters": {
        "respondWith": "json",
        "responseBody": "={{ { \"success\": true, \"message_id\": $json.message_id } }}"
      },
      "name": "Respond to Webhook",
      "type": "n8n-nodes-base.respondToWebhook",
      "position": [650, 300]
    }
  ],
  "connections": {
    "Webhook": {
      "main": [
        [
          {
            "node": "Send Telegram Message",
            "type": "main",
            "index": 0
          }
        ]
      ]
    },
    "Send Telegram Message": {
      "main": [
        [
          {
            "node": "Respond to Webhook",
            "type": "main",
            "index": 0
          }
        ]
      ]
    }
  }
}

# Dockerfile
FROM python:3.11-slim

WORKDIR /app

# Установка зависимостей
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

# Копирование кода
COPY . .

# Переменные окружения
ENV PYTHONUNBUFFERED=1
ENV MCP_SERVER_HOST=0.0.0.0
ENV MCP_SERVER_PORT=8000

# Запуск MCP сервера и Telegram бота
CMD ["sh", "-c", "python mcp_server_http.py & python telegram_bot.py"]

mcp>=0.1.0
python-telegram-bot>=20.0
aiohttp>=3.9.0
fastapi>=0.104.0
uvicorn>=0.24.0
python-dotenv>=1.0.0

TELEGRAM_BOT_TOKEN=your_bot_token_here
MCP_SERVER_URL=http://localhost:8000/mcp
N8N_API_URL=http://your-n8n-instance:5678/api/v1
N8N_API_KEY=your_n8n_api_key

# В MCP сервере добавьте инструмент
Tool(
    name="send_notification",
    description="Отправить уведомление через n8n workflow",
    inputSchema={
        "type": "object",
        "properties": {
            "user_id": {"type": "string"},
            "notification_type": {
                "type": "string",
                "enum": ["alert", "info", "warning", "success"]
            },
            "message": {"type": "string"}
        }
    }
)

@app.read_resource()
async def read_resource(uri: str) -> str:
    if uri.startswith("db://users/"):
        user_id = uri.split("/")[-1]
        user_data = await get_user_from_db(user_id)
        return json.dumps(user_data)
    # ... остальные ресурсы

Tool(
    name="get_weather",
    description="Получить прогноз погоды",
    inputSchema={
        "type": "object",
        "properties": {
            "city": {"type": "string"}
        }
    }
)

async def get_weather(city: str) -> dict:
    # Вызов внешнего API или n8n workflow
    workflow_result = await n8n_client.trigger_workflow(
        "weather-workflow-id",
        {"city": city}
    )
    return workflow_result

# Добавьте middleware для проверки токена
@app.middleware("http")
async def verify_token(request: Request, call_next):
    token = request.headers.get("Authorization")
    if not verify_api_token(token):
        return JSONResponse({"error": "Unauthorized"}, status_code=401)
    return await call_next(request)

# Добавьте метрики в MCP сервер
from prometheus_client import Counter, Histogram

tool_calls = Counter('mcp_tool_calls_total', 'Total tool calls', ['tool_name'])
tool_duration = Histogram('mcp_tool_duration_seconds', 'Tool execution duration', ['tool_name'])

@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    with tool_duration.labels(tool_name=name).time():
        tool_calls.labels(tool_name=name).inc()
        # ... выполнение инструмента

@app.list_prompts()
async def list_prompts() -> list[Prompt]:
    return [
        Prompt(
            name="telegram_response",
            description="Шаблон для ответа пользователю Telegram",
            arguments=[
                {
                    "name": "user_message",
                    "description": "Сообщение от пользователя",
                    "required": True
                }
            ]
        )
    ]

@app.get_prompt()
async def get_prompt(name: str, arguments: dict) -> str:
    if name == "telegram_response":
        user_message = arguments.get("user_message")
        return f"""Вы - помощник Telegram бота.
Пользователь написал: {user_message}
Дайте дружелюбный и полезный ответ."""

@app.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]) -> list[TextContent]:
    if name == "long_running_task":
        # Возвращаем промежуточные результаты через streaming
        async for chunk in process_long_task():
            yield TextContent(type="text", text=chunk)