Python SDK

Install

The package supports Python 3.11 through 3.14. Install the published version:

python -m pip install morphiaone==0.2.2

The SDK is typed and depends on httpx for HTTP requests.

Keep credentials on your server

Set the key and API base URL in the server process, using your secret manager or deployment environment:

export MORPHIAONE_API_KEY="pk_test_..."   # the key you created in the Platform
export MORPHIAONE_BASE_URL="https://api.morphiaone.ai"

Never put the API key in a browser bundle, mobile app, or client-side configuration. Call MorphiaOne from your backend; let your frontend call your backend. See Authentication and the backend-for-frontend pattern.

Create one MorphiaOne client for a short script with with, as in the quickstart. For a long-running service, reuse a client for the service lifespan and call client.close() during shutdown.

Quickstart: upload and analyze

Each product request follows the same path: upload an image as an Asset, submit a job that refers to its id, then wait for the job and read its output. This complete server-side example expects a JPEG, PNG, or WebP path as its first command-line argument. Replace the synthetic idempotency key with one derived from your own durable order or request id. Save the example as quickstart.py, set the two environment variables above, then run python quickstart.py ./face.jpg.

import os
import sys
from pathlib import Path
from morphiaone import MorphiaOne, SkinAnalysisOutput
image_path = Path(sys.argv[1]) if len(sys.argv) > 1 else Path("face.jpg")
with MorphiaOne(
    api_key=os.environ["MORPHIAONE_API_KEY"],
    base_url=os.environ["MORPHIAONE_BASE_URL"],
    operation_timeout=240,
) as client:
    source = client.assets.upload(image_path)
    job = client.skin.analyze(
        asset_id=source.id,
        concerns=["acne", "redness"],
        visualizations=[{"concern": "acne", "kind": "mask"}],
        idempotency_key="example-order-48213-skin",
    )
    # This example prints the handle; production code should persist it before waiting.
    print(f"job_id={job.id}")
    done = client.jobs.wait(job.id, timeout=240)
    if not isinstance(done.output, SkinAnalysisOutput):
        raise RuntimeError("skin analysis returned an unrecognized output type")
    for result in done.output.concerns:
        print(result.concern, result.status, result.score)
    for visualization in done.output.visualizations:
        image = client.assets.get_content(visualization)
        Path("acne-mask.png").write_bytes(image.content)

assets.upload accepts a path, Path, or raw bytes. It infers image/jpeg, image/png, or image/webp from the filename or image bytes; pass media_type when needed. The SDK rejects uploads larger than its default 10 MiB max_image_bytes limit before sending them. Uploads have no idempotency key, so an ambiguous retry can create an extra input Asset; use the returned expires_at to know its lifetime.

Product submissions are asynchronous. The submit returns a Job; jobs.wait polls until it settles and returns a succeeded job. A failed job raises ProcessingError. The job output contains typed result data and references to result Assets, not image bytes. Fetch result bytes with assets.get_content before the Asset’s expires_at. That method uses the configured API origin and key rather than following href as an arbitrary URL.

Skin analysis

skin.analyze accepts the source Asset id, the concerns to analyze, and optional visualizations. It returns exactly the concerns and visualizations requested. For a measured concern, score is from 0 to 10 and higher means the concern is less visible. An unavailable result has no score; do not treat it as zero. Skin analysis is cosmetic, not a medical diagnosis, and score semantics remain draft.

The result is a SkinAnalysisOutput with analysis_version, concerns, and visualizations. Keep analysis_version with any score you store: scores are only comparable within the same version.

Makeup try-on

makeup.try_on takes one source Asset and an ordered list of products. The order is the rendering layer order. Supported categories are foundation, lipstick, and lip_liner; include each category at most once. A product has a #RRGGBB color and optional opacity from 0 to 1 (default 1). Replace face.jpg with a path available to your server.

