sibi-flux 2025.12.0

Sibi Toolkit: A collection of tools for Data Analysis/Engineering.

SibiFlux

SibiFlux is a production-grade resilient data engineering ecosystem designed to bridge the gap between local development, distributed computing, and agentic AI workflows. It provides a unified engine for hybrid data loading (batch + streaming), self-healing distributed operations, and native interfaces for AI agents via the Model Context Protocol (MCP).

graph TD
    subgraph "Agentic Interface (MCP)"
        Agent["AI Agent / Claude"] <--> Router["MCP Router"]
        Router <--> Resources["SibiFlux Resources"]
    end

    subgraph "Solutions Layer (Business Logic)"
        Logistics["Logistics Solutions"]
        Enrichment["Enrichment Pipelines"]
        Cubes["DataCubes"]
    end

    subgraph "SibiFlux Core Engine"
        DfHelper["DfHelper (Unified Loader)"]
        Cluster["Resilient Dask Cluster"]
        Managed["ManagedResource Lifecycle"]
    end

    Resources --> Cubes
    Logistics --> DfHelper
    Cubes --> DfHelper
    DfHelper --> Cluster

Core Architecture

1. The Flux Engine (sibi_flux)

The foundational library providing resilient distributed primitives.

• DfHelper: A unified API for loading data from SQLAlchemy, Parquet, or HTTP sources into Pandas or Dask DataFrames.
• Dataset: A high-level abstraction for hybrid data loading, seamlessly merging historical (Parquet) and live (SQL) data sources.
• DfValidator: A robust schema enforcement tool that validates DataFrames against strict type maps and generates ClickHouse DDL.
• ArtifactOrchestrator: An async engine for managing concurrent artifact updates with retries, backoff, and worker isolation.
• ManagedResource: A rigorous lifecycle management system for async resources, ensuring clean shutdown, signal handling, and observability.
• Dask Cluster: A self-healing distributed runtime that detects worker failures, manages re-connection, and enforces "The Nuclear Option" for test isolation.

2. Agentic Interface (mcp)

Native support for the Model Context Protocol (MCP), allowing AI agents to directly interact with SibiFlux data structures.

• Expose DataCubes: Automatically turn any DataCube into a queryable MCP Resource.
• Tooling: Register Python functions as tools callable by agents.

Key Capabilities

Hybrid Data Loading

SibiFlux implements a "Hot/Cold" architecture for seamless data access:

• Historical Data: Read efficiently from partitioned Parquet archives (S3/Local).
• Live Data: Query operational SQL databases for real-time changes.
• Automatic Merge: DfHelper and Dataset automatically stitch these sources together, handling schema evolution and deduplication.

Resilient Distributed Compute

The SibiFlux Dask wrapper provides:

• Auto-Healing: Clients that automatically reconnect if the scheduler dies.
• Safe Persistence: Wrappers like safe_persist that retry operations on network jitter.
• Smart Partitioning: Automated repartitioning to prevent "small file problem" in Parquet outputs.

Observability

Built-in integration with OpenTelemetry (OTel):

• structured logging with correlation IDs.
• distributed tracing across async boundaries.

Quick Start

Installation

# Base installation (Core Engine only)
pip install sibi-flux

# For Distributed Computing (Dask Cluster support)
pip install "sibi-flux[distributed]"

# For Geospatial capabilities (OSMnx, GeoPandas, etc.)
pip install "sibi-flux[geospatial]"

# For MCP Agentic Interface support
pip install "sibi-flux[mcp]"

# Complete installation (Simulates the "All-in-One" environment)
pip install "sibi-flux[complete]"

API Examples

1. Data Loading (Dataset)

The Dataset class provides a high-level abstraction for hybrid loading.

from sibi_flux import Dataset
from solutions.logistics.readers.products import ProductsParquetReader, ProductsSqlReader

class ProductsDataset(Dataset):
    historical_reader = ProductsParquetReader
    live_reader = ProductsSqlReader
    date_field = "created_at"

# Load hybrid data (merges Parquet + SQL)
ds = ProductsDataset(start_date="2023-01-01", end_date="2023-01-31")
df = await ds.aload()  # Returns a Dask DataFrame

2. Schema Validation (DfValidator)

Ensure your data meets strict type requirements and generate DDL for ClickHouse.

from sibi_flux import DfValidator

# Define expected schema
SCHEMA = {
    "product_id": "Int64[pyarrow]",
    "price": "Float64[pyarrow]",
    "name": "string[pyarrow]"
}

validator = DfValidator(df)
validator.validate_schema(SCHEMA)
validator.standardize_data_quality()

# Generate ClickHouse DDL
ddl = validator.generate_clickhouse_ddl("products_table")
print(ddl)

3. Data Enrichment (AsyncDfEnricher)

Enrich a base DataFrame with data from other sources using AttachmentSpec.

from sibi_flux.df_enricher import AsyncDfEnricher, AttachmentSpec

specs = [
    AttachmentSpec(
        key="customer_info",
        required_cols={"customer_id"},
        attachment_fn=fetch_customer_data,  # Async function returning DF
        left_on=["customer_id"],
        right_on=["id"],
        drop_cols=["id"]
    )
]

enricher = AsyncDfEnricher(base_df=orders_df, specs=specs)
enriched_df = await enricher.enrich()

4. Orchestration (ArtifactOrchestrator)

Manage concurrent updates of multiple artifacts with retries and worker isolation.

from sibi_flux.orchestration import ArtifactOrchestrator

orchestrator = ArtifactOrchestrator(
    wrapped_classes={"daily": [ProductsDataset, OrdersDataset]},
    max_workers=3,
    retry_attempts=3
)

# Updates all 'daily' artifacts concurrently
results = await orchestrator.update_data("daily")

5. Distributed Compute (Dask Cluster)

Execute resilient operations that survive scheduler restarts.

from sibi_flux.dask_cluster import safe_compute, get_persistent_client

client = get_persistent_client()

# Safe compute with auto-retry logic
result = safe_compute(df.groupby("category").sum())

6. Resource Management (ManagedResource)

Create lifecycle-safe components.

from sibi_flux.core import ManagedResource

class MyResource(ManagedResource):
    async def _acleanup(self):
        await self.db_connection.close()
        self.logger.info("Cleaned up!")

async with MyResource() as res:
    await res.do_work()
# Automatically cleaned up here

7. Agentic Interface (sibi_flux.mcp)

Seamlessly bridge your data with AI Agents.

Server Side (Expose Resources)

from sibi_flux.mcp import BaseMCPRouter
from products import ProductsDataset

# Create an MCP Router compatible with FastAPI
router = BaseMCPRouter(name="data-server")

# Automatically register a Dataset as an MCP Resource
# Agent can now read `sibi://ProductsDataset`
router.register_cube_resource(ProductsDataset)

# Register a custom tool
@router.tool()
def calculate_vat(amount: float) -> float:
    return amount * 0.2

Client Side (Consume Resources)

from sibi_flux.mcp import GenericMcpClient

# Connect to the MCP Server
async with GenericMcpClient(url="http://localhost:8000/sse") as client:
    # Read the resource (returns JSON data from the Dataset)
    data = await client.read_resource("sibi://ProductsDataset")
    
    # Call a tool
    vat = await client.call_tool("calculate_vat", arguments={"amount": 100.0})