Битрикс24 API Python в 2025: инструкция по интеграции и автоматизации

Интеграция Битрикс24 с внешними системами через API на языке Python открывает широкие возможности для автоматизации бизнес-процессов. В этой статье мы разберем, как настроить подключение к API Битрикс24, рассмотрим основные методы работы с данными и приведем практические примеры использования Python для решения типовых задач.

Что такое API Битрикс24 и зачем он нужен

API (Application Programming Interface) Битрикс24 — это набор методов и инструментов, которые позволяют внешним приложениям взаимодействовать с данными и функциями системы. С помощью API можно:

• Синхронизировать данные между Битрикс24 и другими системами
• Автоматизировать создание и обновление сделок, контактов, компаний
• Получать отчеты и аналитику
• Интегрировать внешние сервисы с CRM
• Создавать собственные приложения для Битрикс24

Python идеально подходит для работы с API благодаря своей простоте, богатой экосистеме библиотек и отличной поддержке HTTP-запросов.

Настройка подключения к API Битрикс24

Получение доступа к API

Для работы с API Битрикс24 необходимо получить авторизационные данные. Существует несколько способов авторизации:

• Входящий webhook — простой способ для базовых операций
• Исходящий webhook — для получения уведомлений о событиях
• Локальное приложение — для комплексных решений
• OAuth 2.0 — для публичных приложений

Рассмотрим настройку входящего webhook как наиболее простого варианта:

1. Войдите в административную панель Битрикс24
2. Перейдите в раздел «Разработчикам» → «Другое» → «Входящий webhook»
3. Нажмите «Добавить webhook»
4. Выберите необходимые права доступа
5. Скопируйте полученный URL

Установка необходимых библиотек Python

Для работы с API Битрикс24 понадобятся следующие библиотеки:

pip install requests
pip install json
pip install datetime

Основы работы с API Битрикс24 на Python

Базовая структура запроса

Все запросы к API Битрикс24 имеют следующую структуру:

import requests
import json
# Базовый URL webhook
webhook_url = "https://your-domain.bitrix24.ru/rest/1/webhook_code/"
# Функция для выполнения запросов
def bitrix_request(method, params=None):
url = webhook_url + method
if params is None:
params = {}
response = requests.post(url, json=params)
if response.status_code == 200:
return response.json()
else:
print(f"Ошибка: {response.status_code}")
return None

Получение информации о портале

Начнем с простого запроса для получения информации о портале:

# Получение информации о портале
result = bitrix_request("app.info")
if result:
print(f"Статус: {result['result']['status']}")
print(f"Язык: {result['result']['language_id']}")

Работа с основными сущностями CRM

Управление контактами

Контакты — одна из ключевых сущностей в Битрикс24. Рассмотрим основные операции:

Создание контакта

def create_contact(name, phone, email):
params = {
"fields": {
"NAME": name,
"PHONE": [{"VALUE": phone, "VALUE_TYPE": "WORK"}],
"EMAIL": [{"VALUE": email, "VALUE_TYPE": "WORK"}]
}
}
result = bitrix_request("crm.contact.add", params)
if result and result.get("result"):
contact_id = result["result"]
print(f"Контакт создан с ID: {contact_id}")
return contact_id
return None
# Пример использования
create_contact("Иван Иванов", "+7 (999) 123-45-67", "ivan@example.com")

Получение списка контактов

def get_contacts(limit=50):
params = {
"select": ["ID", "NAME", "PHONE", "EMAIL"],
"start": 0
}
result = bitrix_request("crm.contact.list", params)
if result and result.get("result"):
contacts = result["result"]
for contact in contacts:
print(f"ID: {contact['ID']}, Имя: {contact['NAME']}")
return contacts
return []
# Получение контактов
contacts = get_contacts()

Обновление контакта

def update_contact(contact_id, fields):
params = {
"id": contact_id,
"fields": fields
}
result = bitrix_request("crm.contact.update", params)
if result and result.get("result"):
print(f"Контакт {contact_id} обновлен")
return True
return False
# Обновление контакта
update_contact(123, {"NAME": "Иван Петрович Иванов"})

Работа со сделками

Сделки — основа работы с продажами в Битрикс24. Рассмотрим ключевые операции:

Создание сделки

