morphe-adapter-sdk 1.0.5

Python adapter SDK for emitting events to the Morphe OS

morphe-adapter-sdk

Python adapter SDK for emitting events to the Morphe OS.

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())

Event type format

event_type accepts two formats:

• Fully-qualified (always works): "namespace.version.event_name", e.g. "dataops.v1.sync_failed"
• Bare name (when event_namespace is configured): "sync_failed" — Morphe OS prepends the configured namespace automatically. Requires that system_id maps to a registered Morphe system with event_namespace set; if not, the bare name is stored unchanged.

When in doubt, use the fully-qualified form.

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