eka-webhook-sdk 0.1.0

Eka Care Webhook SDK for Python — process incoming webhook events with signature verification, automatic appointment data enrichment, schedule persistence, and one-shot reminder scheduling.

eka-webhook-sdk

Eka Care Webhook SDK for Python — process incoming webhook events from the Eka Care platform with signature verification, automatic appointment data enrichment, schedule persistence, and one-shot reminder scheduling.

This is the Python port of @eka-care/webhook-sdk. It provides the same feature set, the same on-the-wire behaviour, and the same eka_webhook_schedule_appointment table layout — so a Node service and a Python service can share the same backing table without conflict.

Features

• Signature verification — HMAC-SHA256 verification of incoming webhook requests (optional, but recommended)
• Automatic data enrichment — fetches full appointment, patient, doctor, and clinic details from Eka Care API for appointment events
• Multiple integration modes:
  • Standalone server — starts its own HTTP server, no framework needed
  • Flask view — drop-in view function for Flask apps
  • FastAPI route — drop-in async route handler for FastAPI apps
  • Framework-agnostic handler — works with Django, Starlette, aiohttp, Sanic, Tornado, Bottle, or any framework with sdk.handle_request()
• Schedule persistence — pluggable backend (MySQL or DynamoDB) that auto-creates a table on first use, writes a row for every bookable / followup appointment, and migrates the schema on subsequent runs
• Range queries with enrichment — find_appointments_between() returns full appointment + patient + doctor + clinic details for every booking in a given time window, split into normal vs followup
• Reminder scheduling — schedule_reminder() schedules a one-shot reminder via APScheduler and fires the callback with freshly fetched appointment context
• Type-friendly — full type hints, py.typed marker shipped
• Silent by default — debug=True opt-in to surface internal warnings/errors; otherwise the SDK runs silent
• Python 3.10+ — uses standard library hmac, http.server, concurrent.futures. Persistence and reminder backends are optional extras you only install if you opt in.

Supported Events

Event	Enrichment	Persistence behavior
appointment.created	Full appointment + patient + doctor + clinic details fetched	Inserts a schedule row when status = "BK" (booking) or (status = "IN" AND visit_type = "FLW") (followup)
appointment.updated	Full appointment + patient + doctor + clinic details fetched	Updates the existing row's status to whatever the new status is (cancellation, completion, no-show, anything)
prescription.created	Raw payload passed through	—
prescription.updated	Raw payload passed through	—

Installation

pip install eka-webhook-sdk

Optional extras

The SDK imports each of the optional backends lazily, so you only install what your code path actually uses:

Feature you use	Install
Standalone server, handle_request(), integrating with Django/Starlette/aiohttp/etc.	nothing extra
sdk.flask_view()	pip install eka-webhook-sdk[flask]
sdk.fastapi_route()	pip install eka-webhook-sdk[fastapi]
persistence=MySQLPersistenceConfig(...)	pip install eka-webhook-sdk[mysql] (uses PyMySQL)
persistence=DynamoDBPersistenceConfig(...)	pip install eka-webhook-sdk[dynamodb] (uses boto3)
sdk.schedule_reminder(...)	pip install eka-webhook-sdk[schedule] (uses APScheduler)
Everything	pip install eka-webhook-sdk[all]

If you forget one, the relevant SDK method raises a clear RuntimeError that names the missing package.

Quick Start

Option 1: Standalone Server (No Framework Needed)

The simplest way to start receiving webhooks. The SDK runs its own HTTP server backed by the standard-library http.server.

from eka_webhook_sdk import WebhookSDK

def on_event(event):
    print("Event type:", event.type)
    print("Appointment:", event.appointment_details)
    # Your business logic here:
    # - Save to database
    # - Send notifications
    # - Trigger workflows

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    api_key="your-api-key",          # optional
    signing_key="your-signing-key",  # optional, enables signature verification
    on_event=on_event,
)

server = sdk.listen(3000, on_listen=lambda: print("Webhook server listening on port 3000"))

# Block the main thread until Ctrl+C
import signal
signal.pause()

The standalone server provides:

• POST / — webhook endpoint
• GET /eka-webhook-health — health check endpoint ({"status": "ok"})

Option 2: Flask View

Drop into an existing Flask application. Use on_event for your business logic, or read result.event directly from handle_request().

