Railengine Ingestion SDK

Python SDK for ingesting data into Railtown AI Railengine - handles embeddings, storage, indexing, and webhook publishing.

Overview

The rail-engine-ingest package provides a Pythonic interface for publishing data to Railengine. It supports async/await patterns, Pydantic model validation, and familiar configuration patterns.

Installation

uv pip install rail-engine-ingest

Quick Start

# Copy your ENGINE_TOKEN from https://cndr.railtown.ai Project Settings

import asyncio
import os
from railtown.engine.ingest import RailengineIngest
import base64
import json

engine_token = os.getenv("ENGINE_TOKEN")

async def main():
    # Initialize client
    async with RailengineIngest(engine_token=engine_token) as client:
        # Ingest data
        data = {
            "name": "Chicken Shawarma",
            "meal": "lunch",
            "calories": 95
        }
        response = await client.upsert(data)
        print(f"Status: {response.status_code}")

asyncio.run(main())

Configuration

Environment Variable

• ENGINE_TOKEN - Base64-encoded JSON string containing ingestion credentials

Constructor Parameters

• engine_token (optional) - ENGINE_TOKEN string (if not provided, reads from env)
• model (optional) - Pydantic model type for validating ingested data
• timeout (optional) - Request timeout in seconds or httpx.Timeout (default 30)
• verify (optional) - TLS verification. Defaults to True. Pass a CA bundle path or ssl.SSLContext for custom trust. False disables verification and must be set explicitly.
• http_client (optional) - Caller-owned httpx.AsyncClient for proxies, custom transports, or tests. The SDK does not close this client.

Earlier SDK versions disabled TLS verification. 0.4.0 verifies certificates by default. Use verify= with your CA bundle if the ingestion host uses a private certificate.

Features

• Single upsert() method - Unified interface for ingesting data
• Multiple data formats - Accepts Pydantic models, dictionaries, or JSON strings
• Model validation - Optional Pydantic model validation
• Direct JSON payload - Sends data directly as JSON to the ingestion endpoint
• Webhook handler - get_webhook_handler() returns a payload parser when the client is constructed with model=
• Async/await support - Built for modern async Python applications

Error Handling

The SDK provides custom exception classes:

• RailtownError - Base exception
• RailtownBadRequestError - 400 Bad Request
• RailtownUnauthorizedError - 401 Unauthorized
• RailtownNotFoundError - 404 Not Found
• RailtownConflictError - 409 Conflict
• RailtownServerError - 5xx Server errors

Exceptions are raised on errors.

Requirements

• uv
• Python 3.11+
• httpx >= 0.24.0
• pydantic >= 1.10.0

Related package

For retrieving and searching data from Railengine, install and use rail-engine.

License

MIT