def create_deal(title, contact_id, opportunity=0):
params = {
"fields": {
"TITLE": title,
"CONTACT_ID": contact_id,
"OPPORTUNITY": opportunity,
"CURRENCY_ID": "RUB",
"STAGE_ID": "NEW"
}
}
result = bitrix_request("crm.deal.add", params)
if result and result.get("result"):
deal_id = result["result"]
print(f"Сделка создана с ID: {deal_id}")
return deal_id
return None
# Создание сделки
create_deal("Продажа товара", 123, 50000)

Получение списка сделок

def get_deals(filter_params=None):
params = {
"select": ["ID", "TITLE", "STAGE_ID", "OPPORTUNITY", "CONTACT_ID"],
"filter": filter_params or {}
}
result = bitrix_request("crm.deal.list", params)
if result and result.get("result"):
deals = result["result"]
for deal in deals:
print(f"ID: {deal['ID']}, Название: {deal['TITLE']}, Сумма: {deal['OPPORTUNITY']}")
return deals
return []
# Получение всех сделок
deals = get_deals()
# Получение сделок в определенной стадии
new_deals = get_deals({"STAGE_ID": "NEW"})

Управление компаниями

Компании позволяют группировать контакты и сделки по организациям:

def create_company(title, phone, email):
params = {
"fields": {
"TITLE": title,
"PHONE": [{"VALUE": phone, "VALUE_TYPE": "WORK"}],
"EMAIL": [{"VALUE": email, "VALUE_TYPE": "WORK"}]
}
}
result = bitrix_request("crm.company.add", params)
if result and result.get("result"):
company_id = result["result"]
print(f"Компания создана с ID: {company_id}")
return company_id
return None
# Создание компании
create_company("ООО Рога и Копыта", "+7 (495) 123-45-67", "info@company.ru")

Работа с лидами

Лиды представляют потенциальных клиентов, которые еще не стали полноценными контактами:

Создание лида

def create_lead(title, name, phone, email):
params = {
"fields": {
"TITLE": title,
"NAME": name,
"PHONE": [{"VALUE": phone, "VALUE_TYPE": "WORK"}],
"EMAIL": [{"VALUE": email, "VALUE_TYPE": "WORK"}],
"SOURCE_ID": "WEB"
}
}
result = bitrix_request("crm.lead.add", params)
if result and result.get("result"):
lead_id = result["result"]
print(f"Лид создан с ID: {lead_id}")
return lead_id
return None
# Создание лида
create_lead("Заявка с сайта", "Петр Сидоров", "+7 (999) 987-65-43", "petr@example.com")

Конвертация лида в контакт и сделку

def convert_lead(lead_id):
params = {
"id": lead_id,
"fields": {
"CONTACT": "Y",
"DEAL": "Y"
}
}
result = bitrix_request("crm.lead.convert", params)
if result and result.get("result"):
print(f"Лид {lead_id} конвертирован")
return result["result"]
return None

Работа с пользовательскими полями

Битрикс24 позволяет создавать собственные поля для различных сущностей:

Получение пользовательских полей

def get_custom_fields(entity_type):
"""
entity_type: CRM_CONTACT, CRM_DEAL, CRM_COMPANY, CRM_LEAD
"""
params = {
"entityTypeId": entity_type
}
result = bitrix_request("crm.userfield.list", params)
if result and result.get("result"):
fields = result["result"]
for field in fields:
print(f"Поле: {field['FIELD_NAME']}, Тип: {field['USER_TYPE_ID']}")
return fields
return []
# Получение пользовательских полей контактов
custom_fields = get_custom_fields("CRM_CONTACT")

Работа с задачами

API Битрикс24 также предоставляет возможность работы с задачами:

Создание задачи

def create_task(title, description, responsible_id, deadline=None):
params = {
"fields": {
"TITLE": title,
"DESCRIPTION": description,
"RESPONSIBLE_ID": responsible_id,
"DEADLINE": deadline
}
}
result = bitrix_request("tasks.task.add", params)
if result and result.get("result"):
task_id = result["result"]["task"]["id"]
print(f"Задача создана с ID: {task_id}")
return task_id
return None
# Создание задачи
create_task("Позвонить клиенту", "Обсудить условия сделки", 1, "2024-12-31T23:59:59")

Обработка ошибок и лимитов API

При работе с API важно учитывать ограничения и правильно обрабатывать ошибки:

Улучшенная функция для запросов

