Metadata-Version: 2.4
Name: enkryptai-sdk
Version: 1.0.42
Summary: A Python SDK with guardrails and red teaming functionality for API interactions
Home-page: https://github.com/enkryptai/enkryptai-sdk
Author: Enkrypt AI Team
Author-email: software@enkryptai.com
Classifier: Programming Language :: Python :: 3
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Requires-Python: >=3.11
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: requests>=2.25
Requires-Dist: pandas>=1.3
Requires-Dist: tabulate>=0.8
Requires-Dist: python-dotenv>=0.20
Requires-Dist: websockets<17,>=13
Requires-Dist: httpx<1.0,>=0.27
Requires-Dist: openai<3.0,>=1.30
Requires-Dist: pydantic<3.0,>=2.5
Dynamic: author
Dynamic: author-email
Dynamic: classifier
Dynamic: description
Dynamic: description-content-type
Dynamic: home-page
Dynamic: license-file
Dynamic: requires-dist
Dynamic: requires-python
Dynamic: summary

# Enkrypt AI Python SDK

![Python SDK test](https://github.com/enkryptai/enkryptai-sdk/actions/workflows/test.yaml/badge.svg)

A Python SDK with Guardrails, Code of Conduct Policies, Endpoints (Models), Deployments, AI Proxy, Datasets, Red Team, Skill Scanner, etc. functionality for API interactions.

**See documentation at [https://docs.enkryptai.com/libraries/python/introduction](https://docs.enkryptai.com/libraries/python/introduction)**

See [https://pypi.org/project/enkryptai-sdk](https://pypi.org/project/enkryptai-sdk)

## Start a red team run

```python
import os
from enkryptai_sdk import RedTeamClient, RTModelConfig, RTRedteamRequest

client = RedTeamClient(api_key=os.environ["ENKRYPTAI_API_KEY"])

run = client.run_redteam(RTRedteamRequest(
    target=RTModelConfig.hosted(
        endpoint="https://api.openai.com/v1/chat/completions",
        api_key=os.environ["OPENAI_API_KEY"],
        model_name="gpt-4o",
    ),
    risk_categories={"safety_harm": {"attack_config": ["basic"]}},
    run_name="nightly probe",
))

print(run.run_id)                  # rt-<uuid> -- save this
print(client.run_url(run.run_id))  # watch it in the dashboard
```

A run takes anywhere from about half an hour to several hours, so starting it
and reading it are usually different sittings. `run_id` is all you need to pick
it back up from a fresh process:

```python
status = client.wait_for_run(run.run_id, on_progress=print)
report = client.get_run_results(run.run_id)
```

`wait_for_run` has no timeout by default and interrupting it does nothing to
the run. For a live feed instead of polling, `client.iter_run_events(run_id)`
yields decoded events and reconnects on its own, resuming where it left off —
which matters, because a run that lasts hours will outlive its connection.

> **One run, three id spellings.** `rt-<uuid>` is what the endpoints want;
> the bare uuid (`job_id_for(run_id)`, also on `status.job_id`) is what relay
> logs and compliance reports use. And a finished run reports `Finished` from
> one endpoint and `completed` from another — `status.state` and
> `status.is_terminal` fold both into one.

### Against a model on your own machine

Same script, one different target — plus a bridge (see below):

```python
    target=RTModelConfig.via_relay(
        bridge_id="my-laptop",
        endpoint="http://localhost:11434/v1/chat/completions",
        model_name="llama3",
    ),
```

## Relay bridge

The SDK also ships the **Enkrypt Sentry Relay bridge**: a tiny
in-network process that lets red-team jobs running in Enkrypt's cloud
reach an LLM that lives inside your private network -- without opening
any inbound ports. The bridge is pure network plumbing: it does no
LLM work itself, just maintains one outbound WSS connection to
`api.enkryptai.com:443`, receives OpenAI-shaped `chat.completions`
requests over it, forwards them to your local LLM, and pushes the
response back. (Industry analogues for the same role are Twingate /
Zscaler *Connector* and Cloudflare's *tunnel*.)

> **Availability:** the public relay route is currently enabled in
> Enkrypt's **dev** environment only. Confirm with your Enkrypt contact
> which URL your bridge should dial before rolling it out — pointed at an
> environment where the route is absent, the bridge does not fail loudly,
> it just reconnect-loops.

### One-command start

```bash
pip install enkryptai-sdk

export ENKRYPT_API_KEY=<your-enkrypt-api-key>
enkryptai-relay --bridge-id my-laptop --target http://localhost:11434
```

Two required values, and no user id: the gateway authenticates your API
key and tells the relay whose bridge this is. The bridge id is the value
you also pass as `bridge_id` on the red-team target — it is the one value
the two sides must agree on.

Every flag has an environment-variable equivalent, which is what you want
under systemd, docker or k8s. Flags win when both are set:

```bash
export RELAY_BRIDGE_ID=my-laptop
export ENKRYPT_API_KEY=<your-enkrypt-api-key>
export TARGET_BASE_URL=http://localhost:11434          # your local LLM
# Optional:
# export BRIDGE_HOOKS_MODULE=my_company.relay_hooks    # custom translation
# export RELAY_TARGET_ALLOWED_HOSTS=local-llm.corp     # host allow-list

enkryptai-relay
```

Run `enkryptai-relay --help` for the full list. Prefer `ENKRYPT_API_KEY`
over `--api-key`, which lands in your shell history.

Keep the bridge up for the whole run: a red team run lasts from about half
an hour to several hours, and if the bridge drops the run pauses and
eventually fails. Run it as a service, not in the terminal you are about
to close.

### Programmatic API

```python
from enkryptai_sdk import RelayBridge

RelayBridge(
    bridge_id="my-laptop",
    api_key="<your-enkrypt-api-key>",
    target_base_url="http://localhost:11434",
).run()
```

Arguments are **keyword-only** — positional construction raises
`TypeError` rather than silently rebinding fields.

### Translation hooks (non-OpenAI local LLMs)

The relay wire format is OpenAI `chat.completions` end-to-end. If your
local LLM doesn't already speak OpenAI (Anthropic, Bedrock, Vertex,
proprietary shape, ...) write a Python module that exports two
coroutines and point `BRIDGE_HOOKS_MODULE` at its dotted path:

```python
async def before_request(payload: dict) -> dict:
    return translate_openai_to_local(payload)

async def after_response(local_response: dict) -> dict:
    return translate_local_to_openai(local_response)
```

The bridge validates inputs/outputs against the official `openai` SDK
Pydantic types at both boundaries, so a buggy hook surfaces as a
structured error to the red-team worker instead of corrupted traffic.
Nothing about your local LLM's shape has to be known by, or deployed to,
Enkrypt's cloud.

A worked **OpenAI ↔ Anthropic Messages API** example ships inside the
SDK at `enkryptai_sdk.relay.examples.hooks_example`. Either point the
bridge at it directly (smoke test) or copy it into your own repo to
edit:

```bash
# Smoke test (no copy):
export BRIDGE_HOOKS_MODULE=enkryptai_sdk.relay.examples.hooks_example
enkryptai-relay

# Or, copy the template next to your own code:
python -c "from enkryptai_sdk.relay.examples import copy_example; \
    copy_example('hooks_example.py', './my_hooks.py')"
export BRIDGE_HOOKS_MODULE=my_hooks
PYTHONPATH=. enkryptai-relay
```

A `bridge.env.example` env-file template ships alongside it and can
be copied the same way (`copy_example('bridge.env.example',
'./bridge.env')`). See
[`src/enkryptai_sdk/relay/examples/README.md`](src/enkryptai_sdk/relay/examples/README.md)
for the full list.

### Turning the relay on for a run

Routing is switched on by the red-team request, not by the bridge.
`RTModelConfig.via_relay` builds that target for you:

```python
from enkryptai_sdk import RTModelConfig

target = RTModelConfig.via_relay(
    bridge_id="my-laptop",                                  # == --bridge-id
    endpoint="https://local-llm.corp/v1/chat/completions",  # as the bridge sees it
    model_name="their-internal-model",
    # Credentials your local LLM needs. They go here, never in api_key --
    # the bridge is what authenticates to your LLM, so passing api_key raises.
    target_headers={"Authorization": "Bearer customer-side-internal-key"},
)
```

which serialises to the wire shape below. Write it by hand if you prefer:

```json
{
  "target": {
    "endpoint": "https://local-llm.corp/v1/chat/completions",
    "api_key": "",
    "model_name": "their-internal-model",
    "connect_via_relay": true,
    "metadata": {
      "relay": {
        "bridge_id": "my-laptop",
        "target_endpoint": "https://local-llm.corp/v1/chat/completions",
        "model_name": "their-internal-model"
      }
    }
  },
  "risk_categories": { "safety_harm": { "attack_config": { "basic": {} } } }
}
```

`metadata.relay.bridge_id`, `metadata.relay.target_endpoint` and
`metadata.relay.model_name` are all required; `target.api_key` may be
empty because the *bridge* is what authenticates to your LLM (put those
credentials in `metadata.relay.target_headers`). Note that
`connect_via_relay` stays at the *root* of the target — only the relay
block itself lives under `metadata`. A bare `target.relay` block is the
older spelling and is still accepted, so existing integrations keep
working; write `target.metadata.relay` in new ones. Ready-to-send bodies
with a field-by-field reference are in
[`docs/relay/examples/`](docs/relay/examples/).

### Further reading

[`docs/relay/`](docs/relay/) covers the architecture and config
reference ([README](docs/relay/README.md)), how to run both sides on one
laptop ([LOCAL_TESTING](docs/relay/LOCAL_TESTING.md)), deploying the
cloud side ([INFRA_RUNBOOK](docs/relay/INFRA_RUNBOOK.md)), and why the
relay is shaped this way ([DESIGN](docs/relay/DESIGN.md)).

## Scan an agent skill

The Skill Scanner checks an agent "skill" (a directory in a git repo) for
security threats. Submitting is asynchronous — you get a `scan_id` back and
poll it; a scan typically takes 30-90 seconds.

```python
import os
from enkryptai_sdk import SkillScannerClient

client = SkillScannerClient(api_key=os.getenv("ENKRYPTAI_API_KEY"))

queued = client.scan({
    "git_url": "https://github.com/affaan-m/ECC.git",
    "skill_path": ".agents/skills/api-design",
    # Recommended: pins the checkout, and lets an identical repeat scan come
    # back from cache instead of re-running the scanner.
    "commit": "2bc924aa11bb22cc33dd44ee55ff6677889900aa",
})
print(queued.scan_id, queued.status)      # -> "...", "queued"

# Blocks until the scan is done (default budget 15 min, polls every 5s).
record = client.wait_for_scan(queued.scan_id)

if record.succeeded:
    print(record.verdict, record.risk_level, record.findings_count, record.stars)
    print(client.get_report(record.scan_id))   # the full skill-sentinel report
else:
    print("scan failed:", record.error)

# Your own scans, newest first.
for item in client.list_scans(status="succeeded", limit=10).items:
    print(item.scan_id, item.repo, item.skill_name, item.verdict)
```

Two things worth knowing before you build on it:

- **You see your organization's scans.** The gateway derives the owning
  identity from your API key. For an org or project key that is the
  **organization**, so every member sees every scan the org has run, whichever
  project their key belongs to; an individual account sees its own. A scan
  belonging to a different organization is a `403`. There is deliberately no
  `user_email` parameter to pass — sending one is a `400`.
- **A failed scan is a result, not an exception.** `wait_for_scan` returns the
  record for a `failed` scan with `record.error` explaining why; only running
  out of time raises (`SkillScannerTimeoutError`).

`force=True` on `scan()` bypasses the dedup cache and forces a fresh scan.

## Run a compliance scan

A compliance scan continuously reads a provider workspace's export log files,
runs every message through one of your Guardrails policies, and indexes what it
finds. You create the scan; the platform runs the timer.

```python
import os
from enkryptai_sdk import ComplianceClient

client = ComplianceClient(api_key=os.getenv("ENKRYPTAI_API_KEY"))

# Probe the key and the workspace before committing to a scan. A proposed
# cursor also returns a backfill estimate -- worth quoting before you start,
# because volume varies by orders of magnitude between workspaces.
probe = client.test_connection({
    "compliance_api_key": os.getenv("COMPLIANCE_API_KEY"),
    "workspace_id": "00000000-0000-0000-0000-000000000000",
    "proposed_cursor": "2026-09-01T00:00:00Z",
})
print(probe.key_valid, [p.event_type for p in probe.event_types if p.supported])

client.add_scan({
    "scan_name": "chatgpt-enterprise",
    "guardrails_name": "my-guardrail",
    "provider": "openai",
    "workspace_id": "00000000-0000-0000-0000-000000000000",
    "event_types": ["CONVERSATION_MESSAGE"],
    "cursor_end_time": "2026-09-01T00:00:00Z",
    "compliance_api_key": os.getenv("COMPLIANCE_API_KEY"),
})

# How fresh the scan is: lag_s is how far behind the provider's clock it runs.
scan = client.get_scan("chatgpt-enterprise")
print(scan.scan.status, scan.lag_s)

# What it has covered, and one message's text.
files = client.list_log_files("chatgpt-enterprise", per_page=20)
print(files.pagination.total_count, files.log_files[0].flagged_count)
print(client.get_message("chatgpt-enterprise", "EVENT_ID").text)

client.pause_scan("chatgpt-enterprise")
client.start_scan("chatgpt-enterprise")     # idempotent; re-resolves the key
```

Four things worth knowing before you build on it:

- **Every path requires the `governance_officer` role.** Not an org admin, not
  a project admin, and not the org owner unless the role was granted to them
  explicitly. The role resolves through an organization, so an individual
  account cannot hold it and these paths are unreachable on one.
- **A scan is addressed by name** within your API key's project, so there is no
  id to carry around.
- **`provider` and `workspace_id` are immutable** once the scan exists —
  changing either would orphan the cursor and everything already indexed, so
  `modify_scan` rejects them. Delete and recreate instead.
- **`compliance_api_key` is write-only.** No read returns it, not even masked.
  Send a fresh one through `modify_scan` to clear a `needs_reconnect`, then
  `start_scan`.

A bad key from `test_connection` is a `200` with `key_valid=False`, not an
error. `get_message` fetches the text live and raises
`ComplianceMessageExpiredError` (HTTP 410) once the provider ages the file out
of its retention window -- expected for old messages, not a failure.

## Copyright, License and Terms of Use

© 2025 Enkrypt AI. All rights reserved.

Enkrypt AI software is provided under a proprietary license. Unauthorized use, reproduction, or distribution of this software or any portion of it is strictly prohibited.

Terms of Use: [https://www.enkryptai.com/terms-and-conditions](https://www.enkryptai.com/terms-and-conditions)

Enkrypt AI and the Enkrypt AI logo are trademarks of Enkrypt AI, Inc.
