tawaguard-sdk

The official Python SDK for Tawa Guard, 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

pip install tawaguard-sdk

Python 3.9+ is required. The only runtime dependency is httpx.

Quick Start

Sync

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

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_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):

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

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,
)

analyze() (and every other scan method) also accepts cuid, tenant_id, policy_overrides, and language as keyword arguments, forwarded straight to the worker per call.

Error Handling

All API and network errors raise TawaAPIError:

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

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.

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 ships tawa_input_guardrail/tawa_output_guardrail, thin wrappers over analyze() shaped as Agents SDK input_guardrail/output_guardrail. Install with pip install "tawaguard-sdk[agents]" (an optional extra, openai-agents isn't a dependency of the base package). See examples/openai_guard.py.

from agents import Agent, InputGuardrailTripwireTriggered, Runner
from tawaguard import AsyncTawaGuard
from tawaguard.agents_sdk import tawa_input_guardrail, tawa_output_guardrail

tawa = await AsyncTawaGuard.create("YOUR_API_KEY")

agent = Agent(
    name="Assistant",
    instructions="You are a helpful assistant.",
    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

Both guardrails accept TawaGuard or AsyncTawaGuard (the sync client's blocking analyze() call runs off-thread via asyncio.to_thread so it doesn't stall the Agents SDK's event loop), plus should_block, extract_text, and cuid/tenant_id/policy_overrides/language overrides, see the docstrings in src/tawaguard/agents_sdk.py for the full option set.

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_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".

Both guardrails take on_review, called with every flagged result so you can route it to a second model for an opinion:

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

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) -> ScanResult

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

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

Development

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

License

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