Custom Integrations

Mandatum's unique feature: write custom code that runs alongside your prompts for pre-processing, post-processing, and complex workflows.

Why Custom Code?

While most prompt platforms only allow you to manage prompts, Mandatum lets you write code that:

• Pre-processes inputs before sending to the LLM
• Post-processes outputs to extract, validate, or transform results
• Implements complex logic that prompts alone can't handle
• Integrates with external APIs and databases

All integration code is versioned alongside your prompts with git-like history.

Creating an Integration

Via Dashboard

1. Navigate to your prompt
2. Click the Integrations tab
3. Click New Integration
4. Choose language (Python or TypeScript)
5. Write your code in the Monaco editor
6. Click Save & Test

Via API

Python Integration

from mandatum import Mandatum

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

integration_code = """
def pre_process(input_variables: dict) -> dict:
    # Clean and validate input
    message = input_variables.get('message', '').strip()

    if len(message) > 1000:
        message = message[:1000]

    return {
        'message': message,
        'char_count': len(message)
    }

def post_process(llm_output: str, context: dict) -> dict:
    # Extract JSON from LLM response
    import json

    try:
        result = json.loads(llm_output)
        result['processed'] = True
        return result
    except json.JSONDecodeError:
        return {
            'error': 'Invalid JSON response',
            'raw_output': llm_output,
            'processed': False
        }
"""

integration = client.integrations.create(
    prompt_name="customer-classifier",
    language="python",
    code=integration_code
)

TypeScript Integration

const integrationCode = `
export async function preProcess(inputVariables: Record<string, any>): Promise<Record<string, any>> {
  const message = (inputVariables.message || '').trim();

  if (message.length > 1000) {
    return {
      message: message.substring(0, 1000),
      charCount: 1000,
      truncated: true
    };
  }

  return {
    message,
    charCount: message.length,
    truncated: false
  };
}

export async function postProcess(llmOutput: string, context: Record<string, any>): Promise<Record<string, any>> {
  try {
    const result = JSON.parse(llmOutput);
    return {
      ...result,
      processed: true
    };
  } catch (error) {
    return {
      error: 'Invalid JSON response',
      rawOutput: llmOutput,
      processed: false
    };
  }
}
`;

const integration = await client.integrations.create({
  promptName: 'customer-classifier',
  language: 'typescript',
  code: integrationCode
});

Integration Lifecycle

When you run a prompt with integrations:

1. Pre-process → 2. LLM Call → 3. Post-process → 4. Return Result

Pre-processing

Transform input variables before the LLM call:

def pre_process(input_variables: dict) -> dict:
    # Fetch additional context from database
    user_id = input_variables.get('user_id')
    user_data = fetch_user_context(user_id)

    # Combine input with context
    return {
        **input_variables,
        'user_history': user_data['recent_interactions'],
        'user_preferences': user_data['preferences']
    }

Post-processing

Process LLM output before returning:

def post_process(llm_output: str, context: dict) -> dict:
    # Parse structured output
    import re

    category_match = re.search(r'Category: (\w+)', llm_output)
    confidence_match = re.search(r'Confidence: ([\d.]+)', llm_output)

    return {
        'category': category_match.group(1) if category_match else 'unknown',
        'confidence': float(confidence_match.group(1)) if confidence_match else 0.0,
        'raw_output': llm_output
    }

Code Execution Environment

Sandboxing

All integration code runs in secure, isolated containers with:

• Resource limits: CPU, memory, and execution time caps
• Network isolation: No outbound connections by default
• File system isolation: No access to host file system

Available Libraries

Python

Pre-installed packages:

import json
import re
import datetime
import requests  # HTTP requests
import pandas   # Data processing
import numpy    # Numerical operations

TypeScript

Pre-installed packages:

import axios from 'axios';      // HTTP requests
import dayjs from 'dayjs';      // Date handling
import lodash from 'lodash';    // Utilities
import zod from 'zod';          // Validation

