Metadata-Version: 2.5
Name: tawaguard-sdk
Version: 0.3.0
Summary: Prompt-injection detection for LLM and agent traffic. Official Python SDK for Tawa Guard.
Project-URL: Homepage, https://tawaguard.com
Project-URL: Documentation, https://docs.tawaguard.com/docs/python
Project-URL: Repository, https://github.com/Tawa-Group/tawaguard-sdk-py
Project-URL: Changelog, https://github.com/Tawa-Group/tawaguard-sdk-py/blob/main/CHANGELOG.md
Project-URL: Issues, https://github.com/Tawa-Group/tawaguard-sdk-py/issues
Author-email: Tawa Guard Ltd <support@tawaguard.com>
License-Expression: MIT
License-File: LICENSE
Keywords: agents,ai-security,guardrails,llm-security,prompt-injection,rag
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
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: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: httpx>=0.27.0
Provides-Extra: agents
Requires-Dist: eval-type-backport>=0.2.0; (python_version < '3.10') and extra == 'agents'
Requires-Dist: openai-agents>=0.8.4; extra == 'agents'
Requires-Dist: openai<2.45; (python_version < '3.10') and extra == 'agents'
Provides-Extra: dev
Requires-Dist: eval-type-backport>=0.2.0; (python_version < '3.10') and extra == 'dev'
Requires-Dist: openai-agents>=0.8.4; extra == 'dev'
Requires-Dist: openai<2.45; (python_version < '3.10') and extra == 'dev'
Requires-Dist: pyright>=1.1.0; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23.0; extra == 'dev'
Requires-Dist: pytest>=8.2.0; extra == 'dev'
Requires-Dist: respx>=0.21.0; extra == 'dev'
Requires-Dist: ruff>=0.4.0; extra == 'dev'
Description-Content-Type: text/markdown

# tawaguard-sdk

