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

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

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

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

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:

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

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

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.

Licence

The SDK source code is MIT licensed. Use of the Tawa Guard API and service, including the detection engines, models and documentation, is governed by the Tawa Guard Terms of Service and is not covered by the MIT licence.