IXT Python SDK

Python SDK for building IXT modules that run as separate processes and communicate with the IXT runtime via IPC.

Features

• ✅ Simple base class for module development
• ✅ Automatic IPC connection handling
• ✅ JSON message parsing and framing
• ✅ Lifecycle management (init, handle, shutdown, health_check)
• ✅ Base64 payload encoding/decoding helpers
• ✅ Error handling and logging
• ✅ Type hints for better IDE support
• ✅ Zero external dependencies (Python 3.8+)

Installation

# From source
cd sdk/ixt-sdk-python
pip install -e .

# Or copy ixt_process_module.py to your project
cp sdk/ixt-sdk-python/ixt_process_module.py my-module/

Quick Start

1. Create Your Module

#!/usr/bin/env python3
from ixt_process_module import IxtProcessModule

class MyModule(IxtProcessModule):
    def on_init(self, config):
        """Initialize module with configuration"""
        self.db_url = config.get('database_url', 'sqlite:///data.db')
        self.log(f"Connecting to database: {self.db_url}")
        # Initialize your resources here

    def on_handle(self, message, context):
        """Handle incoming messages"""
        topic = message['topic']
        payload_data = self.decode_payload(message['payload'])
        
        if topic == 'my.service.process':
            result = self.process_data(payload_data)
            return self.success({
                'result': self.encode_payload(result)
            })
        elif topic == 'my.service.query':
            data = self.query_data(payload_data)
            return self.success({
                'data': self.encode_payload(data)
            })
        else:
            return self.error('UNKNOWN_TOPIC', f'Unknown topic: {topic}')

    def on_shutdown(self):
        """Clean up resources"""
        self.log("Shutting down module")
        # Close connections, save state, etc.

    def on_health_check(self):
        """Report health status (optional)"""
        return {
            'healthy': True,
            'database_connected': self.check_db_connection(),
            'requests_processed': self.request_count
        }

if __name__ == '__main__':
    MyModule().run()

2. Create Manifest

# manifest.toml
schema_version = "3.0"

[module]
name = "my-module"
version = "1.0.0"
description = "My IXT module"
author = "Your Name"
license = "MIT"
runtime_targets = ["Process"]

[provides]
services = ["my.service.process", "my.service.query"]

[config]
database_url = { type = "string", default = "sqlite:///data.db" }

3. Run Your Module

The IXT runtime will spawn your module automatically. For development testing:

# Runtime sets these environment variables:
export IXT_IPC_SOCKET=/tmp/ixt-module-abc123.sock
export IXT_MODULE_ID=my-module:1.0.0/instance-001

# Run your module
python3 my_module.py

API Reference

IxtProcessModule Base Class

Abstract base class for IXT modules. Implement these methods:

on_init(config: Dict[str, Any]) -> None

Initialize your module with configuration from the manifest.

def on_init(self, config):
    self.setting = config.get('my_setting', 'default')
    # Initialize resources, connect to services, etc.

on_handle(message: Dict[str, Any], context: Dict[str, Any]) -> Dict[str, Any]

Handle incoming messages. This is where your business logic goes.

def on_handle(self, message, context):
    topic = message['topic']
    payload = self.decode_payload(message['payload'])
    
    # Process the message
    result = self.process(payload)
    
    # Return success or error
    return self.success({'result': self.encode_payload(result)})

Message structure:

{
    'topic': 'service.name',           # Service topic
    'payload': 'base64-encoded-data',  # Base64-encoded payload
    'priority': 1,                     # Priority level
    'ttl_ms': 5000                     # Time-to-live in milliseconds
}

Context structure:

{
    'trace_id': 'unique-trace-id',     # Distributed tracing ID
    'source_module': 'caller:1.0.0',   # Calling module
    'timestamp': 1234567890            # Unix timestamp
}

on_shutdown() -> None

Gracefully shutdown your module.

def on_shutdown(self):
    self.db.close()
    self.save_state()

on_health_check() -> Dict[str, Any] (optional)

Report health status. Default returns {'healthy': True}.

def on_health_check(self):
    return {
        'healthy': self.is_healthy(),
        'uptime_seconds': time.time() - self.start_time
    }

Helper Methods

success(data: Dict = None) -> Dict

Create a success response.

return self.success({'result': 42})

error(code: str, message: str, details: Dict = None) -> Dict

Create an error response.

return self.error('INVALID_INPUT', 'Missing required field: name')

encode_payload(data: bytes) -> str

Encode binary data as base64 string.

encoded = self.encode_payload(b"hello world")

decode_payload(payload: str) -> bytes

Decode base64 payload to binary.

data = self.decode_payload(message['payload'])
text = data.decode('utf-8')

log(message: str)

Log informational message (to stderr, captured by runtime).

self.log(f"Processing request {request_id}")

log_error(message: str)

Log error message.

self.log_error(f"Failed to connect: {error}")

Message Flow

IXT Runtime                    Your Module
     |                              |
     |----(1) init message--------->|
     |<---(2) success response------|
     |                              |
     |----(3) handle message------->|
     |<---(4) success/error---------|
     |                              |
     |----(5) handle message------->|
     |<---(6) success/error---------|
     |                              |
     |----(7) shutdown message----->|
     |<---(8) success response------|
     |                              |

Examples

See examples/python-echo-module/ for a complete working example.

Testing

# Install dev dependencies
pip install -e ".[dev]"

# Run tests
pytest tests/

# Type checking
mypy ixt_process_module.py

# Code formatting
black ixt_process_module.py

Requirements

• Python 3.8 or higher
• No external dependencies (uses only Python standard library)

License

MIT License - see LICENSE file for details.

Support

• Documentation: https://www.ixtaat.com
• Issues: https://git.h-dv.de/h-dv/ixt/issues
• Community: https://discord.gg/ixt-dev