import time
from datetime import datetime
def safe_bitrix_request(method, params=None, max_retries=3):
"""
Безопасная функция для выполнения запросов с обработкой ошибок
"""
url = webhook_url + method
if params is None:
params = {}
for attempt in range(max_retries):
try:
response = requests.post(url, json=params, timeout=30)
if response.status_code == 200:
result = response.json()
# Проверка на ошибки в ответе
if 'error' in result:
print(f"Ошибка API: {result['error']}")
if result['error'] == 'QUERY_LIMIT_EXCEEDED':
print("Превышен лимит запросов, ожидание...")
time.sleep(2)
continue
return None
return result
elif response.status_code == 429:
print("Слишком много запросов, ожидание...")
time.sleep(2 ** attempt)
continue
else:
print(f"HTTP ошибка: {response.status_code}")
return None
except requests.exceptions.RequestException as e:
print(f"Ошибка соединения: {e}")
if attempt < max_retries - 1:
time.sleep(2 ** attempt)
continue
return None
print("Превышено максимальное количество попыток")
return None

Массовые операции и пакетные запросы

Для повышения производительности при работе с большим количеством данных используйте пакетные запросы:

Пакетное создание контактов

def batch_create_contacts(contacts_data):
"""
Создание контактов пакетом
contacts_data: список словарей с данными контактов
"""
commands = {}
for i, contact_data in enumerate(contacts_data):
commands[f"contact_{i}"] = {
"method": "crm.contact.add",
"params": {"fields": contact_data}
}
params = {"cmd": commands}
result = bitrix_request("batch", params)
if result and result.get("result"):
created_ids = []
for key, value in result["result"]["result"].items():
if value:
created_ids.append(value)
print(f"Контакт {key} создан с ID: {value}")
return created_ids
return []
# Пример использования
contacts = [
{"NAME": "Контакт 1", "PHONE": [{"VALUE": "+7 (999) 111-11-11"}]},
{"NAME": "Контакт 2", "PHONE": [{"VALUE": "+7 (999) 222-22-22"}]},
{"NAME": "Контакт 3", "PHONE": [{"VALUE": "+7 (999) 333-33-33"}]}
]
batch_create_contacts(contacts)

Работа с файлами

API позволяет загружать и работать с файлами:

Загрузка файла

def upload_file(file_path, filename):
"""
Загрузка файла в Битрикс24
"""
with open(file_path, 'rb') as file:
files = {'file': (filename, file, 'application/octet-stream')}
response = requests.post(
webhook_url + "disk.folder.uploadfile",
files=files,
data={'id': 1}  # ID папки
)
if response.status_code == 200:
result = response.json()
if result.get('result'):
file_id = result['result']['ID']
print(f"Файл загружен с ID: {file_id}")
return file_id
return None

Мониторинг и уведомления

Создание системы мониторинга для отслеживания изменений:

Получение последних изменений

from datetime import datetime, timedelta
def get_recent_deals(hours=24):
"""
Получение сделок, измененных за последние N часов
"""
cutoff_time = datetime.now() - timedelta(hours=hours)
cutoff_str = cutoff_time.strftime("%Y-%m-%dT%H:%M:%S")
params = {
"select": ["ID", "TITLE", "STAGE_ID", "DATE_MODIFY"],
"filter": {
">=DATE_MODIFY": cutoff_str
}
}
result = bitrix_request("crm.deal.list", params)
if result and result.get("result"):
deals = result["result"]
print(f"Найдено {len(deals)} измененных сделок")
return deals
return []
# Получение сделок за последние 24 часа
recent_deals = get_recent_deals(24)

Создание класса для работы с Битрикс24

Для удобства работы создадим класс, объединяющий все функции:

class Bitrix24API:
def __init__(self, webhook_url):
self.webhook_url = webhook_url
self.session = requests.Session()
def request(self, method, params=None):
"""Базовый метод для выполнения запросов"""
url = self.webhook_url + method
if params is None:
params = {}
try:
response = self.session.post(url, json=params, timeout=30)
if response.status_code == 200:
result = response.json()
if 'error' in result:
raise Exception(f"API Error: {result['error']}")
return result
else:
raise Exception(f"HTTP Error: {response.status_code}")
except requests.exceptions.RequestException as e:
raise Exception(f"Request Error: {e}")
def create_contact(self, name, phone=None, email=None):
"""Создание контакта"""
fields = {"NAME": name}
if phone:
fields["PHONE"] = [{"VALUE": phone, "VALUE_TYPE": "WORK"}]
if email:
fields["EMAIL"] = [{"VALUE": email, "VALUE_TYPE": "WORK"}]
params = {"fields": fields}
result = self.request("crm.contact.add", params)
return result.get("result")
def get_contacts(self, filter_params=None, select=None):
"""Получение списка контактов"""
params = {
"select": select or ["ID", "NAME", "PHONE", "EMAIL"],
"filter": filter_params or {}
}
result = self.request("crm.contact.list", params)
return result.get("result", [])
def create_deal(self, title, contact_id=None, opportunity=0):
"""Создание сделки"""
fields = {
"TITLE": title,
"OPPORTUNITY": opportunity,
"CURRENCY_ID": "RUB"
}
if contact_id:
fields["CONTACT_ID"] = contact_id
params = {"fields": fields}
result = self.request("crm.deal.add", params)
return result.get("result")
# Пример использования класса
api = Bitrix24API("https://your-domain.bitrix24.ru/rest/1/webhook_code/")
# Создание контакта
contact_id = api.create_contact("Новый клиент", "+7 (999) 123-45-67", "client@example.com")
# Создание сделки
deal_id = api.create_deal("Новая сделка", contact_id, 100000)

Примеры практических задач

Синхронизация данных из внешней системы

def sync_external_data():
"""
Синхронизация данных из внешней системы
"""
# Предположим, что данные приходят из внешней системы
external_contacts = [
{"name": "Иван Петров", "phone": "+7 (999) 111-11-11", "email": "ivan@test.com"},
{"name": "Мария Сидорова", "phone": "+7 (999) 222-22-22", "email": "maria@test.com"}
]
api = Bitrix24API("https://your-domain.bitrix24.ru/rest/1/webhook_code/")
for contact_data in external_contacts:
# Проверяем, существует ли контакт
existing_contacts = api.get_contacts(
filter_params={"PHONE": contact_data["phone"]}
)
if not existing_contacts:
# Создаем новый контакт
contact_id = api.create_contact(
contact_data["name"],
contact_data["phone"],
contact_data["email"]
)
print(f"Создан новый контакт: {contact_id}")
else:
print(f"Контакт уже существует: {existing_contacts[0]['ID']}")
# Запуск синхронизации
sync_external_data()

Автоматическое создание задач для менеджеров

def create_follow_up_tasks():
"""
Создание задач для менеджеров по новым сделкам
"""
api = Bitrix24API("https://your-domain.bitrix24.ru/rest/1/webhook_code/")
# Получаем новые сделки
new_deals = api.get_deals(filter_params={"STAGE_ID": "NEW"})
for deal in new_deals:
# Создаем задачу для ответственного менеджера
task_params = {
"fields": {
"TITLE": f"Связаться с клиентом по сделке: {deal['TITLE']}",
"DESCRIPTION": f"Необходимо связаться с клиентом по сделке ID: {deal['ID']}",
"RESPONSIBLE_ID": deal.get("ASSIGNED_BY_ID", 1),
"DEADLINE": (datetime.now() + timedelta(days=1)).strftime("%Y-%m-%dT%H:%M:%S")
}
}
result = api.request("tasks.task.add", task_params)
if result and result.get("result"):
print(f"Создана задача для сделки {deal['ID']}")

Отладка и тестирование

Для эффективной разработки важно правильно настроить отладку:

Логирование запросов

import logging
# Настройка логирования
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)
class Bitrix24APIWithLogging(Bitrix24API):
def request(self, method, params=None):
logger.info(f"Выполняется запрос: {method}")
logger.debug(f"Параметры: {params}")
try:
result = super().request(method, params)
logger.info(f"Запрос {method} выполнен успешно")
return result
except Exception as e:
logger.error(f"Ошибка в запросе {method}: {e}")
raise
# Использование с логированием
api = Bitrix24APIWithLogging("https://your-domain.bitrix24.ru/rest/1/webhook_code/")

Безопасность и лучшие практики

Управление конфигурацией

import os
from configparser import ConfigParser
class Config:
def __init__(self, config_file='bitrix24.ini'):
self.config = ConfigParser()
self.config.read(config_file)
# Загрузка из переменных окружения (приоритет выше)
self.webhook_url = os.getenv('BITRIX24_WEBHOOK_URL') or self.config.get('bitrix24', 'webhook_url')
def get_webhook_url(self):
return self.webhook_url
# Пример файла bitrix24.ini
"""
[bitrix24]
webhook_url = https://your-domain.bitrix24.ru/rest/1/webhook_code/
"""
# Использование
config = Config()
api = Bitrix24API(config.get_webhook_url())