from flask import Flask
from eka_webhook_sdk import WebhookSDK

app = Flask(__name__)

def on_event(event):
    print("Received:", event.type, event.appointment_details)
    # Save to DB, send notifications, etc.

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    signing_key="your-signing-key",
    on_event=on_event,
)

app.add_url_rule(
    "/webhook/eka",
    view_func=sdk.flask_view(),
    methods=["POST"],
)

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

Option 3: FastAPI Route

from fastapi import FastAPI
from eka_webhook_sdk import WebhookSDK

app = FastAPI()

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    signing_key="your-signing-key",
)

app.add_api_route(
    "/webhook/eka",
    sdk.fastapi_route(),
    methods=["POST"],
)

The SDK's core handler is synchronous (it makes blocking HTTP calls to the Eka Care API). The FastAPI adapter offloads it to a threadpool so the async event loop stays responsive.

Option 4: Generic Handler (Any Framework)

Use sdk.handle_request() directly in any Python framework. The result includes the enriched event object — no on_event callback needed.

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    signing_key="your-signing-key",
    # No on_event callback — use result.event instead
)

# In your route handler:
result = sdk.handle_request(request_body, request_headers)

if result.event:
    print(result.event.type)               # "appointment.created"
    print(result.event.appointment_details) # full appointment object
    print(result.event.payload)             # raw webhook payload

    # Your business logic here
    save_to_database(result.event.appointment_details)

return result.body, result.status

Framework Integration Examples

Django

# views.py
import json

from django.http import HttpResponse, JsonResponse
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from eka_webhook_sdk import WebhookSDK

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    signing_key="your-signing-key",
)

@csrf_exempt
@require_POST
def eka_webhook(request):
    try:
        body = json.loads(request.body.decode("utf-8"))
    except json.JSONDecodeError:
        return JsonResponse({"error": "Invalid JSON body"}, status=400)

    headers = {k.lower(): v for k, v in request.headers.items()}
    result = sdk.handle_request(body, headers)

    if result.event:
        # Your business logic here
        print(result.event.type, result.event.appointment_details)

    return JsonResponse(result.body, status=result.status)

# urls.py
from django.urls import path
from .views import eka_webhook

urlpatterns = [
    path("webhook/eka", eka_webhook),
]

Starlette / Plain ASGI

from starlette.applications import Starlette
from starlette.responses import JSONResponse
from starlette.routing import Route
from starlette.concurrency import run_in_threadpool

from eka_webhook_sdk import WebhookSDK

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    signing_key="your-signing-key",
)

async def webhook(request):
    try:
        body = await request.json()
    except Exception:
        return JSONResponse({"error": "Invalid JSON body"}, status_code=400)

    headers = {k.lower(): v for k, v in request.headers.items()}
    result = await run_in_threadpool(sdk.handle_request, body, headers)
    return JSONResponse(result.body, status_code=result.status)

app = Starlette(routes=[Route("/webhook/eka", webhook, methods=["POST"])])

aiohttp

from aiohttp import web
from eka_webhook_sdk import WebhookSDK

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    signing_key="your-signing-key",
)

async def webhook(request: web.Request) -> web.Response:
    try:
        body = await request.json()
    except Exception:
        return web.json_response({"error": "Invalid JSON body"}, status=400)

    headers = {k.lower(): v for k, v in request.headers.items()}

    loop = request.app.loop
    result = await loop.run_in_executor(None, sdk.handle_request, body, headers)
    return web.json_response(result.body, status=result.status)

app = web.Application()
app.router.add_post("/webhook/eka", webhook)
web.run_app(app, port=3000)

Persistence Layer (Schedule Storage)

When you pass a persistence config, the SDK transparently writes appointment-schedule rows for qualifying webhook events into the table eka_webhook_schedule_appointment. This is the foundation for find_appointments_between() and reminder workflows.

Why this exists

You typically need to remember which appointments are scheduled in the future so you can:

• Look up all appointments in a given time window (e.g. "everyone booked between 10am and 11am tomorrow")
• Drive reminder pipelines without having to re-query Eka Care for every webhook event
• Track cancellations / status changes locally

The persistence layer handles all of that without you writing any SQL or DynamoDB code.

Schema

The table has one row per appointment_id:

