signalvault

AI audit logs and guardrails for your OpenAI and Anthropic Python applications.

Installation

# OpenAI only
pip install signalvault openai

# Anthropic only
pip install signalvault[anthropic]

# Both
pip install signalvault openai signalvault[anthropic]

Quick Start — OpenAI (sync)

import os
from signalvault import SignalVaultClient

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

# Use exactly like OpenAI SDK
response = client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Hello!"}],
)
print(response.choices[0].message.content)

Quick Start — OpenAI (async, FastAPI / async Django)

import os
from signalvault import AsyncSignalVaultClient

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

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

Quick Start — Anthropic

import os
from signalvault import AnthropicSignalVaultClient

client = AnthropicSignalVaultClient(
    api_key="sk_live_your_signalvault_key",
    anthropic_api_key=os.environ["ANTHROPIC_API_KEY"],
    base_url="https://api.signalvault.io",
)

response = client.messages.create(
    model="claude-3-5-sonnet-20241022",
    messages=[{"role": "user", "content": "Hello!"}],
    max_tokens=1024,
)
print(response.content[0].text)

Streaming

Streaming is fully supported for all clients and providers:

# OpenAI streaming (sync)
with client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Write a poem"}],
    stream=True,
) as stream:
    for chunk in stream:
        if chunk.choices:
            print(chunk.choices[0].delta.content or "", end="", flush=True)

# OpenAI streaming (async)
stream = await async_client.chat.completions.create(
    model="gpt-4",
    messages=[{"role": "user", "content": "Write a poem"}],
    stream=True,
)
async with stream:
    async for chunk in stream:
        if chunk.choices:
            print(chunk.choices[0].delta.content or "", end="", flush=True)

# Anthropic streaming
with anthropic_client.messages.create(
    model="claude-3-5-sonnet-20241022",
    messages=[{"role": "user", "content": "Write a poem"}],
    max_tokens=1024,
    stream=True,
) as stream:
    for event in stream:
        if event.type == "content_block_delta" and event.delta.type == "text_delta":
            print(event.delta.text, end="", flush=True)

Streams yield what the provider SDK's create(..., stream=True) yields: OpenAI chunks (apart from the usage chunk described below), or Anthropic's raw stream events (message_start, content_block_delta, ...). The request is sent when create() is called, so provider errors such as an invalid key are raised there, and no response is recorded for them.

A stream is recorded once: when it ends, when it raises, or when it's closed. A for loop doesn't close the stream on break, so use with / async with (or call stream.close() / await stream.close()) if you may stop reading early. That records the response and closes the connection to the provider right away. Without it, both happen when the stream is garbage-collected (with the async clients, at the latest when asyncio.run finishes, which then waits up to 5 seconds for the event to be sent), even if it was never read. Like the provider SDK's stream, it also has stream.response, the HTTP response.

For OpenAI streams, the SDK sets stream_options={"include_usage": True} so the response is logged with its token counts. OpenAI then sends one extra final chunk that carries only the usage and no choices; the SDK consumes that chunk rather than passing it to you. If you set include_usage yourself, your setting is kept: with True you receive the usage chunk, whose choices list is empty (hence the if chunk.choices: guard above), and with False the response is logged without token counts. Other stream options you pass are kept. If your OpenAI-compatible server (set through OPENAI_BASE_URL) rejects stream_options, pass stream_options=None and the SDK sends none; responses are then logged without token counts.

Agent Tool-Use Capture

Log every tool invocation made by your agents — name, input, output, duration, and any error — as auditable events alongside your LLM calls.

Wrapper API (recommended)

@client.tool("fetch_weather")
def fetch_weather(city: str):
    res = httpx.get(f"https://api.example.com/weather?city={city}")
    return res.json()

# Call it like the original — SignalVault auto-times and audits.
weather = fetch_weather("London")

