Hivetrace SDK

API 2.5 / SDK 2.0

Python 3.10+. See the migration guide, HTTP contract, architecture and changelog.

input() and output() return analysis results. The calling application decides how to use those results and whether to allow a message or action. Inspect status and errors; accepted and partial results are not full success. Input and output files upload before analysis.

Framework adapters are not supported or included in SDK 2.0. Integrate through SyncHivetraceSDK or AsyncHivetraceSDK using input, output and function_call / tool_call.

The returned ProcessResponse is a Pydantic model. Its analysis sections are dictionary subclasses: old [], .get(), in, .items(), json.dumps() and in-place mutations work, along with attribute access such as response.guardrails.flagged. Use response.model_dump() for a complete dictionary. Missing server keys stay absent from section dictionaries. The SDK supplies a UTC receipt timestamp when the server omits timestamp. For DataClean, explicitly present text fields also populate missing legacy flat keys (cleaned_text, types, entities), including dictionary and JSON exports. See the migration guide for mutation behavior when both layouts are present.

censor_only is not a supported no-storage option. Ordinary ingest can persist messages; do not use that legacy flag as a data-retention control.

Configure timeout/TLS/pools using ClientOptions. For a private CA, pass these options to either SDK client; certificate and hostname verification stay enabled:

import ssl
from hivetrace import ClientOptions

options = ClientOptions(verify=ssl.create_default_context(cafile="/path/to/ca.pem"))

API keys should be scoped to the application. /clean_request and /override remain direct API operations, outside this SDK.

Overview

The Hivetrace SDK lets you integrate with the Hivetrace service to monitor user prompts and LLM responses. It supports both synchronous and asynchronous workflows and can be configured via environment variables.

Installation

Install from PyPI:

pip install "hivetrace[base]>=2.0,<3"

Quick Start

from hivetrace import SyncHivetraceSDK, AsyncHivetraceSDK

You can use either the synchronous client (SyncHivetraceSDK) or the asynchronous client (AsyncHivetraceSDK). Choose the one that fits your runtime.

Synchronous Client

Initialize (Sync)

# The sync client reads configuration from environment variables or accepts an explicit config
client = SyncHivetraceSDK()

Send a user prompt (input)

response = client.input(
    application_id="your-application-id",  # Obtained after registering the application in the UI
    message="User prompt here",
)

# Optionally attach files (filename, bytes, mime_type)
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = client.input(
    application_id="your-application-id",
    message="User prompt with files",
    files=files,
)

Send an LLM response (output)

response = client.output(
    application_id="your-application-id",
    message="LLM response here",
)

# With files
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = client.output(
    application_id="your-application-id",
    message="LLM response with files",
    files=files,
)

Asynchronous Client

Initialize (Async)

# The async client can be used as a context manager
client = AsyncHivetraceSDK()

Send a user prompt (input)

response = await client.input(
    application_id="your-application-id",
    message="User prompt here",
)

# With files (filename, bytes, mime_type)
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = await client.input(
    application_id="your-application-id",
    message="User prompt with files",
    files=files,
)

Send an LLM response (output)

response = await client.output(
    application_id="your-application-id",
    message="LLM response here",
)

# With files
files = [
    ("doc1.txt", open("doc1.txt", "rb"), "text/plain"),
]
response_with_files = await client.output(
    application_id="your-application-id",
    message="LLM response with files",
    files=files,
)

Example with Additional Parameters

response = client.input(
    application_id="your-application-id",
    message="User prompt here",
    additional_parameters={
        "session_id": "your-session-id",
        "user_id": "your-user-id",
        "agents": {
            "agent-1-id": {"name": "Agent 1", "description": "Agent description"},
            "agent-2-id": {"name": "Agent 2"},
            "agent-3-id": {}
        },
    }
)

API

The signatures below use HivetraceResponse, exported from hivetrace. Processing calls normally return ProcessResponse; transport and response-validation failures return error models. See the HTTP contract for status semantics and error handling.

input() and output() share attachment preparation. File keys are computed as sha256:<digest> from the uploaded bytes, replacing even an explicitly supplied Attachment.key. Seekable streams are uploaded from the beginning and their original position is restored; non-seekable streams are consumed from their current position. Caller-owned streams remain open.

