"""
Тестовый чат-бот для Bitrix24 с интеграцией MCP.
Этот бот:
1. Принимает сообщения от пользователей в Bitrix24
2. Использует MCP сервер для получения данных из CRM
3. Отвечает пользователю
Установка:
pip install flask requests
Запуск:
export BITRIX_WEBHOOK_URL="https://your-domain.bitrix24.ru/rest/1/secret/"
export MCP_SERVER_URL="http://localhost:8000/mcp"
python bitrix_bot.py
"""
import json
import os
from flask import Flask, request, jsonify
import requests
app = Flask(__name__)
# Конфигурация
BITRIX_WEBHOOK_URL = os.getenv("BITRIX_WEBHOOK_URL", "")
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8000/mcp")
BOT_NAME = "CRM Assistant"
def call_mcp_tool(tool_name: str, arguments: dict) -> dict:
"""Вызов инструмента MCP сервера."""
try:
# Формируем MCP запрос
mcp_request = {
"jsonrpc": "2.0",
"id": 1,
"method": "tools/call",
"params": {
"name": tool_name,
"arguments": arguments
}
}
response = requests.post(
MCP_SERVER_URL,
json=mcp_request,
headers={"Content-Type": "application/json"},
timeout=30
)
if response.status_code == 200:
result = response.json()
return result.get("result", {})
else:
return {"error": f"MCP error: {response.status_code}"}
except Exception as e:
return {"error": str(e)}
def send_bitrix_message(dialog_id: str, message: str):
"""Отправить сообщение в Bitrix24 чат."""
try:
url = f"{BITRIX_WEBHOOK_URL}im.message.add"
data = {
"DIALOG_ID": dialog_id,
"MESSAGE": message,
"SYSTEM": "N"
}
response = requests.post(url, json=data, timeout=10)
return response.json()
except Exception as e:
print(f"Error sending message: {e}")
return None
def process_command(text: str) -> str:
"""Обработать команду пользователя."""
text_lower = text.lower().strip()
# Команды для работы с контактами
if text_lower.startswith("/контакт ") or text_lower.startswith("/contact "):
# Поиск контакта
query = text.split(" ", 1)[1] if " " in text else ""
if query:
result = call_mcp_tool("search_contacts", {"query": query, "limit": 5})
if "error" in result:
return f"❌ Ошибка: {result['error']}"
return f"🔍 Результаты поиска:\n{json.dumps(result, ensure_ascii=False, indent=2)}"
return "Использование: /контакт <имя или телефон>"
# Команды для работы со сделками
elif text_lower.startswith("/сделки") or text_lower.startswith("/deals"):
result = call_mcp_tool("list_deals", {"active_only": True, "limit": 10})
if "error" in result:
return f"❌ Ошибка: {result['error']}"
return f"📊 Активные сделки:\n{json.dumps(result, ensure_ascii=False, indent=2)}"
elif text_lower.startswith("/сделка ") or text_lower.startswith("/deal "):
# Получить сделку по ID
try:
deal_id = int(text.split(" ")[1])
result = call_mcp_tool("get_deal", {"deal_id": deal_id})
if "error" in result:
return f"❌ Ошибка: {result['error']}"
return f"📋 Сделка #{deal_id}:\n{json.dumps(result, ensure_ascii=False, indent=2)}"
except (ValueError, IndexError):
return "Использование: /сделка <ID>"
# Помощь
elif text_lower in ["/help", "/помощь", "помощь", "help"]:
return """🤖 **CRM Assistant - Команды**
📇 **Контакты:**
• `/контакт <имя>` - поиск контакта
💼 **Сделки:**
• `/сделки` - список активных сделок
• `/сделка <ID>` - информация о сделке
ℹ️ `/помощь` - эта справка
Примеры:
• `/контакт Иванов`
• `/сделка 123`
"""
# Приветствие
elif any(word in text_lower for word in ["привет", "hello", "hi", "здравствуй"]):
return f"""👋 Привет! Я {BOT_NAME}.
Я помогу вам работать с CRM данными Bitrix24.
Напишите /помощь для списка команд."""
# Неизвестная команда
else:
return "🤔 Не понял команду. Напишите /помощь для списка доступных команд."
@app.route("/", methods=["GET"])
def index():
"""Проверка работоспособности."""
return jsonify({
"status": "ok",
"bot": BOT_NAME,
"mcp_server": MCP_SERVER_URL
})
@app.route("/webhook", methods=["POST"])
def webhook():
"""Обработчик вебхука от Bitrix24."""
try:
data = request.form.to_dict() or request.json or {}
# Получаем данные сообщения
event = data.get("event", "")
# Обработка события нового сообщения
if event == "ONIMBOTMESSAGEADD":
message_data = data.get("data", {})
dialog_id = message_data.get("PARAMS", {}).get("DIALOG_ID", "")
message_text = message_data.get("PARAMS", {}).get("MESSAGE", "")
if dialog_id and message_text:
# Обрабатываем команду
response_text = process_command(message_text)
# Отправляем ответ
send_bitrix_message(dialog_id, response_text)
return jsonify({"status": "ok"})
except Exception as e:
print(f"Webhook error: {e}")
return jsonify({"status": "error", "message": str(e)}), 500
@app.route("/install", methods=["GET", "POST"])
def install():
"""Обработчик установки бота."""
return jsonify({"status": "ok", "message": "Bot installed"})
@app.route("/health", methods=["GET"])
def health():
"""Health check."""
return jsonify({"status": "healthy"})
if __name__ == "__main__":
if not BITRIX_WEBHOOK_URL:
print("⚠️ Укажите BITRIX_WEBHOOK_URL!")
print(" export BITRIX_WEBHOOK_URL='https://...'")
print(f"🤖 {BOT_NAME} запущен")
print(f"📡 MCP Server: {MCP_SERVER_URL}")
print(f"🌐 Webhook URL: http://0.0.0.0:5000/webhook")
app.run(host="0.0.0.0", port=5000, debug=True)