The wrapper records the call asynchronously (no impact on your tool's latency) and passes the result through unchanged. Errors are recorded and re-raised.

# Async client + async tool
@async_client.tool("fetch_weather")
async def fetch_weather(city: str):
    async with httpx.AsyncClient() as http:
        res = await http.get(f"https://api.example.com/weather?city={city}")
        return res.json()

weather = await fetch_weather("London")

Manual API

# Sync — blocks on the first HTTP send so a rejected event (4xx) warns at your call;
# if SignalVault is unreachable or rate limiting, the event is retried in the background
client.tools.record(
    tool_name="fetch_weather",
    tool_input={"city": "London"},
    tool_output={"temp": 12.3},
    duration_ms=142,
)

# Async
await async_client.tools.record(
    tool_name="fetch_weather",
    tool_input={"city": "London"},
    tool_output={"temp": 12.3},
    duration_ms=142,
)

Linking tool calls to a parent LLM turn

Wrap your agent loop in with_context and tool calls inside auto-correlate to the given request_id:

# Sync
with client.with_context(request_id="agent-turn-abc"):
    llm_response = client.chat.completions.create(...)
    fetch_weather("London")  # auto-linked to 'agent-turn-abc'

# Async
async with async_client.with_context(request_id="agent-turn-abc"):
    llm_response = await async_client.chat.completions.create(...)
    await fetch_weather("London")  # auto-linked to 'agent-turn-abc'

Without with_context, tool calls are recorded as orphans (no parent request).

What gets captured (and what to keep out)

When you wrap a tool or call tools.record(), SignalVault captures:

• tool_name (truncated to 200 bytes)
• tool_input — the function arguments, JSON-serialized (capped at 256 KB; oversize values are truncated with a marker)
• tool_output — the function return value, JSON-serialized (same 256 KB cap)
• error if the tool raises (truncated to 1900 bytes)
• duration_ms, started_at, and any metadata you attach

tool_input and tool_output are encrypted at rest server-side. tool_name, error, started_at and metadata are stored as sent, not encrypted. All of it goes on the wire to SignalVault's API: if you pass user PII, secrets, or API keys as tool arguments, those values will leave your process and be stored in SignalVault. Recommendations:

• Sanitize sensitive arguments before invoking the wrapped tool, or use the manual tools.record() API and pass a redacted copy.
• Don't put secrets in error messages — they end up in error verbatim and unencrypted.
• Use metadata for non-sensitive identifiers (user_id, feature, workspace_id); avoid putting raw user content in metadata.

Metadata

Attach contextual metadata to every event for user attribution, analytics, and audit trails:

# Set defaults at client level
client = SignalVaultClient(
    api_key="sk_live_...",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    metadata={"workspace_id": "ws_abc", "env": "production"},
)

# Override per-call
response = client.chat.completions.create(
    model="gpt-4",
    messages=[...],
    sv_metadata={"user_id": "u_123", "feature": "support-chat"},
)

sv_metadata goes to SignalVault. A plain metadata= argument is passed through to the provider untouched (for example Anthropic's metadata={"user_id": ...}).

When SignalVault Is Unavailable

The pre-flight guardrail check is in your request's critical path. What the SDK does with its answer, by fail_mode:

The fallback ai.request has the call's request_id and is marked preflight_unavailable, with preflight_status (the HTTP status, or null when there was none) and preflight_reason (timeout, network, redirect, invalid_response, unserializable, SignalVault's own error code, or http_<status>). It is sent as soon as the call goes ahead, before the provider call, so a provider error or a stream that is never read still leaves it, and it is retried on the background queue like any audit event. After a 400, 413, 422 or another 4xx about the request, and for a request that can't be serialized, it is in its minimal form: without the messages and sv_metadata, which SignalVault would refuse again. With a key SignalVault refuses (401, 402, 403, or the trial limit), it refuses the record too: after a 401, 402 or 403 the record is not retried and the SDK warns as below, and after the trial limit it is retried, then logged as dropped.

A refused key. When SignalVault refuses an audit event with 401, 402 or 403 (the fallback ai.request, a response, a tool call or a mirror-mode request), it refuses every event from that key, so requests are not being recorded. The SDK warns once per process for each of the three statuses, whether or not debug is on: on the signalvault logger at WARNING, which a warnings filter does not silence and which Python prints to stderr if your app doesn't configure logging, and as a SignalVaultWarning. The message names the status and what to check:

[SignalVault] SignalVault refused an audit event (401): requests are not being recorded. Check the API key at signalvault.io.

After a 402 it says to check the account, and after a 403 the key's environment and account. An event refused for its content (400, 422) warns only with debug=True (Event not delivered).

No server text in messages. A warning, an error or a log line never quotes text from SignalVault's answer. It gives the HTTP status and, when the answer carries one, SignalVault's code, such as (402: subscription_inactive) or (403: environment_not_allowed); anything else gives the bare status, such as (403). SignalVaultBlockedError and the debug=True warning for a warn decision name the violations' types (contains_pii, keyword_blocklist...); the violations themselves are on the error's violations, and SignalVault stores them with the request.

The pre-flight check tells SignalVault which fail_mode the client uses, so the audit log shows whether a request that could not be checked went to the provider. fail_mode applies to the pre-flight check, so it has no effect in mirror mode.

Rate limit. Rule checks are limited to 120 requests per minute per app. Above that, SignalVault still records the request, without running your rules, and answers 429:

• With fail_mode="open" the request goes to the provider unchecked, with a SignalVaultWarning. Its response and tool calls are recorded as usual, so the audit log stays complete.
• With fail_mode="closed" the SDK raises SignalVaultUnavailableError (status 429) and the provider is never called.

If you run at that volume and need every request checked, use fail_mode="closed". Responses and tool calls do not use rule checks; mirror-mode requests do. Every event type shares one ceiling of 3,000 events a minute per app.

Content the rules cannot read, and scans that do not finish. When a rule that reads content is on (contains_pii, contains_secret, keyword_blocklist, regex_match or prompt_injection) and none of them has the action block, SignalVault records two more kinds of request unchecked and answers the same 429 with "recorded": true, without Retry-After:

• a request carrying content the rules cannot read — an image, audio, a file, a PDF or a content type they do not know — with the reason unscanned_content (SignalVault 0.18.9 and later). The audit event names each content type;
• a request whose rule scan did not finish in time or stopped before the end of the text, with the reason scan_timeout.

fail_mode applies as above, and the SignalVaultWarning, or the message of SignalVaultUnavailableError, gives the reason:

[SignalVault] The request carries content SignalVault's rules cannot read (an image, audio, a file, a PDF or a content type they do not know), so SignalVault recorded it unchecked (429: unscanned_content). Guardrails were NOT applied — it is being sent to the provider (fail_mode='open').

Under a block rule that reads content, such a request is blocked instead (see Error Handling).

client = SignalVaultClient(
    api_key="sk_live_...",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    fail_mode="closed",      # "open" | "closed". Default: "open"
    preflight_timeout=2.0,   # seconds — pre-flight check. Default: 2.0
    timeout=30.0,            # seconds — background/post-flight calls. Default: 30.0
)

base_url defaults to https://api.signalvault.io. Plain http:// is refused except for localhost, as is a URL with credentials (user:password@: pass the key as api_key), a query string or a fragment. These errors never quote the URL.

Background delivery

Audit events (the fallback ai.request, responses, mirror-mode events, tool calls) are queued in memory and sent in the background, without blocking your call:

• Retries. An event that fails with a connection error, a timeout, a 5xx or a 429 is retried with exponential backoff: up to 6 attempts over about two minutes (waits of about 4, 8, 16, 32 and 60 seconds). A 429's Retry-After is honoured, up to 60 seconds. Other 4xx responses mean the event was rejected, and it is not retried. A 429 with "recorded": true is not retried either: SignalVault has stored the request, unchecked, and would answer a retry the same way, so it counts as delivered and its response is sent. Each event carries an event_id, so a retry is never double-counted. tools.record() makes the first attempt itself; if that one can be retried, the rest happen in the background.
• Queue. Up to 1,000 events per client wait to be sent. When the queue is full, the oldest are dropped to make room.
• Order. A call's request is sent before its response, retries included.
• While flush(), close(), aclose() or the exit flush is waiting, retries wait at most a second, so queued events go out within the timeout you gave.

The sync clients send from up to 4 background threads per client; the async clients send from tasks on the running event loop.

Dropped events

Dropping is the last resort. An event is dropped only when:

• every attempt failed (retries_exhausted),
• the queue was full (queue_full),
• it was still unsent when close() / aclose() or the exit flush reached its timeout, or when the async task sending it was cancelled, usually because the event loop stopped (shutdown),
• or it could not be serialized to JSON (unserializable).

Every drop is logged on the signalvault logger at WARNING, one record per occurrence, with the number of events, the cause and their ids:

SignalVault dropped 1 audit event(s), every attempt failed (6 attempts, last: rate limited, 429): event_id=4f6c...

A fallback ai.request (the SDK's record of a request sent without a decision) has no event_id, so it is logged by its request_id, marked as such. The record also carries the cause and the ids as record.signalvault_drop_cause and record.signalvault_dropped. If your app doesn't configure logging, Python prints these records to stderr. The SDK also emits a SignalVaultWarning, at most once a minute per cause; the log has every drop.

client.dropped_events counts what a client has dropped, by cause. It is a read-only snapshot, on sync and async clients:

dropped = client.dropped_events      # DroppedEvents(retries_exhausted=0, queue_full=0, shutdown=0, unserializable=0)
dropped.retries_exhausted
dropped.total

Shutting Down

Background events are queued in memory, so wait for them before a process exits or is frozen.

Serverless and short-lived processes (AWS Lambda, Cloud Functions, Cloud Run jobs, scripts, Celery tasks): call flush() at the end of each invocation, before you return. A function's process can be frozen as soon as the handler returns, and events still in the queue then are only sent if it is thawed again.

def handler(event, context):
    response = client.chat.completions.create(...)
    client.flush()           # waits up to 5 seconds by default; flush(timeout=...) to change it
    return response

# async
await async_client.flush()

flush() keeps the client open, and events still unsent at its timeout stay queued (with a SignalVaultWarning).

When you are done with a client, close it:

client.close()          # sync: sends queued events (up to 10 seconds), then closes the connection pool
await client.aclose()   # async: same; required, since pending async events are lost if the loop ends first

close() and aclose() take a timeout in seconds (default 10). Nothing is dropped before it runs out. Events still unsent then can't be sent once the connection pool is closed, so they are dropped and logged with their ids. An event waiting to be retried can take up to two minutes through an outage, so pass a longer timeout if you need to wait that long.

Interpreter exit. If a sync client is never closed, the SDK flushes it when the interpreter exits, for up to exit_flush_timeout seconds (default 10), and never holds the process longer. Events still unsent then are dropped and logged with their ids. Set it when you create the client; 0 sends nothing more at exit (the events left are still logged):

client = SignalVaultClient(
    api_key="sk_live_...",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    exit_flush_timeout=10.0,  # seconds. Default: 10.0
)

An async client can't send at exit, because its event loop has ended: await client.aclose() before the loop ends. Events an async client still holds at exit are logged as dropped.

Error Handling

from signalvault import SignalVaultBlockedError, SignalVaultUnavailableError

try:
    response = client.chat.completions.create(model="gpt-4", messages=[...])
except SignalVaultBlockedError as e:
    print("Blocked by guardrails:", [v.type for v in e.violations])
    print("Details:", e.dashboard_url)
except SignalVaultUnavailableError as e:
    # Only raised with fail_mode="closed"
    print("Guardrail check unavailable:", e.status)

Both are subclasses of RuntimeError.

A request carrying content the rules cannot read, under a block rule that reads content, is blocked with a violation of type unscanned_content_guard; its details hold a description and content_types, naming each type (SignalVault 0.18.9 and later). One whose rule scan did not finish is blocked with scan_timeout_guard.

Warnings

Everything the SDK warns about is a SignalVaultWarning. A guardrail warn decision (emitted with debug=True) uses the subclass SignalVaultViolationWarning, so you can treat policy warnings separately from delivery problems:

import warnings
from signalvault import SignalVaultViolationWarning

warnings.simplefilter("error", SignalVaultViolationWarning)

Mirror Mode

In mirror mode, requests go directly to the AI provider first and SignalVault audits them asynchronously — no latency added, never blocks:

client = SignalVaultClient(
    api_key="sk_live_...",
    openai_api_key=os.environ["OPENAI_API_KEY"],
    mirror_mode=True,
)

Features

• Automatic Logging — Requests and responses are recorded, including requests sent to the provider without a guardrail decision
• Pre-flight Guardrails — Block requests that break your rules before they reach the AI provider
• Redaction at rest — Rules with the redact action remove matches from what SignalVault stores. The request sent to the AI provider is not modified.
• PII Detection — Detect emails, phone numbers, SSNs
• Secret Detection — Block API keys and tokens
• Token Limits — Enforce cost controls
• Model Allowlists — Restrict which models can be used
• Streaming — Full streaming support for OpenAI and Anthropic
• Async Support — AsyncSignalVaultClient and AsyncAnthropicSignalVaultClient for async codebases
• Mirror Mode — Observe without blocking
• Metadata — Tag every event with user_id, feature, workspace_id, etc.
• Multi-provider — OpenAI and Anthropic/Claude support

License

MIT