Column	Type	Notes
appointment_id	string	Primary key
start_time	epoch (seconds)	Indexed
end_time	epoch (seconds)
status	string	E.g. BK, IN, CK, CND, CNS, AB, etc.
visit_type	string	E.g. FLW for followup
created_at	epoch (seconds)	When the SDK first stored this row. Preserved across re-upserts

In MySQL the table is created with INDEX idx_start_time (start_time). In DynamoDB it has a Global Secondary Index named status_start_time_index (PK=status, SK=start_time, projecting visit_type).

When rows get written

Webhook	Condition	Action
appointment.created	status = "BK" (normal booking) or (status = "IN" AND visit_type = "FLW") (followup booked for today)	Upsert row
appointment.created	otherwise	No-op
appointment.updated	always (status not empty)	Update existing row's status to the new value. If no row exists for that appointment_id, log and skip

Configuration: MySQL

from eka_webhook_sdk import (
    WebhookSDK,
    MySQLPersistenceConfig,
    MySQLConnectionConfig,
)

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    persistence=MySQLPersistenceConfig(
        connection=MySQLConnectionConfig(
            host="localhost",
            port=3306,            # optional
            user="your-user",
            password="your-password",
            database="your-database",
        ),
    ),
)

The SDK uses PyMySQL and maintains an internal connection pool of size 10. Connections are health-checked with ping(reconnect=True) so they survive idle timeouts on the database side.

Configuration: DynamoDB

from eka_webhook_sdk import WebhookSDK, DynamoDBPersistenceConfig

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    persistence=DynamoDBPersistenceConfig(
        region="ap-south-1",

        # Optional. If omitted, the standard AWS credential chain is used
        # (env vars, instance role, ~/.aws/credentials, etc.).
        access_key_id="AKIA...",
        secret_access_key="...",

        # Optional. For DynamoDB Local during development:
        endpoint="http://localhost:8000",
    ),
)

The DynamoDB table is created with BillingMode=PAY_PER_REQUEST (no capacity to provision).

Lazy initialization

ensure_table() runs on the first webhook (or first call to find_appointments_between). It is cached so subsequent webhooks pay no overhead. If it fails, the cache is reset so the next webhook can retry. Persistence errors are logged and never fail the webhook response — the reminder layer is auxiliary; the webhook still returns 200.

Schema migration on existing tables

If the SDK starts up and the table already exists:

• MySQL: it queries INFORMATION_SCHEMA.COLUMNS and runs ALTER TABLE ... ADD COLUMN for every non-key column that is missing (e.g. created_at, added in a later release). Each missing column is added with NOT NULL DEFAULT 0 / DEFAULT '' so existing rows are backfilled safely.
• DynamoDB: non-key attributes don't need DDL (they appear on items at write time), but the GSI does. If status_start_time_index is missing, the SDK fires update_table to create it and returns immediately. GSI creation is asynchronous (can take several minutes); find_appointments_between returns errors until it is ACTIVE. Other operations (upsert, update_status) are unaffected.

This means upgrading the SDK on a project that already has the table is safe — the migration is automatic.

Querying Scheduled Appointments

sdk.find_appointments_between(start_time, end_time)

Find every booking whose start_time falls inside [start_time, end_time] (inclusive epoch range) and return the full appointment + patient + doctor + clinic details for each one.

result = sdk.find_appointments_between(
    start_time=1714521600,
    end_time=1714608000,
)
# result.appointments          → list of dicts (status === "BK")
# result.followup_appointments → list of dicts (status === "IN" AND visit_type === "FLW")

How it works:

1. Asks the persistence layer for the matching appointment_ids, split into normal bookings vs followups.
2. Calls get_appointment_details_by_id (via the Eka Care API) in parallel for every ID using a bounded thread pool.
3. Drops any IDs whose detail fetch fails (the failure is logged when debug=True).

from datetime import datetime, timezone

tomorrow_start = int(datetime(2026, 5, 9, tzinfo=timezone.utc).timestamp())
tomorrow_end   = tomorrow_start + 24 * 60 * 60

result = sdk.find_appointments_between(
    start_time=tomorrow_start,
    end_time=tomorrow_end,
)

print(
    f"{len(result.appointments)} bookings, "
    f"{len(result.followup_appointments)} followups"
)

Heads up on rate limits: a window with N matching appointments fans out to N parallel API calls (capped at 16 concurrent). For very large windows, throttle in the caller (or split the range) to stay under the Eka Care rate limits.

