motia/bitbylaw/services/ESPOCRM_SERVICE.md

EspoCRM API Service

Overview

Python client for EspoCRM REST API integration. Provides type-safe, async operations for managing entities in EspoCRM.

Features

• ✅ API Key authentication
• ✅ Async/await support (aiohttp)
• ✅ Full CRUD operations
• ✅ Entity search and filtering
• ✅ Error handling with custom exceptions
• ✅ Optional Redis integration for caching
• ✅ Logging via Motia context

Installation

from services.espocrm import EspoCRMAPI

# Initialize with optional context for logging
espo = EspoCRMAPI(context=context)

Configuration

Add to .env or environment:

# EspoCRM API Configuration
ESPOCRM_API_BASE_URL=https://crm.bitbylaw.com/api/v1
ESPOCRM_MARVIN_API_KEY=your_api_key_here
ESPOCRM_API_TIMEOUT_SECONDS=30

Required in config.py:

class Config:
    ESPOCRM_API_BASE_URL = os.getenv('ESPOCRM_API_BASE_URL', 'https://crm.bitbylaw.com/api/v1')
    ESPOCRM_API_KEY = os.getenv('ESPOCRM_MARVIN_API_KEY', '')
    ESPOCRM_API_TIMEOUT_SECONDS = int(os.getenv('ESPOCRM_API_TIMEOUT_SECONDS', '30'))

API Methods

Get Single Entity

async def get_entity(entity_type: str, entity_id: str) -> Dict[str, Any]

Usage:

# Get Beteiligter by ID
result = await espo.get_entity('Beteiligte', '64a3f2b8c9e1234567890abc')
print(result['name'])

List Entities

async def list_entities(
    entity_type: str,
    where: Optional[List[Dict]] = None,
    select: Optional[str] = None,
    order_by: Optional[str] = None,
    offset: int = 0,
    max_size: int = 50
) -> Dict[str, Any]

Usage:

# List all Beteiligte with status "Active"
result = await espo.list_entities(
    'Beteiligte',
    where=[{
        'type': 'equals',
        'attribute': 'status',
        'value': 'Active'
    }],
    select='id,name,email',
    max_size=100
)

for entity in result['list']:
    print(entity['name'])
print(f"Total: {result['total']}")

Complex Filters:

# OR condition
where=[{
    'type': 'or',
    'value': [
        {'type': 'equals', 'attribute': 'status', 'value': 'Zurückgestellt'},
        {'type': 'equals', 'attribute': 'status', 'value': 'Warte auf neuen Anruf'}
    ]
}]

# AND condition
where=[
    {'type': 'equals', 'attribute': 'status', 'value': 'Active'},
    {'type': 'greaterThan', 'attribute': 'createdAt', 'value': '2026-01-01'}
]

Create Entity

async def create_entity(entity_type: str, data: Dict[str, Any]) -> Dict[str, Any]

Usage:

# Create new Beteiligter
result = await espo.create_entity('Beteiligte', {
    'name': 'Max Mustermann',
    'email': 'max@example.com',
    'phone': '+49123456789',
    'status': 'New'
})
print(f"Created with ID: {result['id']}")

Update Entity

async def update_entity(
    entity_type: str,
    entity_id: str,
    data: Dict[str, Any]
) -> Dict[str, Any]

Usage:

# Update Beteiligter status
result = await espo.update_entity(
    'Beteiligte',
    '64a3f2b8c9e1234567890abc',
    {'status': 'Converted'}
)

Delete Entity

async def delete_entity(entity_type: str, entity_id: str) -> bool

Usage:

# Delete Beteiligter
success = await espo.delete_entity('Beteiligte', '64a3f2b8c9e1234567890abc')

Search Entities

async def search_entities(
    entity_type: str,
    query: str,
    fields: Optional[List[str]] = None
) -> List[Dict[str, Any]]

Usage:

# Full-text search
results = await espo.search_entities('Beteiligte', 'Mustermann')
for entity in results:
    print(entity['name'])

Common Entity Types

Based on EspoCRM standard and VMH customization:

• Beteiligte - Custom entity for VMH participants
• CVmhErstgespraech - Custom entity for VMH initial consultations
• Contact - Standard contacts
• Account - Companies/Organizations
• Lead - Sales leads
• Opportunity - Sales opportunities
• Case - Support cases
• Meeting - Calendar meetings
• Call - Phone calls
• Email - Email records

Error Handling

from services.espocrm import EspoCRMError, EspoCRMAuthError

try:
    result = await espo.get_entity('Beteiligte', entity_id)
except EspoCRMAuthError as e:
    # Invalid API key
    context.logger.error(f"Authentication failed: {e}")
except EspoCRMError as e:
    # General API error (404, 403, etc.)
    context.logger.error(f"API error: {e}")

