morphe-adapter-sdk 1.0.3

Python adapter SDK for emitting events to the Morphe Control Plane

morphe-adapter-sdk

Python adapter SDK for emitting events to the Morphe Control Plane.

Installation

pip install morphe-adapter-sdk

Quick start

import asyncio
from morphe_adapter_sdk import MorpheClient

async def main():
    async with MorpheClient(token="your-token") as client:
        result = await client.emit({
            "event_type": "dataops.v1.sync_failed",
            "system_id":  "your-system-uuid",
            "payload":    {"records_failed": 42, "source": "warehouse"},
        })
        print(result)

asyncio.run(main())

Configuration

Parameter	Type	Default	Description
token	str	required	API token from your Morphe workspace
hmac_secret	str	None	Optional HMAC secret for request signing
validate_schema	bool	False	Validate payload against event schema before emit
endpoint	str	https://api.morphe.io	Override for self-hosted deployments
max_retries	int	3	Number of retries on transient network errors

Error handling

from morphe_adapter_sdk import MorpheClient
from morphe_adapter_sdk.errors import ApiError, NetworkError, SchemaValidationError

async with MorpheClient(token="your-token") as client:
    try:
        await client.emit(event)
    except SchemaValidationError as e:
        print("Schema errors:", e.errors)
    except ApiError as e:
        print(f"API error {e.status_code}: {e.message}")
    except NetworkError as e:
        print("Network error:", e)

License

MIT