Norcaster — Python SDK

Thin wrapper around the Norcaster REST API for deploy, run, status, and usage.

Installation

pip install norcaster

Python 3.10 or later. The SDK's only dependency is httpx.

From a checkout of the Norcaster repository: pip install ./sdk/python.

Command line

The norcaster command comes with the cli extra, which adds typer, rich and python-dotenv:

pip install "norcaster[cli]"
export NORCASTER_API_KEY="<your-api-key>"
norcaster deploy qx-example
norcaster run <deployment_id> -i '{"prompt": "Hello!"}'

Without the extra, norcaster only says how to install it. The CLI reads NORCASTER_API_KEY and NORCASTER_API_URL from the shell first, then from .env in the current folder, then from ~/.norcaster.env; it never reads .env files in parent folders. norcaster --help lists the commands.

Usage

from norcaster import NorcasterCloudClient

client = NorcasterCloudClient(api_key="your-api-key")

# Deploy a model
deploy = client.deploy("my-llama-7b", model_path="models/user/llama")
print(deploy.deployment_id, deploy.status)

# Run inference
run = client.run(deploy.deployment_id, {"prompt": "Hello"})
print(run.job_id, run.status)

# Check status
status = client.status(run.job_id)
print(status.status, status.output_data)

# Get usage
usage = client.usage()
print(usage.tokens_used, usage.compute_seconds)

Response types

• DeployResult — deployment_id, status, message
• RunResult — job_id, status, message
• StatusResult — id, type, status, output_data, tokens_used, etc.
• UsageResult — user_id, tokens_used, compute_seconds, job_count
• CheckResult — decision, receipt, prompt or messages (when the decision is hide), code, reason
• CheckReceipt — id, content_sha256, policy_version, decided_at

Check before your app calls an AI provider

Your app can ask Norcaster for a decision and then call the AI provider itself. Norcaster answers allow, hide or block from the app's rules, and keeps a record of every answer.

What the record shows: the app asked Norcaster before it called the provider. Norcaster does not see the provider call, so the record does not show what the app sent afterwards.

Ask for a decision

from norcaster import NorcasterCloudClient

client = NorcasterCloudClient(api_key="your-api-key")

prompt = "Summarise this ticket for the customer"
result = client.check(prompt)  # or client.check(messages=[...]) for a conversation

if result.decision == "block":
    ...  # do not send it; result.reason says why
elif result.decision == "hide":
    prompt = result.prompt  # send this version: the sensitive data is hidden

receipt = result.receipt
print(receipt.id)              # the record in Norcaster
print(receipt.content_sha256)  # SHA-256 of what you asked about
print(receipt.policy_version)  # the version of the app's rules that decided
print(receipt.decided_at)      # when

• Pass prompt or messages, not both. When you ask with messages, a hide answer comes back in result.messages.
• deployment_id="..." is the app's id. Leave it out when the key belongs to one app.
• The key needs the runs:write scope.
• In an async app, use await client.acheck(...).
• content_sha256(prompt) or content_sha256(messages=...) gives the hash for what you are about to send, to compare with the receipt.

Wrap your OpenAI or Anthropic client

from openai import OpenAI
from norcaster import NorcasterBlocked, norcaster_check

openai_client = client.wrap_openai(OpenAI())

try:
    answer = openai_client.chat.completions.create(model="gpt-4o-mini", messages=messages)
except NorcasterBlocked as blocked:
    print(blocked.code, blocked.receipt_id)   # the provider was not called
else:
    print(norcaster_check(answer).receipt_id)  # the record behind this answer

Before each call the wrapper asks Norcaster:

• allow: the request goes to the provider as it is.
• hide: the wrapper sends the version Norcaster returned, with the sensitive data hidden.
• block: the provider is not called and NorcasterBlocked is raised. The error names the decision, the reason and the receipt.

client.wrap_anthropic(Anthropic()) does the same for messages.create. AsyncOpenAI and AsyncAnthropic clients work the same way with await. You pass in your own client: this SDK does not install or import openai or anthropic.

What is covered:

• OpenAI: chat.completions.create, streamed or not.
• Anthropic: messages.create, streamed or not.
• Any other call on the wrapped client raises an error. Make that call on your own client, where it is not checked.
• The check is of what your app sends. The provider's answer is not checked unless you turn that on (next section).
• An Anthropic request that holds an image or a document is not sent, because a check can't read those yet.
• In an Anthropic conversation, the model's signed thinking is left out of the check and sent on unchanged.

Check the model's answer too

This is off unless you turn it on. With check_output=True the wrapper also asks Norcaster about the answer's text before your app gets it:

from norcaster import NorcasterBlocked, norcaster_output_check

openai_client = client.wrap_openai(OpenAI(), check_output=True)

try:
    answer = openai_client.chat.completions.create(model="gpt-4o-mini", messages=messages)