Custom Dependencies

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

# Upload custom dependencies
client.integrations.upload_dependencies(
    prompt_name="customer-classifier",
    dependencies_file="requirements.txt"
)

Execution Limits

Resource	Limit
Execution Time	30 seconds
Memory	512 MB
CPU	1 vCPU
Network Requests	10 per execution

Advanced Examples

API Integration

Call external APIs in your integration:

def pre_process(input_variables: dict) -> dict:
    import requests

    # Fetch real-time data from external API
    weather_data = requests.get(
        f"https://api.weather.com/v1/current",
        params={"location": input_variables.get("location")}
    ).json()

    return {
        **input_variables,
        'current_weather': weather_data
    }

Data Validation

Validate and sanitize inputs:

def pre_process(input_variables: dict) -> dict:
    from pydantic import BaseModel, EmailStr, validator

    class UserInput(BaseModel):
        email: EmailStr
        message: str

        @validator('message')
        def message_not_empty(cls, v):
            if not v.strip():
                raise ValueError('Message cannot be empty')
            return v.strip()

    # Validate input
    validated = UserInput(**input_variables)

    return {
        'email': validated.email,
        'message': validated.message
    }

Complex Parsing

Extract structured data from LLM output:

def post_process(llm_output: str, context: dict) -> dict:
    import json
    import re

    # Extract JSON embedded in markdown
    json_match = re.search(r'```json\n(.*?)\n```', llm_output, re.DOTALL)

    if json_match:
        try:
            structured_data = json.loads(json_match.group(1))
            return {
                'success': True,
                'data': structured_data,
                'raw_output': llm_output
            }
        except json.JSONDecodeError:
            pass

    return {
        'success': False,
        'error': 'Could not extract valid JSON',
        'raw_output': llm_output
    }

Chaining Multiple LLM Calls

Use integrations to chain prompts:

def post_process(llm_output: str, context: dict) -> dict:
    from mandatum import Mandatum

    client = Mandatum(api_key=context['api_key'])

    # First prompt generates draft
    draft = llm_output

    # Second prompt reviews and improves draft
    review_response = client.prompts.run(
        prompt_name="content-reviewer",
        input_variables={'draft': draft}
    )

    return {
        'original_draft': draft,
        'reviewed_content': review_response.output,
        'iterations': 2
    }

Testing Integrations

Local Testing

Test integrations locally before deploying:

# Test pre-processor
result = pre_process({
    'message': 'Test message',
    'user_id': '123'
})

assert result['message'] == 'Test message'
assert 'user_history' in result

Integration Tests

Run integration tests through the API:

test_result = client.integrations.test(
    prompt_name="customer-classifier",
    test_input={
        'message': 'I need help with billing'
    }
)

print(test_result)
# {
#   'pre_process_output': {...},
#   'llm_output': '...',
#   'post_process_output': {...},
#   'execution_time_ms': 1234
# }

Versioning

Integration code is versioned alongside prompts:

# Version 1
client.integrations.create(
    prompt_name="classifier",
    code=code_v1
)

# Version 2 (auto-created on update)
client.integrations.update(
    prompt_name="classifier",
    code=code_v2
)

# Run specific version
response = client.prompts.run(
    prompt_name="classifier",
    version=1,  # Uses integration version 1
    input_variables={...}
)

Learn more about versioning.

Security Best Practices

• Never hardcode secrets in integration code
• Use environment variables for API keys and sensitive data
• Validate all external API responses
• Set appropriate timeout values
• Handle errors gracefully
• Log security-relevant events

Monitoring

View integration execution logs:

1. Navigate to Logs in the sidebar
2. Filter by prompt name
3. Expand a request to see:
   • Pre-process execution time
   • Post-process execution time
   • Any errors or warnings

Next Steps

• View request logs
• Learn about advanced features
• Explore versioning
• Track analytics