Python SDK

A drop-in wrapper for the OpenAI and Anthropic clients, adding guardrails and audit logging.

Installation

pip install signalvault
# add Anthropic support:
pip install "signalvault[anthropic]"

Requires Python 3.9 or newer. openai and httpx are installed as dependencies; anthropic comes from the extra. Quote the extra — most shells, zsh included, treat bare brackets as a glob.

Choosing a client

There are four classes, one per provider and per sync or async style. They take the same options and behave the same way; only the provider key argument and the call surface differ.

Configuration

Usage

from signalvault import SignalVaultClient
import os

client = SignalVaultClient(
    api_key="sk_live_...",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://api.signalvault.io",
)

response = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Hello!"}],
    sv_metadata={"user_id": "u_123"},
)
print(response.choices[0].message.content)

Anthropic

import os
from signalvault import AnthropicSignalVaultClient

client = AnthropicSignalVaultClient(
    api_key="sk_live_...",
    anthropic_api_key=os.environ["ANTHROPIC_API_KEY"],
)

message = client.messages.create(
    model="claude-sonnet-5",
    max_tokens=1024,
    messages=[{"role": "user", "content": "Hello!"}],
)

Handling a blocked request

The SDK does not hand you the decision. When a rule blocks a prompt it raises SignalVaultBlockedError and the provider is never called:

from signalvault import SignalVaultBlockedError

def ask(client, user_input):
    try:
        return client.chat.completions.create(
            model="gpt-4o-mini",
            messages=[{"role": "user", "content": user_input}],
        )
    except SignalVaultBlockedError as err:
        # Never reached the provider. err.violations is a list of Violation objects.
        raise PolicyRefused(
            "That request was blocked by policy.",
            [v.type for v in err.violations],
        ) from err

The exception carries request_id, violations and, when available, dashboard_url. Violations are objects, so read v.type rather than indexing them.

Streaming

Streaming works for both providers, sync and async. The audit event is sent once the stream finishes, and a stream you abandon early still records what was received.

stream = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Write a poem"}],
    stream=True,
)

for chunk in stream:
    # The final chunk carries usage and no choices — guard before indexing.
    if chunk.choices:
        print(chunk.choices[0].delta.content or "", end="")

The async clients return an iterator you await, then iterate with async for:

import asyncio
import os
from signalvault import AsyncSignalVaultClient

async def main():
    client = AsyncSignalVaultClient(
        api_key="sk_live_...",
        openai_api_key=os.environ["OPENAI_API_KEY"],
    )
    stream = await client.chat.completions.create(
        model="gpt-4o-mini",
        messages=[{"role": "user", "content": "Hello!"}],
        stream=True,
    )
    async for chunk in stream:
        if chunk.choices:
            print(chunk.choices[0].delta.content or "", end="")
    await client.aclose()

asyncio.run(main())

When SignalVault is unreachable

fail_mode decides what happens when the guardrail check times out or fails. The default is "open":

• "open" — the request proceeds unchecked and a SignalVaultWarning is issued. Traffic keeps flowing; your guardrails do not apply to it.
• "closed" — the SDK raises SignalVaultUnavailableError and the provider is never called.

To make a silent fail-open impossible to miss, promote the warning: warnings.simplefilter("error", SignalVaultWarning). When SignalVault refuses the check outright — a bad key, an inactive account, or the trial limit reached — nothing is logged for that call at all.

Over the rule-check limit

Rule checks are limited to 120 requests a minute per app. Over it, SignalVault still records the request, marked unchecked, but runs no rules on it and stores no prompt, so there is no decision and fail_mode applies: with "open" the request goes to the provider with a warning, and its response and tool calls are recorded as usual; with "closed" the SDK raises SignalVaultUnavailableError and the provider is never called. The SDK sends its fail_mode with the check, so the record says whether the request was sent on. Requires version 0.7.0 or later.

Mirror Mode

In mirror mode the provider is called first and the audit events follow, off the request path. Nothing is ever blocked:

client = SignalVaultClient(
    api_key="sk_live_...",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    base_url="https://api.signalvault.io",
    mirror_mode=True,
)

Auditing tool calls

client.tool(name, fn) wraps a callable — it also works as a decorator — so every call is recorded as an agent.tool_call event. client.tools.record(...) records one by hand, and picks up the request id from a surrounding with_context block.

Before your process exits

Audit events are sent in the background, so in a short-lived process call client.flush() before exiting or events still in flight are abandoned. close() — or aclose() on an async client — flushes and then shuts the connection pool, after which the client cannot be reused; both work as context managers and wait up to 10 seconds. At interpreter exit, queued events get up to exit_flush_timeout (10 seconds by default). From version 0.7.0, a failed send is retried, up to 6 attempts over about two minutes, with the same event_id so nothing is stored twice. An event is dropped only when 1,000 are already waiting, after its last attempt, or at a close or exit deadline. Each drop is logged to the signalvault logger with its event_id, and client.dropped_events counts them by cause.