Metadata-Version: 2.5
Name: norcaster
Version: 0.1.0
Summary: Norcaster Python SDK: ask Norcaster for a decision (allow, hide or block) before your app calls an AI provider, and keep a record of every answer. Wraps your own OpenAI or Anthropic client.
Project-URL: Homepage, https://www.norcaster.com
Author-email: "Norcaster (Navego AB)" <security@norcaster.com>
License: MIT
License-File: LICENSE
Keywords: ai-governance,ai-runtime-control-plane,anthropic,audit-evidence,guardrails,openai
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Typing :: Typed
Requires-Python: >=3.10
Requires-Dist: httpx<1,>=0.24
Provides-Extra: cli
Requires-Dist: python-dotenv>=1.0; extra == 'cli'
Requires-Dist: rich>=13; extra == 'cli'
Requires-Dist: typer>=0.16; extra == 'cli'
Provides-Extra: test
Requires-Dist: pytest>=7; extra == 'test'
Description-Content-Type: text/markdown

# Norcaster — Python SDK

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

## Installation

```bash
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`:

```bash
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

```python
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

```python
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

```python
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:

```python
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.

```python
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:

```python
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:

```python
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.
