clavata-sdk 1.1.0

Clavata SDK for Python

SDK

This is the Clavata SDK for Python.

Quick Start

For most users, you'll be using the default production API, so unless you've been told otherwise, we recommend using the steps below to create a client.

Set your Clavata API Key in the environment:

export CLAVATA_API_KEY="YOUR_API_KEY"

Then, you can instantiate our client with default settings:

from clavata_sdk import ClavataClient

client = ClavataClient.create()

If you wish, you can also inject the API token directly:

from clavata_sdk import ClavataClient

client = ClavataClient.create(auth_token="API_KEY")

Custom host/port

If you've been given a different host and port to connect to, you'll instantiate the client as follows:

from clavata_sdk import ClavataClient

Next, you'll need a Clavata API key to instantiate the client:

api_key = "YOUR_API_KEY"

# Now instantiate the client with your API key:
client = ClavataClient(host="gateway.app.clavata.ai", port=443, auth_token=api_key)

As above, the auth_token parameter can be omitted if your API key is set in the environment before instantiation.

Basic Use

The client includes methods for evaluation and job retrieval. It also provides access to the request and response types you'll need to pass in.

from clavata_sdk import CreateJobRequest, EvaluateRequest, EvaluateLabelsRequest, ContentData

# Async request
response = await client.create_job(CreateJobRequest(content=ContentData(text="..."), policy_id="policy-123", wait_for_completion=True))

# Streaming request - one response per piece of content
async for response in client.evaluate(EvaluateRequest(content="The quick brown fox", policy_id="policy-123")):
    print(response)

# Labels v2 evaluation is currently beta and exposed through client.beta.evaluate_labels.
# Requires Labels v2 on your customer account; one content item, 1-10 explicit label version IDs.
response = await client.beta.evaluate_labels(request=EvaluateLabelsRequest(
    content=ContentData(text="The quick brown fox"),
    label_version_ids=["label-version-id-1", "label-version-id-2"],
    threshold=0.5,
))

for result in response.results:
    if result.evaluation:
        print(
            result.label_id,
            result.label_version_id,
            result.job_uuid,
            result.evaluation.outcome,
            result.evaluation.score,
            result.evaluation.threshold,
        )
    elif result.error:
        print(result.label_id, result.label_version_id, result.job_uuid, result.error.message)

Labels v2 responses preserve requested label version ID order. The standard response includes content_hash and, for each result, label_id, label_version_id, job_uuid, either evaluation (outcome, score, threshold) or error, and no nested report. This is typically sufficient for most SDK uses. Optional nested reports are populated only when include_evaluation_report=True and the backend returns report data; use them when you need full report details such as matches from section_evaluation_reports entries whose review_result.outcome is "TRUE". Generated LabelsService CRUD, version, draft, tag, batch, and dataset APIs are private and are not part of the public SDK surface.

Understanding Responses

Policy evaluation methods return response objects containing a policy_evaluation_report with the evaluation results:

response = await client.create_job(CreateJobRequest(
    content=ContentData(text="Sample text"),
    policy_id="policy-123",
    wait_for_completion=True
))

# Access the policy-level decision
print(response.results[0].report.review_result.outcome)  # "TRUE", "FALSE", or "FAILED"
print(response.results[0].report.review_result.score)    # Numeric score

# Access label-level results
for section in response.results[0].report.section_evaluation_reports:
    print(f"Label: {section.name}")
    print(f"Result: {section.review_result.outcome}")
    print(f"Score: {section.review_result.score}")

# Or use the convenience 'matches' field to get only labels that matched
print(response.results[0].matches)  # Dict of label_name -> score for TRUE labels

Webhooks

The client.create_job() method is an async endpoint, which means the job will be started but you'll receive a response before the job has completed. This response is a Job and will include the job's ID and status. You can use the job ID to get the results later using the get_job() method, but if you'd like to be notified when the job is complete (so you don't have to "poll"), you can add a "webhook". A webhook is a URL that Clavata's platform will make a request to when the job is complete.

To add a webhook to a create_job() request, do the following:

from clavata_sdk import ClavataClient, CreateJobRequest

client = ClavataClient.create()
request = CreateJobRequest(content="The quick brown fox", ...).add_webhook(url="https://youdomain.com/path/to/hook")
job = client.create_job(request)

print(job)

You can call add_webhook on the request object and provide a URL to call. Clavata will make a POST request to this URL when the job is complete. You can also optionally add an extra_headers named parameter and provide a dictionary of key/value pairs. Clavata will add these headers to the POST request, if provided.

Dealing with Refusal Errors

Under certain circumstances, the Clavata API may refuse to evaluate a piece of content. Reasons include:

• The content (image) was found in a database of known CSAM material
• The content (image) is in a format that is not currently supported (webp, png, jpg supported as of publishing)
• The content is corrupt, incomplete or otherwise invalid

When the API refuses, an exception will be raised by the SDK. Because exceptions may also be raised for other reasons (i.e., network errors, invalid requests, etc), the SDK provides a distinct error type that both helps you tell when an error is due to a refusal, and also allows you to get more information about why the request was refused.

The EvaluationRefusedError type will only be raised if an evaluation was refused for one of the reasons mentioned above.

When a refusal occurs, the response from the API is processed to hydrate this error with information on which pieces of content included in the request were refused, and why each one was refused (if there was more than one piece of content in the request which was refused).

The simplest way to examine refusals is to look at the refusals attribute on the error type. It will always be a list containing one entry for each piece of content that was refused. Each entry is of type RefusedContent and includes a reason as well as the content_hash that identifies the piece of content.

try:
    client.evaluate(...)
except EvaluationRefusedError as err:
    print("Refused content: ", err.refusals)

The error type also comes with a number of additional properties that are useful if you don't want to have to iterate over the entire list of refusals (for example, if you only sent 1 piece of content in the request). These properties are

• top_reason: Returns the "most important" reason for the refusal, with CSAM > UNSUPPORTED_FORMAT > INVALID_CONTENT.
• most_common_reason: If there are many refusals, returns the most common reason among them. In math terms, the "mode".
• first_reason: Always returns the first reason in the list, regardless of how many refusals there were.

You can also access all the content that was refused in a flattened list by using the refused_content_hashes property on the error type.

try:
    client.evaluate(...)
except EvaluationRefusedError as err:
    print("Refused content: ", err.refusals)

    # Get the first refusal reason
    print("First refusal reason: ", err.first_reason)

    # Get all the content that was refused
    print("Refused hashes: ", err.refused_content_hashes)

Finally, the refusal "reasons" are implemented as a python enum, allowing you to easily compare the "reason" in the error to the canonical "reason" values to determine why the content was refused (and then what action to take).

from clavata_sdk import RefusalReason, EvaluationRefusedError
try:
    client.evaluate(...)
except EvaluationRefusedError as err:
    if err.first_reason == RefusalReason.CSAM:
        # Now you know that CSAM was found. Act accordingly.
    # etc ...