Custom Code Execution

Deep dive into Mandatum's unique custom code execution capabilities.

Overview

Mandatum allows you to write and version custom code that runs alongside your prompts. This enables:

• Complex data transformations
• External API integrations
• Advanced validation logic
• Multi-step workflows
• Custom parsing and formatting

Execution Model

┌─────────────┐     ┌──────────────┐     ┌─────────────┐     ┌──────────────┐
│ Your Code   │────▶│ Pre-Process  │────▶│  LLM Call   │────▶│ Post-Process │────▶ Result
└─────────────┘     └──────────────┘     └─────────────┘     └──────────────┘
                     (Your code)                                (Your code)

Writing Integration Code

Python Example

import json
import re
from typing import Dict, Any

def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    """
    Pre-process input before sending to LLM.

    Args:
        input_variables: Original input variables

    Returns:
        Modified input variables for the prompt
    """
    message = input_variables.get('message', '').strip()

    # Clean input
    message = re.sub(r'\s+', ' ', message)

    # Fetch additional context
    user_id = input_variables.get('user_id')
    if user_id:
        user_context = fetch_user_context(user_id)
        return {
            'message': message,
            'user_history': user_context.get('recent_messages', []),
            'user_preferences': user_context.get('preferences', {})
        }

    return {'message': message}

def post_process(llm_output: str, context: Dict[str, Any]) -> Dict[str, Any]:
    """
    Post-process LLM output before returning.

    Args:
        llm_output: Raw output from LLM
        context: Original request context (includes input_variables, metadata)

    Returns:
        Processed result
    """
    # Extract structured data from LLM response
    try:
        # Try to parse JSON from markdown code blocks
        json_match = re.search(r'```json\n(.*?)\n```', llm_output, re.DOTALL)
        if json_match:
            result = json.loads(json_match.group(1))
        else:
            # Try to parse as JSON directly
            result = json.loads(llm_output)

        # Validate result
        if 'category' not in result:
            raise ValueError('Missing required field: category')

        return {
            'success': True,
            'category': result['category'],
            'confidence': result.get('confidence', 1.0),
            'raw_output': llm_output
        }
    except (json.JSONDecodeError, ValueError) as e:
        return {
            'success': False,
            'error': str(e),
            'raw_output': llm_output
        }

def fetch_user_context(user_id: str) -> Dict[str, Any]:
    """Helper function to fetch user context."""
    import requests

    response = requests.get(
        f"https://api.yourapp.com/users/{user_id}/context",
        timeout=5
    )
    return response.json()

TypeScript Example

interface InputVariables {
  [key: string]: any;
}

interface Context {
  input_variables: InputVariables;
  metadata: Record<string, any>;
  prompt_name: string;
  version: number;
}

export async function preProcess(
  inputVariables: InputVariables
): Promise<InputVariables> {
  const message = (inputVariables.message || '').trim();

  // Clean input
  const cleanedMessage = message.replace(/\s+/g, ' ');

  // Fetch additional context
  const userId = inputVariables.user_id;
  if (userId) {
    const userContext = await fetchUserContext(userId);
    return {
      message: cleanedMessage,
      user_history: userContext.recent_messages || [],
      user_preferences: userContext.preferences || {}
    };
  }

  return { message: cleanedMessage };
}

export async function postProcess(
  llmOutput: string,
  context: Context
): Promise<Record<string, any>> {
  try {
    // Extract JSON from markdown code blocks
    const jsonMatch = llmOutput.match(/```json\n(.*?)\n```/s);
    let result: any;

    if (jsonMatch) {
      result = JSON.parse(jsonMatch[1]);
    } else {
      result = JSON.parse(llmOutput);
    }

    // Validate result
    if (!result.category) {
      throw new Error('Missing required field: category');
    }

    return {
      success: true,
      category: result.category,
      confidence: result.confidence ?? 1.0,
      raw_output: llmOutput
    };
  } catch (error) {
    return {
      success: false,
      error: error instanceof Error ? error.message : 'Unknown error',
      raw_output: llmOutput
    };
  }
}