find_appointments_between raises if the SDK was constructed without a persistence config.

Reminder Scheduling

sdk.schedule_reminder(...)

Schedule a one-shot reminder for an appointment. When the trigger fires, the SDK fetches fresh appointment + patient + doctor + clinic details and invokes your callback with that context. After the callback runs (success or failure), the underlying job is automatically cancelled so it cannot fire again.

Requires APScheduler to be installed.

from datetime import datetime, timedelta
from eka_webhook_sdk import AppointmentReminderContext

start_time = datetime.fromtimestamp(appointment["start_time"])

def remind(ctx: AppointmentReminderContext) -> None:
    patient = ctx.patient_details or {}
    doctor = ctx.doctor_details or {}
    clinic = ctx.clinic_details or {}
    send_sms(
        patient.get("mobile"),
        f"Reminder: appointment with {doctor.get('firstname')} at {clinic.get('name')}",
    )

reminder = sdk.schedule_reminder(
    # Either a datetime (one-shot) OR a 5-/6-field cron expression
    cron_time=start_time - timedelta(minutes=5),
    appointment_id=appointment["appointment_id"],
    callback=remind,
)

# You can still cancel the reminder BEFORE its fire time, e.g. if the
# appointment is cancelled by the user:
reminder.cancel()

Behavior details

• Fresh data at fire time: the callback never receives stale data. Each fire re-fetches from Eka Care so reschedules / cancellations are reflected.
• One-shot: even if cron_time is a recurring cron expression, the SDK auto-cancels the job after the first fire. If you actually need recurring behavior, schedule a fresh reminder from inside your callback.
• Sync OR async callback: callback may be a regular function or a coroutine function — the SDK runs it correctly either way.
• In-memory only: jobs do not survive process restart. If you need durable reminders, persist (appointment_id, cron_time) yourself and re-schedule on startup. (The schedule persistence layer described above does not automatically re-hydrate reminders.)
• Errors are caught: a fetch failure or a throwing callback gets logged (when debug=True); the scheduler keeps running.

cron_time accepted formats

Form	Example	Meaning
datetime	datetime(2026, 5, 9, 14, 30)	One-shot fire at the given absolute time
5-field cron	"30 14 9 5 *"	minute hour day month day_of_week
6-field cron	"0 30 14 9 5 *"	second minute hour day month day_of_week

AppointmentReminderContext shape

@dataclass
class AppointmentReminderContext:
    appointment_details: dict[str, Any]
    patient_details: dict[str, Any] | None = None
    doctor_details: dict[str, Any] | None = None
    clinic_details: dict[str, Any] | None = None

patient_details / doctor_details / clinic_details are best-effort — if a sub-fetch fails, the corresponding field is None and the failure is logged.

API Reference

WebhookSDK

The main class. Construct it once and reuse across your application.

Constructor

WebhookSDK(
    *,
    client_id: str,
    client_secret: str,
    api_key: str | None = None,
    signing_key: str | None = None,
    allowed_events: list[str] | None = None,
    on_event: Callable[[WebhookEvent], None | Awaitable[None]] | None = None,
    path: str = "/",
    persistence: PersistenceConfig | None = None,
    debug: bool = False,
)

Parameter	Type	Required	Description
client_id	str	Yes	Eka Care Connect client ID
client_secret	str	Yes	Eka Care Connect client secret
api_key	str	No	Eka Care API key (improves rate limits)
signing_key	str	No	Webhook signing key. If provided, every request is verified via HMAC-SHA256. If omitted, signature verification is skipped.
allowed_events	list[str]	No	Event types to accept. Defaults to all supported events.
on_event	Callable[[WebhookEvent], None | Awaitable[None]]	No	Callback for processing events. Required for standalone mode. Optional for handle_request() / Flask view / FastAPI route — you can use result.event instead.
path	str	No	Webhook path for standalone server (default: "/"). Ignored for Flask/FastAPI adapters.
persistence	PersistenceConfig	No	MySQL or DynamoDB backend for schedule rows. See Persistence Layer.
debug	bool	No	When True, internal SDK warnings/errors are written to stdout. Default False (silent).

Example: initializing with everything