The official Python SDK for [Tawa Guard](https://tawaguard.com), prompt-injection detection for
LLM and agent traffic. It scans prompts, retrieved context, tool calls and tool output, and
returns an allow or block decision for each.

Documentation: <https://docs.tawaguard.com>

## Installation

```bash
pip install tawaguard-sdk
```

Python 3.9+ is required. The only runtime dependency is [`httpx`](https://www.python-httpx.org/).

## Quick Start

### Sync

```python
from tawaguard import TawaGuard

client = TawaGuard("YOUR_API_KEY")

result = client.analyze("Hello, world!")

print(result.action)      # "allow" | "block" | "soft_block" | "hard_block"
                           # | "review" | "sandbox" | "ambiguous" | "ood"
print(result.confidence)  # 0-1
print(result.rationale)   # human-readable explanation
```

### Async

```python
import asyncio
from tawaguard import AsyncTawaGuard


async def main():
    async with AsyncTawaGuard("YOUR_API_KEY") as client:
        result = await client.analyze("Hello, world!")
        print(result.action)


asyncio.run(main())
```

By default, the sync `TawaGuard` constructor verifies the API key against `/verify` immediately,
credential issues surface at construction time rather than on first use. For `AsyncTawaGuard`,
use `await AsyncTawaGuard.create(...)` or `async with AsyncTawaGuard(...)` to verify eagerly;
constructing it directly defers verification to the first request.

## How It Works

Each call to `analyze()` scans a plain prompt (`provenance="user_prompt"`) and returns a
`ScanResult`, the same result shape every scan method returns (`scan_response`, `scan_context`,
`scan_chunk`, `scan_tool_call`, `scan_tool_output` all share it, since they all hit the worker's scan pipeline
under the hood, just with different provenance):

| Field             | Type                     | Description                                                       |
| ----------------- | ------------------------ | ------------------------------------------------------------------ |
| `action`          | `ScanAction`             | What the gateway enforced: `"allow"` or `"block"` from a current gateway. An older gateway can still return `"soft_block" \| "hard_block" \| "review" \| "sandbox" \| "ambiguous" \| "ood"` |
| `engine_action`   | `ScanAction`             | The engine's own action before your app's review setting (for example `"review"` or `"hard_block"`). Equals `action` against an older gateway |
| `needs_review`    | `bool`                   | `True` when the engine was not confident enough to call this one cleanly. A tag on an allow or a block, not a hold. Derived as `action == "review"` against an older gateway |
| `confidence`      | `float`                  | Confidence score (0-1)                                             |
| `suggestion`      | `"allow" \| "block" \| None` | Coarse hint for the undecided actions (`ambiguous`/`review`/`sandbox`) |
| `rationale`       | `str`                    | Human-readable explanation of the decision                         |
| `policy`          | `PolicyVerdict \| None`  | Directives signaled by the worker's policy engine, if any          |
| `language`        | `LanguageInfo \| None`   | Language identification detail, `None` while language ID is disabled |
| `score_breakdown` | `dict[str, Any] \| None` | Diagnostic detail behind the decision, shape varies by which decision path fired (fusion, regex-only, OOD, ...) |

The SDK retries automatically (up to 3 attempts with exponential backoff + jitter) for rate
limits (`429`) and transient server errors (`5xx`), honoring `Retry-After` when present.

## Configuration

```python
client = TawaGuard(
    "YOUR_API_KEY",
    base_url="https://gateway.tawaguard.com/v1",     # the default; override for self-hosted or staging
    max_retries=3,
    timeout=30.0,
    verify_auth=True,
)
```

| Option        | Type          | Default                              | Description                                          |
| ------------- | ------------- | ------------------------------------ | ----------------------------------------------------- |
| `detector`    | `str \| None` | `None`                               | Opaque detector name, reserved for future use. The SDK does not send it to the service or interpret it; the service picks the detector |
| `base_url`    | `str`         | `"https://gateway.tawaguard.com/v1"` | API endpoint. The SDK never reads environment variables: pass `base_url` explicitly for staging or self-hosted, for example `base_url=os.environ.get("TAWA_BASE_URL", ...)` in your own code. An explicit value pins the client to that host and skips the region redirect |
| `max_retries` | `int`         | `3`                                | Max retry attempts for `429`/`5xx` responses          |
| `timeout`     | `float`       | `30.0`                             | Request timeout in seconds                            |
| `verify_auth` | `bool`        | `True`                             | Verify the API key against `/verify` on construction (sync client only) |

`analyze()` (and every other scan method) also accepts `cuid`, `tenant_id`, `policy_overrides`,
`language` and `session_id` as keyword arguments, forwarded straight to the worker per call.
`session_id` is an opaque id that correlates every scan in one agent turn (its prompt, tool
calls, retrieval and reply) so they read as one connected timeline. It is purely additive:
leaving it out changes nothing about how a scan is handled. (`scan_tool_call` and
`scan_tool_output` take no `language`.)

## Error Handling

All API and network errors raise `TawaAPIError`:

```python
from tawaguard import TawaGuard, TawaAPIError

try:
    client = TawaGuard("YOUR_API_KEY")
    result = client.analyze("user input")
except TawaAPIError as err:
    print(err.message)     # error description
    print(err.status_code) # HTTP status code (or None for network errors)
    print(err.retryable)   # whether a retry could succeed
```

## Usage Examples

### Basic

```python
from tawaguard import TawaGuard

def moderate_content(text: str):
    client = TawaGuard("YOUR_API_KEY")

    result = client.analyze(text)

    if result.action == "allow":
        print(f"Safe ({result.confidence:.2f} confidence)")
    elif result.action in ("block", "hard_block", "soft_block"):
        print(f"Blocked: {result.rationale}")
    elif result.action in ("ambiguous", "review", "sandbox"):
        # suggestion tells you which way the model leans
        print(f"Undecided ({result.action}), suggested action: {result.suggestion}")

    return result


moderate_content("Hello, world!")
```

### With OpenAI (raw client)

Screen user prompts before they reach the model. If Tawa blocks or leans toward blocking, the
request is rejected without an OpenAI call.

```python
from openai import OpenAI
from tawaguard import TawaGuard

openai = OpenAI(api_key="...")
tawa = TawaGuard("YOUR_API_KEY")


_BLOCKED_ACTIONS = {"block", "hard_block", "soft_block"}


def chat(user_message: str):
    scan = tawa.analyze(user_message)

    if scan.action in _BLOCKED_ACTIONS or scan.suggestion == "block":
        raise ValueError("Message blocked by content policy.")

    return openai.chat.completions.create(
        model="gpt-4o",
        messages=[{"role": "user", "content": user_message}],
    )
```

### With the OpenAI Agents SDK

`tawaguard.agents_sdk` adds Tawa Guard to the Agents SDK at four points. Install with
`pip install "tawaguard-sdk[agents]"` (an optional extra, `openai-agents` isn't a dependency of
the base package; it needs `openai-agents>=0.8.4`). The framework is imported only inside
`tawaguard.agents_sdk`, never by `import tawaguard`. See
[`examples/openai_guard.py`](examples/openai_guard.py).

| Helper | Where it runs | Scans with |
| --- | --- | --- |
| `tawa_input_guardrail` | before the agent starts | `analyze()` |
| `tawa_output_guardrail` | on the agent's final output | `analyze()` |
| `tawa_tool_input_guardrail` | before each function tool runs | `scan_tool_call()` |
| `tawa_tool_output_guardrail` | on each function tool's result | `scan_tool_output()` |

```python
from agents import Agent, InputGuardrailTripwireTriggered, Runner, function_tool
from tawaguard import AsyncTawaGuard
from tawaguard.agents_sdk import (
    guard_function_tools,
    tawa_input_guardrail,
    tawa_output_guardrail,
)

tawa = await AsyncTawaGuard.create("YOUR_API_KEY")


@function_tool
def fetch_page(url: str) -> str:
    """Fetch a web page."""
    ...


agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    tools=guard_function_tools([fetch_page], tawa),  # scans tool calls and tool results
    input_guardrails=[tawa_input_guardrail(tawa)],
    output_guardrails=[tawa_output_guardrail(tawa)],
)

try:
    result = await Runner.run(agent, user_message)
except InputGuardrailTripwireTriggered as err:
    scan = err.guardrail_result.output.output_info  # the ScanResult that tripped it
    raise ValueError(f"Message blocked: {scan.action}") from err
```

All of them accept `TawaGuard` or `AsyncTawaGuard` (the sync client's blocking call runs
off-thread via `asyncio.to_thread` so it doesn't stall the Agents SDK's event loop), plus
`should_block`, `cuid`/`tenant_id`/`policy_overrides`, `session_id`, `on_review`, `on_error` and
`fail_closed`. See the docstrings in
[`src/tawaguard/agents_sdk.py`](src/tawaguard/agents_sdk.py) for the full option set.

#### Tool guardrails

Tool results are the usual route for an indirect prompt injection: a web page, a document or an
API response that a tool returns and the model then reads. The tool guardrails scan them before
the model does.

- **Tool results are scanned by default**, with `scan_tool_outputs=False` as the one switch to turn
  that off. Every scan is a billable request, so a run with many tool calls makes many scans.
- **Tool-call arguments are scanned too**, with their own switch, `scan_tool_calls=False`.
- A blocked call or result is not shown to the model. By default the model receives a short
  message instead (`on_block="reject"`) and the run continues; the message does not repeat the
  blocked content. With `on_block="raise"` the Agents SDK raises
  `ToolInputGuardrailTripwireTriggered` or `ToolOutputGuardrailTripwireTriggered`, and the
  `ScanResult` is at `err.output.output_info`.
- `guard_function_tools(tools, tawa)` adds the guardrails to every `FunctionTool` in a list and
  keeps any guardrails already on them. To attach them yourself use
  `inputs, outputs = tawa_tool_guardrails(tawa)` and pass them as `tool_input_guardrails` /
  `tool_output_guardrails` to `function_tool(...)` or to an MCP server's constructor, or build
  one guardrail with `tawa_tool_input_guardrail` / `tawa_tool_output_guardrail`.

```python
inputs, outputs = tawa_tool_guardrails(tawa, scan_tool_calls=False)  # scan results only


@function_tool(tool_input_guardrails=inputs, tool_output_guardrails=outputs)
def search_docs(query: str) -> str: ...
```

What is not covered: the Agents SDK runs tool guardrails for function tools only (including the
function tools an MCP server exposes). Hosted tools (web search, file search, code interpreter,
hosted MCP) and other tools the model provider executes, plus computer, shell and apply-patch
tools and handoffs, are not function tools, so their inputs and results are not scanned by these
guardrails. `guard_function_tools` leaves such tools untouched. The final-output guardrail still
sees whatever the model says about their results.

#### Gateway and network errors

Every guardrail here **fails open by default**: if the Tawa Guard request fails (after the
client's retries), the error goes to `on_error(exc)` if you set one, the scan is skipped
(`output_info` is `None`), and the agent run continues. Pass `fail_closed=True` to re-raise the
error instead, which aborts the run. `on_error` is called only for errors the guardrail absorbed.

### The review flag

`needs_review` marks results the engine was less sure about. The call is still allowed or blocked
as `action` says; the flag only tells you the verdict is worth a second look. It is set on every
result the SDK returns (`analyze`, `scan_response`, `scan_context`, `scan_chunk`,
`scan_tool_call`, `scan_tool_output`, each entry of `scan_retrieval().per_chunk`), and `scan_retrieval().aggregate`
adds `needs_review` and `review_chunk_ids`. When a response omits the new fields (an older
gateway), `engine_action` is `action` and `needs_review` is `action == "review"`.

Every guardrail takes `on_review`, called with every flagged result so you can route it to a second
model for an opinion:

```python
from tawaguard.agents_sdk import tawa_input_guardrail


async def ask_second_model(text: str) -> str:
    """Your own callback: ask another model or a human queue. Placeholder."""
    ...


async def on_review(scan):
    opinion = await ask_second_model(scan.rationale)
    audit_log.append({"risk": scan.risk, "engine_action": scan.engine_action, "opinion": opinion})


guardrail = tawa_input_guardrail(
    tawa,
    on_review=on_review,  # sync or async
    on_error=lambda exc: print("review hook failed:", exc),
)
```

When `on_review` fires:

- On every scan whose `needs_review` is `True`, whether the input is then allowed or blocked.
  It does not fire when `needs_review` is `False`, including clean allows, clean blocks and
  `"ood"` results.
- Once per scan, before the guardrail returns its result, and awaited if it is async, so a slow
  callback delays that call.
- It cannot change the decision: the result is a frozen dataclass and the return value is
  ignored.
- It never raises into the guardrail. An exception goes to `on_error(exc)`, or is swallowed when
  `on_error` is not set.

`default_should_block` is unchanged: a flagged result is blocked only if the gateway blocked it
(or, against an older gateway, a `review` result carries `suggestion == "block"`).

### With FastAPI

```python
from fastapi import FastAPI, HTTPException
from tawaguard import AsyncTawaGuard

app = FastAPI()
tawa = AsyncTawaGuard("YOUR_API_KEY")


@app.post("/chat")
async def chat(message: str):
    scan = await tawa.analyze(message)

    if scan.action in ("block", "hard_block", "soft_block") or scan.suggestion == "block":
        raise HTTPException(status_code=403, detail="Message blocked by content policy.")

    return {"ok": True}


@app.on_event("shutdown")
async def shutdown():
    await tawa.aclose()
```

## API Reference

### `TawaGuard` / `AsyncTawaGuard`

#### `analyze(text, *, cuid=None, tenant_id=None, policy_overrides=None, language=None, session_id=None) -> ScanResult`

Scans a plain prompt and returns the decision. `analyze` on `AsyncTawaGuard` is a coroutine.

#### `scan_response(text, *, cuid=None, tenant_id=None, policy_overrides=None, language=None, session_id=None) -> ScanResult`

Scans the model's own reply before it reaches the end user. A reply can carry an injection aimed
at the next turn, the same threat as a tool's output, so it is sent with its own provenance
(`llm_response`) and returns the same `ScanResult`, review fields included.

```python
reply = llm.generate(prompt)
scan = client.scan_response(reply, session_id=turn_id)
if scan.action != "allow":
    reply = "Sorry, I can't share that."
```

#### `scan_context` / `scan_chunk` / `scan_retrieval` / `scan_tool_call` / `scan_tool_output`

RAG- and agent-specific scan methods, same `ScanResult` shape (`scan_retrieval` returns a
per-chunk list plus an aggregate verdict instead), but carry that provenance through to the
decision. See the docstrings on each method for details.

#### `AsyncTawaGuard.create(api_key, **kwargs) -> AsyncTawaGuard` (classmethod, async)

Constructs a client and verifies the API key against `/verify`. Raises `TawaAPIError` if
authentication fails.

#### `close()` / `aclose()`

Closes the underlying HTTP connection pool. Both clients support context manager usage
(`with` / `async with`) which calls this automatically.

### `TawaAPIError`

| Attribute     | Type            | Description                    |
| ------------- | ---------------- | ------------------------------ |
| `message`     | `str`            | Error description              |
| `status_code` | `int \| None`    | HTTP status code               |
| `retryable`   | `bool`           | Whether retrying could succeed |

## Development

```bash
pip install -e ".[dev]"
pytest          # run tests
ruff check .    # lint
pyright         # type-check
```

## License

MIT. See [LICENSE](LICENSE). This licence covers the SDK only; use of the Tawa Guard service is governed by its Terms.