import os
from pathlib import Path
from morphiaone import MakeupTryOnOutput, MorphiaOne
with MorphiaOne(
    api_key=os.environ["MORPHIAONE_API_KEY"],
    base_url=os.environ["MORPHIAONE_BASE_URL"],
) as client:
    source = client.assets.upload("face.jpg")
    job = client.makeup.try_on(
        asset_id=source.id,
        products=[
            {"category": "lip_liner", "color": "#7B2E42", "opacity": 0.8},
            {"category": "lipstick", "color": "#C2185B"},
        ],
        idempotency_key="example-session-91c4-makeup",
    )
    done = client.jobs.wait(job.id)
    if isinstance(done.output, MakeupTryOnOutput):
        image = client.assets.get_content(done.output.image)
        Path("makeup.png").write_bytes(image.content)

Hair color try-on

Hair try-on finds the hair in one source image and recolors only that. It does not change hairstyle or hair length. If you already segment hair yourself, upload your grayscale mask as a second Asset (same dimensions; zero is background, nonzero is hair) and pass hair_mask_asset_id; the server then uses it instead.

Provide a dark and light color stop. Each stop takes an #RRGGBB color and optional opacity from 0 to 1 (default 1). Replace both file paths with files available to your server:

import os
from morphiaone import HairTryOnOutput, MorphiaOne
with MorphiaOne(
    api_key=os.environ["MORPHIAONE_API_KEY"],
    base_url=os.environ["MORPHIAONE_BASE_URL"],
) as client:
    source = client.assets.upload("face.jpg")
    job = client.hair.try_on(
        source_asset_id=source.id,
        color_gradient={
            "dark": {"color": "#3B1F16", "opacity": 0.9},
            "light": {"color": "#C87943", "opacity": 0.8},
        },
        idempotency_key="example-order-48213-hair",
    )
    done = client.jobs.wait(job.id)
    if isinstance(done.output, HairTryOnOutput):
        image = client.assets.get_content(done.output.image)

Hair is a draft capability. See Hair color try-on for its limits and input requirements.

Jobs and outputs

Use client.jobs.get(job_id) to read the latest job or client.jobs.wait(job_id) to poll until it succeeds. You can also pass the returned Job object to wait. Persist the job id before waiting; a local timeout does not cancel work on the server. Job states are queued, running, succeeded, and failed.

Known outputs decode as SkinAnalysisOutput, MakeupTryOnOutput, or HairTryOnOutput. An output type added by the API that this SDK does not know becomes UnknownJobOutput and remains available in its raw data. Check the output type before reading product-specific fields.

wait=True on skin.analyze, makeup.try_on, or hair.try_on waits during the submission call and returns the succeeded job. Without wait=True, each submission returns as soon as the API accepts or replays the job.

Async client

AsyncMorphiaOne supports the same Assets, Jobs, products, and Usage methods. Every call is a coroutine; use async with to close its HTTP client:

import asyncio
import os
from morphiaone import AsyncMorphiaOne, SkinAnalysisOutput
async def main() -> None:
    async with AsyncMorphiaOne(
        api_key=os.environ["MORPHIAONE_API_KEY"],
        base_url=os.environ["MORPHIAONE_BASE_URL"],
    ) as client:
        source = await client.assets.upload("face.jpg")
        job = await client.skin.analyze(
            asset_id=source.id,
            concerns=["redness"],
            idempotency_key="example-order-48213-redness",
        )
        done = await client.jobs.wait(job.id)
        if isinstance(done.output, SkinAnalysisOutput):
            print(done.output.concerns)
asyncio.run(main())

Usage

client.usage.get reads accepted job counts by operation and UTC day. Its optional start_date and end_date use YYYY-MM-DD; the selected interval includes the start and excludes the end. The report is operational data, not an invoice.

import os
from morphiaone import MorphiaOne
with MorphiaOne(
    api_key=os.environ["MORPHIAONE_API_KEY"],
    base_url=os.environ["MORPHIAONE_BASE_URL"],
) as client:
    report = client.usage.get(start_date="2026-09-01", end_date="2026-10-01")
    print(report["totals"])

Idempotency