from eka_webhook_sdk import WebhookSDK, DynamoDBPersistenceConfig

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    api_key="your-api-key",
    signing_key="your-signing-key",
    allowed_events=["appointment.created", "appointment.updated"],
    on_event=lambda e: print("Event:", e.type),
    persistence=DynamoDBPersistenceConfig(region="ap-south-1"),
    debug=True,
)

sdk.handle_request(body, headers) -> WebhookResult

Framework-agnostic core handler. All other request-handling methods delegate to this.

Parameters:

• body — Parsed JSON body (dict) or raw JSON string.
• headers — Request headers as a plain key-value mapping. Names are matched case-insensitively; the SDK reads eka-webhook-signature.

Returns: WebhookResult.

When the request is processed successfully (status == 200), the event field contains the full WebhookEvent.

sdk.listen(port, on_listen=None) -> ThreadingHTTPServer

Starts a standalone threaded HTTP server. Returns the server immediately — the request loop runs on a background thread. Call server.shutdown() for graceful shutdown.

sdk.flask_view() -> Callable

Returns a Flask view function. Mount it on any Flask route::

app.add_url_rule("/webhook/eka", view_func=sdk.flask_view(), methods=["POST"])

sdk.fastapi_route() -> Callable

Returns a FastAPI async route handler. Mount it on any FastAPI app::

app.add_api_route("/webhook/eka", sdk.fastapi_route(), methods=["POST"])

sdk.find_appointments_between(*, start_time, end_time) -> FindAppointmentsBetweenResult

See Querying Scheduled Appointments. Raises RuntimeError if persistence was not configured.

sdk.schedule_reminder(*, cron_time, appointment_id, callback) -> Reminder

See Reminder Scheduling. Requires APScheduler to be installed.

Returns a Reminder object. Call reminder.cancel() to unschedule before the fire time.

sdk.get_appointment_details_by_id(appointment_id, *, partner_id=None)

Fetch appointment details directly from the Eka Care API using SDK credentials.

appointment = sdk.get_appointment_details_by_id(
    appointment_id="your-appointment-id",
)

sdk.get_patient_details_by_id(patient_id)

sdk.get_doctor_details_by_id(doctor_id)

sdk.get_clinic_details_by_id(clinic_id)

All four get_*_by_id methods authenticate against the Eka Care API using the SDK's configured credentials and return the API response as a dict.

WebhookEvent

The enriched event object. Available via result.event (from handle_request()) or as the argument to the on_event callback.

@dataclass
class WebhookEvent:
    type: str                                 # e.g. "appointment.created"
    payload: dict[str, Any]                   # Raw webhook payload
    appointment_details: dict[str, Any] | None = None
    patient_details: dict[str, Any] | None = None
    doctor_details: dict[str, Any] | None = None
    clinic_details: dict[str, Any] | None = None

WebhookResult

@dataclass
class WebhookResult:
    status: int                       # HTTP status code (200, 400, 403, 500, 502)
    body: dict[str, Any]              # JSON response body
    event: WebhookEvent | None = None # Enriched event (present when status == 200)
    error: str | None = None          # Error message (present when status != 200)

PersistenceConfig

Union of MySQLPersistenceConfig and DynamoDBPersistenceConfig.

from dataclasses import dataclass

@dataclass
class MySQLConnectionConfig:
    host: str
    user: str
    password: str
    database: str
    port: int | None = None

@dataclass
class MySQLPersistenceConfig:
    connection: MySQLConnectionConfig
    type: Literal["mysql"] = "mysql"

@dataclass
class DynamoDBPersistenceConfig:
    region: str
    access_key_id: str | None = None
    secret_access_key: str | None = None
    endpoint: str | None = None  # for DynamoDB Local
    type: Literal["dynamodb"] = "dynamodb"

AppointmentSchedule

The row shape stored in the persistence layer.

@dataclass
class AppointmentSchedule:
    appointment_id: str
    start_time: int
    end_time: int
    status: str
    visit_type: str

The constant SCHEDULE_TABLE_NAME ("eka_webhook_schedule_appointment") is also exported.

verify_signature(payload, signature_header, signing_key)

Standalone signature verification function. Exported for advanced use cases where you want to verify signatures without the full SDK pipeline.

from eka_webhook_sdk import verify_signature

result = verify_signature(request_body, signature_header_value, signing_key)
# result.valid → True/False
# result.reason → human-readable reason if not valid

WebhookProcessingError