except NorcasterBlocked as blocked:
    if blocked.about == "answer":
        ...  # the provider answered and the answer was withheld; blocked.receipt_id names the record
    else:
        ...  # the request was blocked and the provider was not called
else:
    print(norcaster_output_check(answer).receipt_id)  # the record of the answer's check

• allow: your app gets the answer as it is.
• hide: your app gets the answer with the sensitive data hidden.
• block: NorcasterBlocked is raised with about == "answer". The error names the decision, the reason and the receipt.

The answer's check is recorded under the same request as the check before the call, so your records show one call with two decisions. It adds one more round trip to Norcaster per call.

Know these limits before you turn it on:

• The app's rules must cover answers. If they don't, Norcaster refuses the check (NorcasterCheckRefused, code output_rules_off) and the answer is withheld.
• Only the answer's text is read. Tool calls the model asks for, images and reasoning are not. An answer with no text (a tool call only) is not asked about, and norcaster_output_check returns None for it.
• More than one answer per call (n above 1) raises NorcasterCannotCheck before anything is sent.
• A streamed answer is held until it has been checked, so your app gets its first chunk later (next section).
• When Norcaster can't be reached after the provider answered, the answer is withheld: NorcasterUnreachable with about == "answer". With when_unreachable="let_through" your app gets the answer, and it is recorded as let through without a check, with the request it belongs to.
• When the request itself was let through without a check, its answer is not asked about. The one let-through record stands for the call.
• The record shows your app asked about a text. It does not show that the text was the provider's answer, or what your app did with it afterwards.

Streamed answers

With check_output=True a streamed call (stream=True) works, but your app gets nothing until the whole answer has been checked. Nothing is checked while it streams:

1. The wrapper reads the provider's whole stream. Your app gets none of it yet.
2. It asks Norcaster once, about the whole answer's text, under the request's check.
3. Only then does create return. allow: you get the provider's chunks as they came. hide: the same chunks, with Norcaster's text where the answer's text starts and the later pieces of text empty. block: NorcasterBlocked (about == "answer") is raised from create, and your app gets no chunk at all.

What it costs: the wait for the first chunk becomes the time the provider takes to finish the whole answer, plus one round trip to Norcaster. An app that streams so that people see the first words early loses that while answer checks are on.

stream = openai_client.chat.completions.create(model="gpt-4o-mini", messages=messages, stream=True)
for chunk in stream:  # the provider's own chunk type
    ...
print(norcaster_output_check(stream).receipt_id)

• You get a CheckedStream (an AsyncCheckedStream from an async client): iterate it once, use it in with, close() it. Anything else, such as response, is the provider's stream's own.
• If the provider's stream fails part-way, its error is raised from create and none of the text before it reaches your app. No answer check is made.
• If your app cancels the call while the answer is held, the provider's stream is closed and nothing reaches your app.
• The whole answer is read before your app sees any of it, so stopping early (leaving the loop) does not stop the provider part-way. The record is of the whole answer.
• Tool-call deltas, refusal text, reasoning and usage chunks are passed on as they came and are not read, as for an answer that is not streamed. A hidden answer's token list (logprobs) is set to None, because it spells out the original text.
• When the request itself was let through without a check, its stream is not held: it reaches your app as it arrives, and its answer is not asked about.

To ask about an answer yourself, pass the request_id of the check you made before the call:

asked = client.check(messages=messages)
...  # call the provider
answer = client.check(output=text, request_id=asked.request_id)
if answer.decision == "hide":
    text = answer.output

When Norcaster can't be reached

By default the request is blocked: NorcasterUnreachable is raised and the provider is not called. "Can't be reached" means no connection, no answer within timeout (10 seconds unless you set it), or an error on Norcaster's side.

You can set an app to let requests through instead:

openai_client = client.wrap_openai(
    OpenAI(),
    when_unreachable="let_through",
    on_unchecked=lambda unchecked: audit_log.warning("sent without a check: %s", unchecked.id),
)

Every request that is let through is recorded, each time:

1. It is written to the norcaster.check log as a warning.
2. It is passed to on_unchecked.
3. It is sent to Norcaster after the next check that succeeds, and shows in your records as "Let through without a check". Call client.report_unchecked() before your app stops, to send what is still waiting.

The record holds a SHA-256 of the request, when it happened and why. It never holds the request itself. No rule ran on a request that was let through, and it is not counted as a check.

Know these limits before you turn it on:

• A request that is let through goes to the provider as your app wrote it. Nothing is hidden.
• Requests waiting to be sent to Norcaster are kept in memory. If the app stops before Norcaster answers again, your own log line is the only record.
• At most 10,000 wait at a time. Past that, requests are blocked.
• A refusal from Norcaster always blocks, whatever the setting: a wrong or revoked key, a missing scope, a stopped app, the rate limit, or more than 1,000,000 characters. These raise NorcasterCheckRefused.

examples/check-mode/sample_app.py runs all of this against a stand-in provider; add --answers for the answer checks.