KeborMed LEDSAS SDK (Python)

A Python client SDK for building LEDSAS (Local External Data Sources Action Service) data-processing services. You write async handler functions; the SDK consumes command messages from RabbitMQ, runs your handler, reads and writes payloads in Azure Blob Storage, and (optionally) publishes the response back to the caller.

It handles the plumbing — message consumption, acknowledgement, retries, timeouts, structured logging, health/readiness endpoints, and blob I/O — so your service code stays focused on the data transformation.

Features

• Handler-based programming model — register async handlers with a decorator.
• RabbitMQ consumption with prefetch, bounded concurrency, ack/nack semantics, and a dead-letter path for poison messages.
• Azure Blob Storage helpers: download_bytes/text/json/stream, upload_bytes/text/json, and memory-bounded streaming for large files.
• Typed models (Pydantic v2) for blob references and message envelopes.
• Built-in retries & timeouts with permanent vs. transient error classes.
• Health & readiness HTTP endpoints (/health, /ready, /livez, /readyz) with pluggable liveness/readiness checks.
• Structured JSON logging out of the box.
• Fully type-hinted (py.typed), Python 3.11+.

Installation

Requires Python 3.11 or newer.

pip install kbm-ledsas-sdk

To work from a clone — or before a release is published to PyPI — install from source:

git clone https://github.com/Kubyk-Health/kbm-ledsas-sdk.git
cd kbm-ledsas-sdk
pip install -e .            # add "[dev]" for the test/lint toolchain: pip install -e ".[dev]"

Quickstart

import logging
from kbm_ledsas_sdk import ServiceApp, errors

logging.basicConfig(level=logging.INFO)

app = ServiceApp(service_name="csv-processor")


@app.handler("ProcessCSV")
async def handle(ctx, req: dict) -> dict:
    csv_uri = req.get("csv_uri")
    if not csv_uri:
        # user_message is what the caller sees; the internal message goes to logs.
        raise errors.Permanent(
            "Validation: csv_uri missing",
            user_message="csv_uri is required",
        )

    text = await ctx.blob.download_text(csv_uri)
    container = csv_uri.replace("azblob://", "").split("/", 1)[0]

    out = await ctx.blob.upload_json(
        container=container,
        obj={"rows": len(text.splitlines()) - 1},
        # Use idempotency_key (not message_id) so a redelivery of the same logical
        # request overwrites the same blob. overwrite=True is required for that.
        path=f"result-{ctx.idempotency_key}.json",
        overwrite=True,
    )
    return {"result_uri": out.uri}


# Optional readiness hook (the SDK ships sensible defaults; this layers on top).
@app.readiness_check("warmup_done")
def _() -> bool:
    return True


if __name__ == "__main__":
    app.run()

Run it against a local RabbitMQ + Azure Blob (Azurite):

export KBM_LEDSAS_RABBITMQ_URL="amqp://guest:guest@127.0.0.1:5672/"
export KBM_LEDSAS_BLOB_CONN_STRING="DefaultEndpointsProtocol=http;AccountName=...;AccountKey=...;BlobEndpoint=http://127.0.0.1:10000/devstoreaccount1;"
python my_service.py

The service also exposes health endpoints at http://127.0.0.1:8090/health and /ready (loopback only by default; set KBM_LEDSAS_HEALTH_HOST=0.0.0.0 to bind all interfaces).

Local development

A ready-to-run stack (RabbitMQ + Azurite) is provided with the example service:

cd examples/hello_world_service
pip install -e ../..                                        # install the SDK from this repo
docker compose -f deploy/local/docker-compose.yml up -d     # start RabbitMQ + Azurite
python main.py                                              # run the example service

See examples/hello_world_service/ for a complete, runnable service (handler, sender script, tests, Dockerfile, and a local docker compose stack under deploy/local/).

Configuration

The SDK is configured through environment variables. The most common are below; see the API reference for the complete list (including KBM_LEDSAS_ALLOW_INSECURE_AMQP, KBM_LEDSAS_MAX_PAYLOAD_BYTES, and other advanced settings):

Documentation

• API Reference
• Example service
• Compatibility & supported versions
• Upgrade guide
• Changelog

Contributing

Contributions are welcome! Please read CONTRIBUTING.md first — it covers the development setup, how to run the tests, the coding style, and the Developer Certificate of Origin (DCO) sign-off we require on every commit. Also see our Code of Conduct.

Security

Please report vulnerabilities privately — see SECURITY.md. Do not open a public issue for security reports.

License

Licensed under the Apache License 2.0. Copyright © 2025-2026 KeborMed. See THIRD_PARTY_NOTICES.txt for third-party dependency licenses.

Maintained by KeborMed (the software's copyright holder) under the Kubyk-Health GitHub organization.