Async snapshot preparation and cleanup run in worker threads. Cancellation waits for them to finish before returning; do not access a stream concurrently with an in-flight call.

input

# Sync
def input(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> HivetraceResponse: ...

# Async
async def input(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> HivetraceResponse: ...

Sends a user prompt to Hivetrace.

• application_id — Application identifier (must be a valid UUID, created in the UI)
• message — The user prompt
• additional_parameters — Optional dictionary with extra context (session, user, agents, etc.)
• files — Optional list of tuples (filename: str, content: bytes, mime_type: str); files are attached to the created analysis record

The application decides what to do with the analysis. SDK 2.0 removes the legacy blocked field. Apply your application's own decision rules to the returned analysis.

Response example:

{
  "status": "success",
  "tokens": {"count": 9, "usage_severity": null},
  "guardrails": {"flagged": false, "truncated": false}
}

output

# Sync
def output(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> HivetraceResponse: ...

# Async
async def output(
    application_id: str,
    message: str,
    additional_parameters: dict | None = None,
    files: list[tuple[str, bytes, str]] | None = None,
) -> HivetraceResponse: ...

Sends an LLM response to Hivetrace.

• application_id — Application identifier (must be a valid UUID, created in the UI)
• message — The LLM response
• additional_parameters — Optional dictionary with extra context (session, user, agents, etc.)
• files — Optional list of tuples (filename: str, content: bytes, mime_type: str)

Files are uploaded by request ID before response analysis using the shared attachment preparation described above.

Analysis uses the filenames returned by the upload endpoint; incomplete receipts or normalized filename collisions stop analysis and return an error with the upload receipt attached.

The application decides what to do with the analysis. SDK 2.0 removes the legacy blocked field. Apply your application's own decision rules to the returned analysis.

Response example:

{
  "status": "success",
  "tokens": {"count": 21, "usage_severity": null},
  "guardrails": {"flagged": false, "truncated": false}
}

Sending Requests in Sync Mode

def main():
    # option 1: context manager
    with SyncHivetraceSDK() as client:
        response = client.input(
            application_id="your-application-id",
            message="User prompt here",
        )

    # option 2: manual close
    client = SyncHivetraceSDK()
    try:
        response = client.input(
            application_id="your-application-id",
            message="User prompt here",
        )
    finally:
        client.close()

main()

Sending Requests in Async Mode

import asyncio

async def main():
    # option 1: context manager
    async with AsyncHivetraceSDK() as client:
        response = await client.input(
            application_id="your-application-id",
            message="User prompt here",
        )

    # option 2: manual close
    client = AsyncHivetraceSDK()
    try:
        response = await client.input(
            application_id="your-application-id",
            message="User prompt here",
        )
    finally:
        await client.close()

asyncio.run(main())

Closing the Async Client

await client.close()

Configuration

The SDK reads configuration from environment variables:

• HIVETRACE_URL — Base URL allowed to call.
• HIVETRACE_ACCESS_TOKEN — API token used for authentication.

These values are read from the process environment when you create a client.

Configuration Sources

Hivetrace SDK can retrieve configuration from the following sources:

.env File:

HIVETRACE_URL=https://your-hivetrace-instance.com/api
HIVETRACE_ACCESS_TOKEN=your-access-token  # obtained in the UI (API Tokens page)

Load the file explicitly before constructing the client:

from dotenv import load_dotenv
from hivetrace import SyncHivetraceSDK

load_dotenv()
client = SyncHivetraceSDK()

The SDK reads the process environment; it does not load .env files itself.

You can also pass a config dict explicitly when creating a client instance.

client = SyncHivetraceSDK(
    config={
        "HIVETRACE_URL": HIVETRACE_URL,
        "HIVETRACE_ACCESS_TOKEN": HIVETRACE_ACCESS_TOKEN,
    },
)

Environment Variables

Set up your environment variables for easier configuration:

# .env file
HIVETRACE_URL=https://your-hivetrace-instance.com/api
HIVETRACE_ACCESS_TOKEN=your-access-token
HIVETRACE_APP_ID=your-application-id

License

This project is licensed under Apache License 2.0.

Development

python -m pip install -e '.[dev]'
ruff check hivetrace tests setup.py
python -m unittest discover -s tests
python -m build