Metadata-Version: 2.4
Name: gjallarhorn-hq-sdk
Version: 1.0.2
Summary: Official Python SDK for the Gjallarhorn prompt-injection detection API
Project-URL: Homepage, https://gjallarhorn.watch
Project-URL: Repository, https://github.com/gjallarhorn-hq/gjallarhorn
Project-URL: Documentation, https://gjallarhorn.watch/docs
Project-URL: Issues, https://github.com/gjallarhorn-hq/gjallarhorn/issues
License: MIT
License-File: LICENSE
Keywords: ai-security,gjallarhorn,llm-security,prompt-injection
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.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Requires-Python: >=3.9
Requires-Dist: httpx>=0.24.0
Description-Content-Type: text/markdown

# gjallarhorn-hq-sdk

Official Python SDK for the [Gjallarhorn](https://gjallarhorn.watch) prompt-injection detection API.

[![PyPI version](https://badge.fury.io/py/gjallarhorn-hq-sdk.svg)](https://pypi.org/project/gjallarhorn-hq-sdk/)
[![License: MIT](https://img.shields.io/badge/License-MIT-blue.svg)](LICENSE)

## Installation

```bash
pip install gjallarhorn-hq-sdk
```

**Requirements**: Python 3.9+, `httpx`.

## Quickstart

```python
import os
from gjallarhorn_hq_sdk import GjallarhornClient

client = GjallarhornClient(api_key=os.environ['GJALLARHORN_API_KEY'])

result = client.scan(input=user_input)
if result.risk_level in ('critical', 'high'):
    raise ValueError('Injection detected — request blocked')
# safe to send to LLM
```

## Async usage

```python
import asyncio
import os
from gjallarhorn_hq_sdk import AsyncGjallarhornClient

async def main():
    async with AsyncGjallarhornClient(api_key=os.environ['GJALLARHORN_API_KEY']) as client:
        result = await client.scan(input='some user input')
        print(result.risk_level, result.detection_layers)

asyncio.run(main())
```

## Methods

| Method | Description |
|--------|-------------|
| `scan(content, *, use_classifier?, context?)` | Scan text for prompt injection |
| `scan_multimodal(file, mime_type, filename?)` | Scan PDF/image/QR code |
| `get_quota()` | Get credit usage and quota |
| `check_canary(llm_response)` | L2 output integrity check (opt-in) |
| `health()` | Server/DB health check — no auth required, endpoint is `/health` (not `/v1/health`) |

Both `GjallarhornClient` (sync) and `AsyncGjallarhornClient` (async) expose the same methods.

## Scan a PDF

```python
from pathlib import Path

result = client.scan_multimodal(Path('user-upload.pdf'), 'application/pdf')
print(result.risk_level, result.billed_pages)
```

## L2 Agentic Integration

`check_canary()` checks whether your LLM's response still contains the Gjallarhorn canary token. Absence signals a possible system-prompt override. **L2 is opt-in** — it only covers agents explicitly registered with Gjallarhorn; LLM calls made outside the integration are not monitored.

> **Privacy**: L2 canary checks are stateless and content-free. Gjallarhorn receives the
> LLM output, performs a token presence check, and returns a boolean result. No output
> content, user data, or conversation text is retained server-side.

### Full L2 integration example

```python
import os
from gjallarhorn_hq_sdk import GjallarhornClient
from gjallarhorn_hq_sdk import AuthError, ServiceUnavailableError

client = GjallarhornClient(api_key=os.environ['GJALLARHORN_API_KEY'])

# Step 1: Register agent once at startup — store canary_token in your secrets.
# (Registration is a one-time REST call to POST /v1/agents/register.)
CANARY_TOKEN = os.environ['AGENT_CANARY_TOKEN']

# Step 2: Inject canary into every system prompt.
def build_system_prompt(base_prompt: str) -> str:
    return f"{base_prompt}\n{CANARY_TOKEN}"

# Step 3: After every LLM call, check output integrity.
def check_agent_output(llm_output: str) -> None:
    try:
        check = client.check_canary(llm_output)
    except AuthError:
        raise RuntimeError('Canary check: invalid API key')
    except ServiceUnavailableError:
        # Canary service down — decide whether to fail-open or fail-closed
        import warnings
        warnings.warn('Canary check unavailable — proceeding with caution')
        return

    if check.alert_code == 'RAGNARÖK':
        # Canary absent: model did not echo the token it should always produce.
        # Possible cause: system prompt was overridden by an injection attack.
        raise RuntimeError(
            f'[RAGNARÖK] Agent output integrity check failed '
            f'for agent {check.agent_id} at {check.checked_at}'
        )

# Usage in your agent loop:
system_prompt = build_system_prompt('You are a helpful customer support agent.')
llm_output = call_your_llm(system_prompt, user_message)
check_agent_output(llm_output)
# Safe to proceed
```

## Error handling

```python
from gjallarhorn_hq_sdk import (
    AuthError, QuotaExceededError, RateLimitError,
    ExtractionFailedError, ServiceUnavailableError,
)

try:
    result = client.scan(input=user_input)
except AuthError:
    print('Invalid API key')
except QuotaExceededError:
    print('Monthly quota hit')
except RateLimitError as e:
    print(f'Rate limited — retry after {e.retry_after}s')
except ExtractionFailedError:
    print('Could not extract file content')
```

The SDK retries automatically on rate-limit (respecting `retry_after`) and 503 errors (exponential backoff, max 3 retries).

## License

MIT — see [LICENSE](LICENSE).

## Links

- [API Reference](https://gjallarhorn.watch/#api)
- [Documentation](https://gjallarhorn.watch/#readme)
- [Issues](https://github.com/gjallarhorn-hq/gjallarhorn/issues)