async function fetchUserContext(userId: string): Promise<any> {
  const response = await fetch(
    `https://api.yourapp.com/users/${userId}/context`,
    { signal: AbortSignal.timeout(5000) }
  );
  return response.json();
}

Sandboxed Environment

All integration code runs in secure, isolated containers.

Security Features

• Network isolation: Restricted outbound connections
• Resource limits: CPU, memory, and time constraints
• File system isolation: No access to host filesystem
• Process isolation: Separate container per execution
• Security scanning: Code is scanned for vulnerabilities

Resource Limits

Resource	Limit
Execution Time	30 seconds
Memory	512 MB
CPU	1 vCPU
Network Requests	10 per execution
Disk I/O	Read-only access

Allowed Operations

✓ HTTP/HTTPS requests to external APIs ✓ JSON parsing and manipulation ✓ String operations and regex ✓ Date/time operations ✓ Mathematical calculations ✓ Data validation

✗ File system writes ✗ Database connections (use API instead) ✗ Subprocess execution ✗ Network scanning ✗ Cryptocurrency mining

Available Libraries

Python

Pre-installed packages:

import json          # JSON encoding/decoding
import re            # Regular expressions
import datetime      # Date/time operations
import math          # Mathematical functions
import random        # Random number generation
import hashlib       # Hashing functions
import base64        # Base64 encoding/decoding

import requests      # HTTP client
import pandas        # Data processing
import numpy         # Numerical operations
import pydantic      # Data validation
import jmespath      # JSON querying

TypeScript

Pre-installed packages:

import axios from 'axios';           // HTTP client
import dayjs from 'dayjs';           // Date handling
import lodash from 'lodash';         // Utility functions
import zod from 'zod';               // Schema validation
import { parse } from 'json5';       // JSON5 parsing
import pLimit from 'p-limit';        // Concurrency control

Custom Dependencies

Upload a requirements.txt (Python) or package.json (TypeScript):

# Upload custom dependencies
with open('requirements.txt', 'rb') as f:
    client.integrations.upload_dependencies(
        prompt_name="classifier",
        dependencies_file=f
    )

Example requirements.txt:

beautifulsoup4==4.12.2
lxml==4.9.3
Pillow==10.0.0

Advanced Patterns

Chaining LLM Calls

Use integrations to chain multiple prompts:

def post_process(llm_output: str, context: Dict[str, Any]) -> Dict[str, Any]:
    from mandatum import Mandatum

    client = Mandatum(api_key=context.get('api_key'))

    # First LLM call generates draft
    draft = llm_output

    # Second LLM call reviews the draft
    review = client.prompts.run(
        prompt_name="content-reviewer",
        input_variables={'draft': draft}
    )

    # Third LLM call improves based on review
    final = client.prompts.run(
        prompt_name="content-improver",
        input_variables={
            'draft': draft,
            'feedback': review.output
        }
    )

    return {
        'original_draft': draft,
        'review_feedback': review.output,
        'final_content': final.output,
        'iterations': 3
    }

Parallel Processing

Make multiple requests in parallel:

def post_process(llm_output: str, context: Dict[str, Any]) -> Dict[str, Any]:
    import asyncio
    from concurrent.futures import ThreadPoolExecutor

    async def process_batch(items):
        tasks = [process_item(item) for item in items]
        return await asyncio.gather(*tasks)

    # Extract items from LLM output
    items = extract_items(llm_output)

    # Process in parallel
    with ThreadPoolExecutor(max_workers=5) as executor:
        results = list(executor.map(enrich_item, items))

    return {
        'items': results,
        'total': len(results)
    }

Caching

Cache expensive operations:

from functools import lru_cache

@lru_cache(maxsize=100)
def fetch_user_context(user_id: str) -> Dict[str, Any]:
    """Cached user context fetching."""
    import requests

    response = requests.get(
        f"https://api.yourapp.com/users/{user_id}/context"
    )
    return response.json()

def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    user_id = input_variables.get('user_id')
    if user_id:
        # This will be cached across executions
        context = fetch_user_context(user_id)
        return {**input_variables, 'user_context': context}

    return input_variables

Error Handling

