AgentBuck Python SDK

AgentBuck: A real-time trust enforcement layer for AI agents.

agentbuck is its Python SDK for validating records with a running AgentBuck engine over gRPC. It accepts Python records or CSV data and returns the engine's status, message, and per-record decisions.

Requirements

• Python 3.10 or newer. The package is tested on Python 3.10–3.13.
• A reachable AgentBuck engine endpoint in tcp://host:port format.
• An AgentBuck API key, agent ID, and application ID (idApp) with access to the operation.

Installation

python -m pip install agentbuck

See the package on PyPI for available versions.

Quick start

Set your API key in the process environment. For example:

# macOS / Linux
export AGENTBUCK_API_KEY="your-api-key"

# PowerShell
$env:AGENTBUCK_API_KEY = "your-api-key"

Then validate a batch of records:

import os

from agentbuck import AgentBuckClient

records = [
    {"email": "alice@example.com"},
    {"email": "bob@example.com"},
]

with AgentBuckClient(
    api_key=os.environ["AGENTBUCK_API_KEY"],
    endpoint="tcp://your-agentbuck-host:50051",
) as client:
    result = client.validate_bulk(
        agent_id="your-agent-id",
        idApp=123,  # Replace with your application ID.
        data=records,
    )

if result["status"] != "success":
    raise RuntimeError(result["message"])

if len(result["decisions"]) != len(records):
    raise RuntimeError("Decision count does not match input record count")

for record, decision in zip(records, result["decisions"]):
    print(decision, record)

Use the AgentBuckClient context manager to close its gRPC channel when finished. If you do not use a context manager, call client.close().

Input formats

validate_bulk(agent_id=..., idApp=..., data=...) accepts:

For a CSV file, the first row supplies field names; each following row becomes one record. CSV field values are read as strings. With an open client:

from pathlib import Path

result = client.validate_bulk(
    agent_id="your-agent-id",
    idApp=123,
    data=Path("records.csv"),
)

Record values should be compatible with protobuf Struct (JSON-style values). The SDK loads the input into memory and sends one bulk request; it does not stream records.

Results and errors

validate_bulk returns a dictionary with status, message, and decisions:

{
    "status": "success",
    "message": "",
    "decisions": ["PROCEED", "BLOCK"],
}

Decision values are determined by the AgentBuck engine. On a non-success engine response, the SDK returns the engine's status and message with an empty decisions list. Check status before pairing decisions with input records.

The SDK raises these exceptions for local configuration or gRPC failures:

All four inherit from AgentBuckError. Unsupported input shapes raise TypeError.

Endpoint configuration

Pass an endpoint directly, or provide AGENTBUCK_GRPC_TARGET through an environment file or the process environment:

import os

from agentbuck import AgentBuckClient

client = AgentBuckClient(
    api_key=os.environ["AGENTBUCK_API_KEY"],
    endpoint="tcp://your-agentbuck-host:50051",
    timeout_seconds=30.0,
)

When endpoint is omitted, the SDK resolves the endpoint in this order:

1. AGENTBUCK_GRPC_TARGET from the file passed as env_file=..., if one was supplied.
2. Otherwise, AGENTBUCK_GRPC_TARGET from the process environment. A .env file in the current working directory is loaded first when present; it does not override an existing process value.

For example, a custom environment file can contain:

AGENTBUCK_GRPC_TARGET=tcp://your-agentbuck-host:50051

client = AgentBuckClient(
    api_key=os.environ["AGENTBUCK_API_KEY"],
    env_file="path/to/agentbuck.env",
)

api_key is always passed explicitly to AgentBuckClient. The SDK does not automatically read AGENTBUCK_API_KEY; that variable is used only by the examples above. The default request timeout is 30 seconds.

Health check

import os

from agentbuck import AgentBuckClient

with AgentBuckClient(
    api_key=os.environ["AGENTBUCK_API_KEY"],
    endpoint="tcp://your-agentbuck-host:50051",
) as client:
    health = client.health()

print(health.status, health.message, health.version, health.rules_loaded)

health() returns a HealthResult with status, message, version, and rules_loaded fields.

Security

Keep API keys out of source control and logs. The current SDK uses an unencrypted gRPC channel for tcp:// endpoints, including the API key sent as request metadata. Connect over a trusted private network or a protected tunnel.

Support and license

Report bugs and request features in GitHub Issues. The SDK is licensed under Apache-2.0; see NOTICE for the copyright notice.