Custom error class raised during webhook processing. Includes an HTTP status_code.

from eka_webhook_sdk import WebhookProcessingError

try:
    result = sdk.handle_request(body, headers)
except WebhookProcessingError as err:
    print(err.status_code, err)

In normal usage you do not need to catch this — handle_request() already turns it into a WebhookResult with the appropriate status.

Constants

from eka_webhook_sdk import (
    SUPPORTED_EVENTS,
    APPOINTMENT_EVENTS,
    SCHEDULE_TABLE_NAME,
)

# SUPPORTED_EVENTS = (
#   "appointment.created", "appointment.updated",
#   "prescription.created", "prescription.updated",
# )
# APPOINTMENT_EVENTS = ("appointment.created", "appointment.updated")
# SCHEDULE_TABLE_NAME = "eka_webhook_schedule_appointment"

on_event Callback vs result.event

The SDK provides two ways to access the processed webhook data:

Approach	When to use
on_event callback	Standalone mode (sdk.listen()), where the SDK owns the HTTP server and you don't control the request/response cycle directly. Also works with Flask/FastAPI if you prefer the callback pattern.
result.event	handle_request() and any framework integration where you control the route handler. The enriched WebhookEvent is returned directly in the result — no callback needed.

Both approaches can be used together (the callback fires first, then result.event is available in the return value), but typically you'll use one or the other.

Signature Verification

When you provide a signing_key in the SDK configuration, every incoming webhook request is verified using HMAC-SHA256:

1. The SDK reads the Eka-Webhook-Signature header from the request.
2. The header format is: t=<unix_timestamp>,v1=<hex_signature>.
3. The signed payload is constructed as: <timestamp>.<json_serialised_body>, using JSON with no whitespace separators (matching JavaScript's JSON.stringify default).
4. The expected signature is computed: HMAC-SHA256(signing_key, signed_payload).
5. The signatures are compared using hmac.compare_digest (constant-time, prevents timing attacks).

If verification fails, the SDK returns a 403 response and does not invoke the on_event callback or write to persistence.

Disabling Signature Verification

Simply omit signing_key:

sdk = WebhookSDK(
    client_id="your-client-id",
    client_secret="your-client-secret",
    # No signing_key = signature verification disabled
)

Request Lifecycle

Eka Care Platform
    |
    v  POST (with JSON body + Eka-Webhook-Signature header)
Your Server / Standalone SDK Server
    |
    v  WebhookSDK.handle_request(body, headers)
    |
    +-- 1. Parse body (if string)
    +-- 2. Verify signature (if signing_key configured)
    |      |-- FAIL -> return 403
    |      |-- PASS -> continue
    +-- 3. Validate event type against allowed_events
    |      |-- NOT ALLOWED -> return 400
    |      |-- ALLOWED -> continue
    +-- 4. For appointment events:
    |      |-- Authenticate with Eka Care API (auto-managed by SDK)
    |      |-- Fetch appointment details by ID
    |      |-- Fetch patient/doctor/clinic details (best-effort)
    |      |-- Attach to WebhookEvent.{appointment_details, ...}
    +-- 5. For prescription events:
    |      |-- Pass through raw payload (no API enrichment)
    +-- 6. If persistence is configured (and event is appointment.*):
    |      |-- Lazily ensure_table() on first call (creates / migrates schema)
    |      |-- appointment.created with status=BK or (IN+FLW) -> upsert row
    |      |-- appointment.updated -> update_status on existing row
    |      |-- Errors are LOGGED, do NOT fail the webhook response
    +-- 7. Invoke on_event(webhook_event) callback (if provided)
    |      |-- THROWS -> return 500
    |      |-- OK -> continue
    +-- 8. Return 200 with success response + result.event

Debug Logging

By default the SDK is silent — it does not write any warnings or errors to stdout. This is to keep the host application's logs clean. Errors still propagate normally to your callers and to HTTP responses; only the console output is suppressed.

To turn internal logging on, pass debug=True:

sdk = WebhookSDK(
    client_id="...",
    client_secret="...",
    debug=True,
)

When enabled, you will see messages like:

• [ERROR] WebhookSDK: failed to fetch patient details for ...
• [ERROR] WebhookSDK: MySQL upsert failed for appointment_id=... : <error>
• WebhookSDK: DynamoDB table eka_webhook_schedule_appointment is missing GSI status_start_time_index; submitting UpdateTable. ...
• [WARN] WebhookSDK: appointment.updated missing appointment_id; skipping update

The flag is process-global. If you construct multiple WebhookSDK instances in the same process with different debug values, the last one wins.

Error Handling

The SDK handles errors at each stage and returns appropriate HTTP status codes:

Status	Cause
200	Webhook processed successfully (persistence errors do not affect this)
400	Invalid JSON body, missing fields, or unsupported event type
403	Signature verification failed
500	Error in on_event callback or unhandled exception
502	Failed to fetch appointment details from Eka Care API

Errors are returned in the response (WebhookResult.error) and — when debug=True — also printed to stdout.

Environment Variables

The examples use environment variables for configuration. You can set them however fits your deployment:

Variable	Description
EKA_CLIENT_ID	Eka Care Connect client ID
EKA_CLIENT_SECRET	Eka Care Connect client secret
EKA_API_KEY	Eka Care API key (optional)
EKA_SIGNING_KEY	Webhook signing key (optional)
PORT	Server port (default: 3000)

Building from Source

# Install in editable mode with dev tools
pip install -e ".[dev,all]"

# Run tests
pytest

# Type check
mypy eka_webhook_sdk

# Lint
ruff check eka_webhook_sdk

Project Structure

webhook-python-package/
├── eka_webhook_sdk/
│   ├── __init__.py             # Public API exports
│   ├── types.py                # Public dataclasses, TypedDicts, constants
│   ├── exceptions.py           # WebhookProcessingError
│   ├── signature.py            # HMAC-SHA256 signature verification
│   ├── eka_api.py              # Thin HTTP client for Eka Care endpoints + token cache
│   ├── enrichment.py           # Reusable appointment + patient + doctor + clinic fetcher
│   ├── webhook_consumer.py     # Webhook validation + Eka Care enrichment for incoming events
│   ├── webhook_sdk.py          # Main WebhookSDK class
│   ├── scheduler.py            # APScheduler wrapper for one-shot reminders
│   ├── standalone_server.py    # Built-in HTTP server (standalone mode)
│   ├── logger.py               # Internal logger gated by the `debug` config
│   ├── adapters/
│   │   ├── __init__.py
│   │   ├── flask.py            # Flask view adapter
│   │   └── fastapi.py          # FastAPI route adapter
│   └── persistence/
│       ├── __init__.py
│       ├── base.py             # PersistenceConfig + SchedulePersistence interface
│       ├── factory.py          # Picks adapter based on config type
│       ├── mysql.py            # MySQL adapter (PyMySQL, lazy import)
│       ├── dynamodb.py         # DynamoDB adapter (boto3, lazy import)
│       └── schedule.py         # Schedule decision logic (when to insert/update)
├── examples/
│   ├── standalone.py           # Standalone server example
│   ├── flask_app.py            # Flask integration example
│   ├── fastapi_app.py          # FastAPI integration example
│   └── django_view.py          # Django view example
├── pyproject.toml
├── README.md
└── LICENSE

Requirements

• Python >= 3.10
• An Eka Care Connect account with client_id and client_secret
• (Optional) PyMySQL or boto3 if you opt into persistence
• (Optional) APScheduler if you call schedule_reminder()
• (Optional) Flask if you use flask_view()
• (Optional) fastapi if you use fastapi_route()

Equivalence with the Node SDK

Node (@eka-care/webhook-sdk)	Python (eka-webhook-sdk)
mysql2	PyMySQL
@aws-sdk/client-dynamodb	boto3
node-schedule	APScheduler
express	Flask (and FastAPI)
Built-in http standalone server	Built-in http.server standalone server
JSON.stringify(body) for signature canonicalisation	json.dumps(body, separators=(",", ":"), ensure_ascii=False)
eka_webhook_schedule_appointment table	Same table — both SDKs are wire-compatible
WebhookSDK.scheduleReminder()	WebhookSDK.schedule_reminder()
WebhookSDK.findAppointmentsBetween()	WebhookSDK.find_appointments_between()
appointmentDetails, patientDetails, ...	appointment_details, patient_details, ...

The SDKs share the same table layout, the same signature scheme, and the same Eka Care API contracts — you can mix and match Node and Python services pointing at the same persistence layer without conflict.

License

MIT