Robust error handling:

def post_process(llm_output: str, context: Dict[str, Any]) -> Dict[str, Any]:
    try:
        # Try primary parsing method
        result = parse_json_output(llm_output)
    except JSONParseError:
        try:
            # Fallback to regex extraction
            result = extract_with_regex(llm_output)
        except ExtractionError:
            # Last resort: return raw output with error
            return {
                'success': False,
                'error': 'Failed to parse LLM output',
                'raw_output': llm_output,
                'fallback_required': True
            }

    # Validate result
    validation_errors = validate_result(result)
    if validation_errors:
        return {
            'success': False,
            'error': 'Validation failed',
            'validation_errors': validation_errors,
            'partial_result': result
        }

    return {
        'success': True,
        'data': result
    }

Retry Logic

Implement retry logic for API calls:

import time
from typing import Any, Callable

def retry_with_backoff(
    func: Callable,
    max_retries: int = 3,
    initial_delay: float = 1.0
) -> Any:
    """Retry function with exponential backoff."""
    for attempt in range(max_retries):
        try:
            return func()
        except Exception as e:
            if attempt == max_retries - 1:
                raise
            delay = initial_delay * (2 ** attempt)
            time.sleep(delay)

def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    user_id = input_variables.get('user_id')

    # Fetch with retry
    user_context = retry_with_backoff(
        lambda: fetch_user_context(user_id),
        max_retries=3
    )

    return {**input_variables, 'user_context': user_context}

Testing

Local Testing

Test your integration code locally:

# test_integration.py
from integration import pre_process, post_process

def test_pre_process():
    input_vars = {
        'message': '  Hello   World  ',
        'user_id': 'user_123'
    }

    result = pre_process(input_vars)

    assert result['message'] == 'Hello World'
    assert 'user_context' in result

def test_post_process():
    llm_output = '{"category": "billing", "confidence": 0.95}'
    context = {
        'input_variables': {'message': 'test'},
        'metadata': {},
        'prompt_name': 'classifier',
        'version': 1
    }

    result = post_process(llm_output, context)

    assert result['success'] is True
    assert result['category'] == 'billing'
    assert result['confidence'] == 0.95

Integration Tests

Test through the API:

def test_end_to_end():
    from mandatum import Mandatum

    client = Mandatum(api_key="test-api-key")

    # Run prompt with integration
    response = client.prompts.run(
        prompt_name="classifier",
        input_variables={'message': 'I need help with billing'},
        metadata={'test': True}
    )

    assert response.success is True
    assert response.output['category'] == 'billing'

Debugging

Logging

Add logging to your integration code:

import logging

logger = logging.getLogger(__name__)

def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    logger.info(f"Pre-processing input: {input_variables}")

    result = process_input(input_variables)

    logger.info(f"Pre-process result: {result}")
    return result

Logs are available in the request details.

Error Context

Include context in errors:

def post_process(llm_output: str, context: Dict[str, Any]) -> Dict[str, Any]:
    try:
        result = parse_output(llm_output)
    except Exception as e:
        return {
            'success': False,
            'error': str(e),
            'error_type': type(e).__name__,
            'llm_output_preview': llm_output[:200],
            'context': {
                'prompt_name': context['prompt_name'],
                'version': context['version']
            }
        }

Performance Optimization

Minimize Cold Starts

Keep initialization code light:

# Good: lazy loading
def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    import heavy_library  # Import only when needed
    return heavy_library.process(input_variables)

# Bad: heavy imports at module level
import heavy_library  # Increases cold start time

def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    return heavy_library.process(input_variables)

Async Operations

Use async for I/O operations:

import asyncio
import aiohttp

async def fetch_multiple(urls: list) -> list:
    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        return await asyncio.gather(*tasks)

def pre_process(input_variables: Dict[str, Any]) -> Dict[str, Any]:
    urls = input_variables.get('urls', [])

    # Run async code
    results = asyncio.run(fetch_multiple(urls))

    return {**input_variables, 'fetched_data': results}

Next Steps

• Learn about versioning
• View integration examples
• Explore API reference
• Set up webhooks