IXT Python SDK

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

Features

• Simple base class for module development
• Cap'n Proto serialization for efficient IPC
• Automatic IPC connection handling
• Lifecycle management (init, handle, shutdown, health_check)
• Base64 payload encoding/decoding helpers
• Error handling and logging
• Type hints for IDE support

Installation

# Install dependencies
pip install pycapnp>=1.0.0

# Install SDK
cd sdk/ixt-sdk-python
pip install -e .

Quick Start

1. Create Your Module

#!/usr/bin/env python3
from ixt_sdk 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}")

    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)})
        else:
            return self.error('UNKNOWN_TOPIC', f'Unknown topic: {topic}')

    def on_shutdown(self):
        """Clean up resources"""
        self.log("Shutting down module")

    def on_health_check(self):
        """Report health status"""
        return {'healthy': True}

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"
runtime_targets = ["Process"]

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

3. Run Your Module

The IXT runtime spawns your module automatically. For development:

export IXT_IPC_SOCKET=/tmp/ixt-module.sock
export IXT_MODULE_ID=my-module:1.0.0/instance-001
python3 my_module.py

API Reference

IxtProcessModule Base Class

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

Initialize with configuration from the manifest.

on_handle(message: Dict, context: Dict) -> Dict

Handle incoming messages. Returns success or error response.

on_shutdown() -> None

Gracefully shutdown the module.

on_health_check() -> Dict (optional)

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

Helper Methods

• success(data: Dict = None) - Create success response
• error(code: str, message: str) - Create error response
• encode_payload(data: bytes) -> str - Encode binary as base64
• decode_payload(payload: str) -> bytes - Decode base64 to binary
• log(message: str) - Log to stderr

Requirements

• Python 3.8+
• pycapnp >= 1.0.0

License

MIT License