Валидация данных

import re
def validate_phone(phone):
"""Валидация номера телефона"""
pattern = r'^[\+]?[1-9][\d\s\-\(\)]{7,15}$'
return re.match(pattern, phone) is not None
def validate_email(email):
"""Валидация email"""
pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
return re.match(pattern, email) is not None
def safe_create_contact(api, name, phone=None, email=None):
"""Безопасное создание контакта с валидацией"""
if not name or len(name.strip()) == 0:
raise ValueError("Имя контакта не может быть пустым")
if phone and not validate_phone(phone):
raise ValueError("Неверный формат телефона")
if email and not validate_email(email):
raise ValueError("Неверный формат email")
return api.create_contact(name, phone, email)

Оптимизация производительности

Кэширование результатов

import time
from functools import lru_cache
class CachedBitrix24API(Bitrix24API):
def __init__(self, webhook_url, cache_ttl=300):
super().__init__(webhook_url)
self.cache_ttl = cache_ttl
self._cache = {}
def _get_cache_key(self, method, params):
"""Создание ключа для кэша"""
return f"{method}_{hash(str(params))}"
def _is_cache_valid(self, timestamp):
"""Проверка валидности кэша"""
return time.time() - timestamp < self.cache_ttl
def cached_request(self, method, params=None):
"""Запрос с кэшированием"""
cache_key = self._get_cache_key(method, params)
if cache_key in self._cache:
result, timestamp = self._cache[cache_key]
if self._is_cache_valid(timestamp):
return result
# Выполняем запрос и кэшируем результат
result = self.request(method, params)
self._cache[cache_key] = (result, time.time())
return result
# Использование с кэшированием
api = CachedBitrix24API("https://your-domain.bitrix24.ru/rest/1/webhook_code/", cache_ttl=600)

Мониторинг и метрики

Отслеживание статистики API

class Bitrix24APIWithMetrics(Bitrix24API):
def __init__(self, webhook_url):
super().__init__(webhook_url)
self.metrics = {
'requests_total': 0,
'requests_success': 0,
'requests_failed': 0,
'response_times': []
}
def request(self, method, params=None):
start_time = time.time()
self.metrics['requests_total'] += 1
try:
result = super().request(method, params)
self.metrics['requests_success'] += 1
return result
except Exception as e:
self.metrics['requests_failed'] += 1
raise
finally:
response_time = time.time() - start_time
self.metrics['response_times'].append(response_time)
def get_stats(self):
"""Получение статистики"""
if not self.metrics['response_times']:
return self.metrics
stats = self.metrics.copy()
stats['avg_response_time'] = sum(self.metrics['response_times']) / len(self.metrics['response_times'])
stats['success_rate'] = self.metrics['requests_success'] / self.metrics['requests_total'] * 100
return stats
# Использование
api = Bitrix24APIWithMetrics("https://your-domain.bitrix24.ru/rest/1/webhook_code/")
# Выполнение запросов
api.get_contacts()
api.create_contact("Тестовый контакт")
# Получение статистики
print(api.get_stats())

Заключение

Интеграция Битрикс24 с Python через API открывает безграничные возможности для автоматизации бизнес-процессов. В этой статье мы рассмотрели:

• Основы подключения к API Битрикс24
• Работу с ключевыми сущностями CRM
• Создание безопасных и эффективных запросов
• Обработку ошибок и ограничений
• Оптимизацию производительности
• Практические примеры использования

Правильная реализация интеграции с API Битрикс24 позволит значительно повысить эффективность работы с клиентами, автоматизировать рутинные задачи и обеспечить синхронизацию данных между различными системами.

Мы предоставляем профессиональные услуги по настройке и внедрению Битрикс24:

• Настройка интеграций с внешними системами через API
• Разработка пользовательских приложений на Python
• Автоматизация бизнес-процессов и рабочих потоков
• Миграция данных и синхронизация систем
• Обучение команды работе с API
• Техническая поддержка и сопровождение

Наши специалисты помогут вам максимально эффективно использовать возможности Битрикс24 и создать решения, идеально подходящие для ваших задач. Обращайтесь за консультацией и получите профессиональную помощь в реализации ваших проектов.