The SDK generates an idempotency key when a submit call omits one and reuses it across that call’s automatic retries. This only protects retries within that call. For safe retries after a process restart or lost response, pass a stable key derived from your own durable operation id and reuse it with the same input Asset and request fields. Use a new key when the logical operation or request body changes. Asset uploads are not idempotent.

The same key and same submission recover the same job. Reusing a key with different fields or Assets raises ConflictError with IDEMPOTENCY_KEY_REUSED. See Idempotency.

Timeouts and retries

• request_timeout defaults to 10 seconds for one SDK request operation, including its retries. operation_timeout defaults to 120 seconds for a wait or a submit with wait=True; configure it on the client when work needs longer. Both are seconds, and operation_timeout must be at least request_timeout.
• A method’s timeout overrides its relevant default. On a submit with wait=True, it covers submission and polling together. On jobs.wait, it covers the full wait.
• The default RetryPolicy makes at most four attempts. It retries transport failures, 408, 500, 502, 503, 504, retryable 429 responses, and 409 IDEMPOTENCY_IN_PROGRESS. It does not blindly retry QUOTA_EXCEEDED or INSUFFICIENT_CREDITS; a queue-limit 429 is retried only when the server sends Retry-After.
• For retried HTTP requests, the SDK honors Retry-After up to 60 seconds and only when the remaining deadline can accommodate the wait; otherwise it raises the API error instead of retrying early. jobs.wait polls using Retry-After, capped at 60 seconds, or about one second when the server sends none. Without Retry-After, request retries use capped exponential backoff with jitter.

For durable integrations, persist each job id before waiting and resume with jobs.get or jobs.wait. See Retries and backoff and Asynchronous jobs.

Errors

Every API or client-runtime error derives from MorphiaOneError. An HTTP error is an ApiError with status, machine-readable code, human message, details, request_id, and retry_after in seconds when the server supplies Retry-After. Local transport, protocol, and deadline errors have no HTTP status or request id. Branch on code, not on message, and log request_id for support.

Invalid local arguments and uploads exceeding max_image_bytes raise ValueError before the SDK sends a request. ProcessingError is not an HTTP error: the job request succeeded, but its work failed. Its error carries the job error code and details.

Handle errors inside the client context around an SDK call. This fragment assumes client and job came from your submission code:

from morphiaone import ApiError, MorphiaOneError, ProcessingError
try:
    done = client.jobs.wait(job.id)
except ProcessingError as error:
    print("job failed", error.job_id, error.error.code)
    raise
except ApiError as error:
    print("request failed", error.status, error.code, error.request_id)
    raise
except MorphiaOneError:
    # Transport, protocol, or deadline failure; a submitted job may still be running.
    raise

See the error reference and status-code guide for response codes and recovery guidance.

Lists, cancel, delete, credits and webhooks

from morphiaone import JobCancelledError, webhooks
# One page, or every page by next_cursor.
page = client.jobs.list(status="running", limit=20)
for asset in client.assets.list_all(purpose="input", max_items=200):
    print(asset.id)
# Cancel a job that has not started processing; delete a finished one.
client.jobs.cancel(page.data[0].id)
try:
    client.jobs.wait(page.data[0].id)
except JobCancelledError:
    pass
client.jobs.delete("job_4ryn2an9hm2qxnb6xvbrav094b")
client.assets.delete("asset_6pa8ad0ds2t7g1z6pg4tcr4zqp")
# Credits, and the rate-limit reading of the last response that spent a policy.
balance = client.credits.balance()
print(balance["spendable_units"], balance["enforced"])
print(client.last_rate_limit)

Verify a webhook delivery against the raw request body before you parse it:

event = webhooks.verify(
    raw_body,
    request.headers.get("MorphiaOne-Signature"),
    os.environ["MORPHIAONE_WEBHOOK_SECRET"],
)

The async client, AsyncMorphiaOne, has the same methods. See Jobs, Assets, Usage and credits, Rate limits and Webhooks for what each call answers.