NeutralAI Python SDK

Package Status

Published on PyPI as neutralai-sdk.

Install

python -m pip install neutralai-sdk

For async web services and batch workers, install the async extra:

python -m pip install "neutralai-sdk[async]"

To work from a local checkout instead:

cd sdk/python
pip install -e ".[async]"

Point base_url at the hosted gateway (https://api.neutralai.co.uk) in production, or at your own gateway (e.g. http://localhost:8001) for local development.

Usage (sync)

from neutralai_sdk import NeutralAIClient

client = NeutralAIClient(base_url="https://api.neutralai.co.uk", api_key="test-key")
res = client.mask("My email is user@example.com", reversible=False)
print(res.masked_text)

Usage (async)

import asyncio
from neutralai_sdk import AsyncNeutralAIClient

async def main():
    async with AsyncNeutralAIClient(base_url="https://api.neutralai.co.uk", api_key="test-key") as client:
        res = await client.mask("Call me at +90 555 123 45 67")
        print(res.masked_text)

asyncio.run(main())

The async client keeps one httpx.AsyncClient open for connection pooling and retries transient 408, 429, and 5xx responses by default.

Usage (async streaming)

import asyncio
from neutralai_sdk import AsyncNeutralAIClient

async def main():
    async with AsyncNeutralAIClient(
        base_url="https://api.neutralai.co.uk",
        service_token="signed-service-token",
    ) as client:
        async for event in client.chat_stream(
            session_id="session-123",
            prompt="Summarize this text",
            provider="openai",
            model="gpt-4-turbo",
        ):
            if event.type == "token":
                print(event.token, end="")

asyncio.run(main())

Integration registry

The SDK includes a small dependency-free adapter registry for framework wrappers. The default registry exposes langchain, and teams can register their own adapters without adding optional framework dependencies to the core package.

from neutralai_sdk import NeutralAIClient, create_integration

client = NeutralAIClient(base_url="https://api.neutralai.co.uk", agent_token="agent-token")
guard = create_integration("langchain", client, target_agent_id="agent-target")
safe_input = guard.sanitize_input({"input": "Email alice@example.com"})

Usage (document masking)

Upload a document (e.g. a PDF) and get back a redacted copy plus the list of findings. Document endpoints require a tenant-scoped service token.

import base64
from neutralai_sdk import NeutralAIClient

client = NeutralAIClient(base_url="https://api.neutralai.co.uk", service_token="signed-service-token")

with open("contract.pdf", "rb") as fh:
    pdf = fh.read()

result = client.redact_document(data=pdf, filename="contract.pdf", content_type="application/pdf")

# result.findings lists each redaction (page_number, entity_type, bounding_box).
with open("contract.redacted.pdf", "wb") as out:
    out.write(base64.b64decode(result.redacted_document_base64))

# For RAG-style ingestion that returns a sanitized reference instead of a redacted file:
ingested = client.ingest_document(data=pdf, filename="contract.pdf", label="contracts")
print(ingested.sanitized_reference.document_id)

AsyncNeutralAIClient exposes the same redact_document / ingest_document coroutines.

Supported Endpoints

• POST /v1/shield/mask
• POST /v1/shield/unmask
• POST /v1/agents/mask
• POST /v1/chat/stream
• POST /v1/documents/redact
• POST /v1/documents/ingest

POST /v1/shield/unmask requires tenant-scoped authentication via API key or service token and only restores reversible vault tokens that still match the stored scope and TTL. POST /v1/documents/* require a tenant-scoped service token.

Async Throughput

Use asyncio.gather with one shared AsyncNeutralAIClient to run many requests concurrently while reusing pooled connections:

async with AsyncNeutralAIClient(base_url="https://api.neutralai.co.uk", api_key="...") as client:
    results = await asyncio.gather(*(client.mask(prompt) for prompt in prompts))

In internal local benchmarks with mocked 100 ms network latency, 50 concurrent async requests complete about 10x faster than the same requests issued sequentially through the sync client because the async version overlaps network wait time.

To reproduce the local latency benchmark shape:

python sdk/python/examples/async_benchmark.py

Observability Hooks

You can pass observability_hook to capture SDK request lifecycle events:

• request_start
• request_success
• request_error

Diagnostics are redacted by default:

• API keys/tokens are reported as <REDACTED>
• prompt/tokenized values are summarized as <REDACTED len=N>

License

MIT — see LICENSE. This SDK is a thin, dependency-light HTTP client for the NeutralAI Gateway API; all PII detection and masking happens server-side.