"""
Чат-бот Bitrix24 с YandexGPT и MCP.
Бот использует:
- YandexGPT для понимания естественного языка
- MCP сервер для получения данных из Bitrix24 CRM
Установка:
pip install flask requests yandexcloud
Переменные окружения:
BITRIX_WEBHOOK_URL - URL вебхука Bitrix24
MCP_SERVER_URL - URL MCP сервера
YANDEX_FOLDER_ID - ID каталога Yandex Cloud
YANDEX_API_KEY - API ключ или IAM токен
"""
import json
import os
import requests
from flask import Flask, request, jsonify
app = Flask(__name__)
# =============================================================================
# Конфигурация
# =============================================================================
BITRIX_WEBHOOK_URL = os.getenv("BITRIX_WEBHOOK_URL", "")
MCP_SERVER_URL = os.getenv("MCP_SERVER_URL", "http://localhost:8000/mcp")
# YandexGPT
YANDEX_FOLDER_ID = os.getenv("YANDEX_FOLDER_ID", "")
YANDEX_API_KEY = os.getenv("YANDEX_API_KEY", "") # Или IAM токен
YANDEX_GPT_MODEL = "yandexgpt-lite" # или yandexgpt для Pro версии
BOT_NAME = "CRM AI Assistant"
# =============================================================================
# YandexGPT
# =============================================================================
SYSTEM_PROMPT = """Ты — умный ассистент для работы с CRM Bitrix24.
У тебя есть доступ к следующим функциям:
1. search_contacts(query) - поиск контактов по имени/телефону/email
2. get_contact(contact_id) - получить контакт по ID
3. list_deals(active_only, limit) - список сделок
4. get_deal(deal_id) - получить сделку по ID
5. update_deal_stage(deal_id, stage_id) - обновить стадию сделки
Когда пользователь просит найти контакт или сделку, определи какую функцию вызвать
и верни JSON в формате:
{"function": "имя_функции", "arguments": {...}}
Если нужно просто ответить пользователю без вызова функций, верни обычный текст.
Примеры:
- "найди контакт Иванов" → {"function": "search_contacts", "arguments": {"query": "Иванов"}}
- "покажи сделку 123" → {"function": "get_deal", "arguments": {"deal_id": 123}}
- "активные сделки" → {"function": "list_deals", "arguments": {"active_only": true, "limit": 10}}
"""
def call_yandexgpt(user_message: str, context: str = "") -> str:
"""Вызов YandexGPT API."""
if not YANDEX_FOLDER_ID or not YANDEX_API_KEY:
return '{"error": "YandexGPT не настроен"}'
url = "https://llm.api.cloud.yandex.net/foundationModels/v1/completion"
headers = {
"Content-Type": "application/json",
"Authorization": f"Api-Key {YANDEX_API_KEY}",
# Или для IAM токена: "Authorization": f"Bearer {YANDEX_API_KEY}"
}
messages = [
{"role": "system", "text": SYSTEM_PROMPT},
]
if context:
messages.append({"role": "assistant", "text": f"Контекст из CRM: {context}"})
messages.append({"role": "user", "text": user_message})
payload = {
"modelUri": f"gpt://{YANDEX_FOLDER_ID}/{YANDEX_GPT_MODEL}",
"completionOptions": {
"stream": False,
"temperature": 0.3,
"maxTokens": 1000
},
"messages": messages
}
try:
response = requests.post(url, headers=headers, json=payload, timeout=30)
if response.status_code == 200:
result = response.json()
return result.get("result", {}).get("alternatives", [{}])[0].get("message", {}).get("text", "")
else:
print(f"YandexGPT error: {response.status_code} - {response.text}")
return f"Ошибка YandexGPT: {response.status_code}"
except Exception as e:
print(f"YandexGPT exception: {e}")
return f"Ошибка: {e}"
# =============================================================================
# MCP Client
# =============================================================================
def call_mcp_tool(tool_name: str, arguments: dict) -> dict:
"""Вызов инструмента MCP сервера."""
try:
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 execute_function(func_name: str, args: dict) -> str:
"""Выполнить функцию через MCP."""
valid_functions = [
"search_contacts", "get_contact", "list_contacts",
"list_deals", "get_deal", "update_deal_stage"
]
if func_name not in valid_functions:
return f"Неизвестная функция: {func_name}"
result = call_mcp_tool(func_name, args)
return json.dumps(result, ensure_ascii=False, indent=2)
# =============================================================================
# Bitrix24 Messenger
# =============================================================================
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
# =============================================================================
# Message Processing
# =============================================================================
def process_message(user_message: str) -> str:
"""Обработать сообщение пользователя через YandexGPT + MCP."""
# Шаг 1: Отправляем в YandexGPT для понимания интента
gpt_response = call_yandexgpt(user_message)
# Шаг 2: Проверяем, нужно ли вызвать функцию
try:
# Пробуем распарсить как JSON (вызов функции)
func_call = json.loads(gpt_response)
if "function" in func_call:
func_name = func_call["function"]
func_args = func_call.get("arguments", {})
# Вызываем функцию через MCP
crm_data = execute_function(func_name, func_args)
# Отправляем результат обратно в YandexGPT для форматирования
formatted_response = call_yandexgpt(
f"Пользователь спросил: {user_message}\n\nДанные из CRM:\n{crm_data}\n\nСформулируй понятный ответ для пользователя.",
context=crm_data
)
return formatted_response
elif "error" in func_call:
return f"❌ {func_call['error']}"
except json.JSONDecodeError:
# Это обычный текстовый ответ
pass
return gpt_response
# =============================================================================
# Flask Routes
# =============================================================================
@app.route("/", methods=["GET"])
def index():
"""Проверка работоспособности."""
return jsonify({
"status": "ok",
"bot": BOT_NAME,
"mcp_server": MCP_SERVER_URL,
"yandexgpt": "configured" if YANDEX_API_KEY else "not configured"
})
@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:
# Обработка через AI
response_text = process_message(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("/health", methods=["GET"])
def health():
return jsonify({"status": "healthy"})
# =============================================================================
# Main
# =============================================================================
if __name__ == "__main__":
print(f"🤖 {BOT_NAME}")
print(f"═" * 40)
# Проверка конфигурации
checks = [
("BITRIX_WEBHOOK_URL", bool(BITRIX_WEBHOOK_URL)),
("MCP_SERVER_URL", bool(MCP_SERVER_URL)),
("YANDEX_FOLDER_ID", bool(YANDEX_FOLDER_ID)),
("YANDEX_API_KEY", bool(YANDEX_API_KEY)),
]
for name, ok in checks:
status = "✅" if ok else "❌"
print(f"{status} {name}")
print(f"═" * 40)
print(f"🌐 Webhook: http://0.0.0.0:5000/webhook")
app.run(host="0.0.0.0", port=5000, debug=True)