Authentication

EspoCRM uses API Key authentication via X-Api-Key header.

Create API Key in EspoCRM:

1. Login as admin
2. Go to Administration → API Users
3. Create new API User
4. Copy API Key
5. Set permissions for API User

Headers sent automatically:

X-Api-Key: your_api_key_here
Content-Type: application/json
Accept: application/json

Integration Examples

In Motia Step

from services.espocrm import EspoCRMAPI

config = {
    'type': 'event',
    'name': 'Sync Beteiligter to Advoware',
    'subscribes': ['vmh.beteiligte.create']
}

async def handler(event, context):
    entity_id = event['data']['entity_id']
    
    # Fetch from EspoCRM
    espo = EspoCRMAPI(context=context)
    beteiligter = await espo.get_entity('Beteiligte', entity_id)
    
    context.logger.info(f"Processing: {beteiligter['name']}")
    
    # Transform and sync to Advoware...
    # ...

In Cron Step

from services.espocrm import EspoCRMAPI
from datetime import datetime, timedelta

config = {
    'type': 'cron',
    'cron': '*/5 * * * *',
    'name': 'Check Expired Callbacks'
}

async def handler(input, context):
    espo = EspoCRMAPI(context=context)
    
    # Find expired callbacks
    now = datetime.utcnow().isoformat() + 'Z'
    
    result = await espo.list_entities(
        'CVmhErstgespraech',
        where=[
            {'type': 'lessThan', 'attribute': 'nchsterAnruf', 'value': now},
            {'type': 'equals', 'attribute': 'status', 'value': 'Warte auf neuen Anruf'}
        ]
    )
    
    # Update status for expired entries
    for entry in result['list']:
        await espo.update_entity(
            'CVmhErstgespraech',
            entry['id'],
            {'status': 'Neu'}
        )
        context.logger.info(f"Reset status for {entry['id']}")

Helper Script: Compare Structures

Compare entity structures between EspoCRM and Advoware:

# Compare by EspoCRM ID (auto-search in Advoware)
python bitbylaw/scripts/compare_beteiligte.py 64a3f2b8c9e1234567890abc

# Compare with specific Advoware ID
python bitbylaw/scripts/compare_beteiligte.py 64a3f2b8c9e1234567890abc 12345

Output:

• Entity data from both systems
• Field structure comparison
• Suggested field mappings
• JSON output saved to scripts/beteiligte_comparison_result.json

Performance

Timeout

Default: 30 seconds (configurable via ESPOCRM_API_TIMEOUT_SECONDS)

# Custom timeout for specific call
result = await espo.api_call('/Beteiligte', timeout_seconds=60)

Pagination

# Fetch in pages
offset = 0
max_size = 50

while True:
    result = await espo.list_entities(
        'Beteiligte',
        offset=offset,
        max_size=max_size
    )
    
    entities = result['list']
    if not entities:
        break
    
    # Process entities...
    
    offset += len(entities)
    
    if len(entities) < max_size:
        break  # Last page

Rate Limiting

Optional Redis-based rate limiting can be implemented:

# Check rate limit before API call
rate_limit_key = f'espocrm:rate_limit:{entity_type}'
if espo.redis_client:
    count = espo.redis_client.incr(rate_limit_key)
    espo.redis_client.expire(rate_limit_key, 60)  # 1 minute window
    
    if count > 100:  # Max 100 requests per minute
        raise Exception("Rate limit exceeded")

Testing

import pytest
from services.espocrm import EspoCRMAPI

@pytest.mark.asyncio
async def test_get_entity():
    espo = EspoCRMAPI()
    
    # Mock or use test entity ID
    result = await espo.get_entity('Contact', 'test-id-123')
    
    assert 'id' in result
    assert result['id'] == 'test-id-123'

Logging

All operations are logged via context.logger:

[INFO] [EspoCRM] EspoCRM API initialized with base URL: https://crm.bitbylaw.com/api/v1
[DEBUG] [EspoCRM] API call: GET https://crm.bitbylaw.com/api/v1/Beteiligte/123
[DEBUG] [EspoCRM] Response status: 200
[INFO] [EspoCRM] Getting Beteiligte with ID: 123

Related Files

• services/espocrm.py - Implementation
• scripts/compare_beteiligte.py - Comparison tool
• steps/crm-bbl-vmh-reset-nextcall_step.py - Example usage
• config.py - Configuration

EspoCRM API Documentation

Official docs: https://docs.espocrm.com/development/api/

Key Concepts:

• RESTful API with JSON
• Entity-based operations
• Filter operators: equals, notEquals, greaterThan, lessThan, like, contains, in, isNull, isNotNull
• Boolean operators: and (default), or
• Metadata API: /Metadata (for entity definitions)