BobSentry Python SDK

Runtime authorization for consequential AI-agent actions.

BobSentry returns an authorization decision (ALLOW / REQUIRE_APPROVAL / BLOCK). BobSentry does not execute downstream actions. Authorization ≠ execution ≠ outcome.

Install

Once published, install with:

pip install bobsentry

The package is currently available in private preview for design partners. Until then, install from the repository:

pip install -e bobsentry-runtime/sdk/python

Requires Python 3.12+.

Authorize (public client)

from bobsentry import BobSentry

client = BobSentry(
    api_key="...",
    base_url="https://runtime.bobsentry.com",
)

decision = client.authorize(
    agent_id="infra-agent",
    action="deploy_production",
    action_environment="production",
    context={"service": "payments"},
)

if decision.is_allowed:
    # Integrator may execute its own downstream logic.
    pass
elif decision.requires_approval:
    # Do not execute yet. A human authorizes in the BobSentry console.
    later = client.get_decision(decision.decision_id)
    # Proceed only when later.approval_status == "APPROVED".
elif decision.is_blocked:
    # Do not execute.
    pass

authorize calls POST /v1/runtime/evaluate with x-api-key. It does not wait for approval and does not run the downstream action. See examples/authorize_deploy_production.py.

Async: await client.authorize_async(...) and await client.get_decision_async(...).

Decorator guard (existing API)

from bobsentry import Sentry, SdkConfig

sentry = Sentry(
    api_key="bobsentry_live_xxx",
    agent_id="ehr-agent",
    base_url="https://runtime.bobsentry.com",
    config=SdkConfig(
        fail_mode="closed",          # closed | open | observe
        metadata_only_transport=True, # default; do not disable in production
        cache_ttl_s=0.0,
    ),
)


@sentry.guard(action="send_external_email", target="external_email")
def send_email(to: str, body: str) -> None:
    # Your business logic; BobSentry evaluates *before* this body runs.
    smtp_send(to, body)

Async functions are supported identically — decorate an async def and the SDK awaits authorization before your coroutine runs.

Protect an existing tool

protect_tool wraps any existing sync or async callable (a plain function, a LangChain StructuredTool.func, an OpenAI tool dispatcher, a CrewAI tool's _run, ...) without importing any framework. It preserves callable metadata, calls authorization exactly once, executes the callable at most once, and never serializes the callable's args/kwargs into the request.

from bobsentry import Sentry
from bobsentry.integrations import protect_tool

sentry = Sentry(api_key="...", agent_id="ops-agent")
safe_deploy = protect_tool(sentry, deploy, action="deploy_production")

Approval waiting (optional)

By default a REQUIRE_APPROVAL decision raises ApprovalRequiredError immediately. Opt in to waiting per guard; the SDK polls only the existing decision id (it never re-runs authorization) using a monotonic-clock deadline:

@sentry.guard(
    action="run_database_migration",
    wait_for_approval=True,
    approval_timeout_s=600,
    approval_poll_interval_s=5,
)
def run_migration(migration_id: str) -> str:
    return apply(migration_id)

• APPROVED -> the function runs exactly once.
• DENIED -> ExecutionDeniedError.
• EXPIRED -> ApprovalExpiredError.
• local timeout -> ApprovalTimeoutError (the ticket may still be PENDING server-side; the SDK simply stopped waiting).
• NOT_REQUIRED, missing, malformed, or unknown status -> ApprovalProtocolError; execution remains blocked.

A human approves/denies in the BobSentry console; a tenant SDK key cannot self-approve.

Local simulation (no key, no network)

Evaluate decisions entirely offline to iterate on policy before wiring up the runtime. Sentry(simulate=True) requires no API key and makes no network calls:

sentry = Sentry(agent_id="dev-agent", simulate=True)
sentry.simulation_engine.add_policy("external_api_call", "REQUIRE_APPROVAL")

@sentry.guard(action="delete_repository")
def wipe(name: str) -> None:
    ...  # simulated decision is BLOCK -> ExecutionBlockedError

Simulated decisions are non-production: source="LOCAL_SIMULATION", signed_evidence=False, production_authoritative=False, and policy_id begins with "LOCAL-SIM-". Simulation refuses to run under a detected production environment (e.g. BOBSENTRY_ENV=production) unless you pass simulate_unsafe_override=True, and bobsentry verify rejects simulation records as NON-PRODUCTION / NOT VERIFIABLE. The engine loads a snapshot of the canonical taxonomy generated from packages/taxonomy/src/actions.ts.

Under the hood, when send_email is called:

1. The SDK builds an ActionEnvelope, classifies the args locally for PHI, computes a payload_hash, and builds a metadata-only request.
2. The request is POSTed to /v1/runtime/evaluate with header x-bobsentry-metadata-only: true. The raw payload is never transmitted.
3. The runtime returns a signed RuntimeDecision. The SDK raises ExecutionBlockedError or ApprovalRequiredError if the decision blocks the action; otherwise execution proceeds.

Configuration

Fail modes

• closed (recommended): if the runtime is unreachable, the action is blocked with ExecutionBlockedError. Highest safety; surfaces outages immediately.
• open: if the runtime is unreachable, the action is allowed. Use only where a fail-open posture is documented and accepted by your risk function.
• observe: if the runtime is unreachable, the action is allowed AND a sanitized observe_failure telemetry event is emitted. Useful for migration and observability windows before flipping to closed.

Idempotency

/v1/runtime/evaluate accepts a nonce (UUID or ULID) per request. Repeating the same nonce within the runtime's replay window returns 409 nonce_replay_detected. The SDK generates a fresh nonce per call by default; override correlation_id to group retries deterministically.

Verification

Export an evidence bundle from your tenant console and verify it offline:

bobsentry verify ./evidence-bundle.json
# {"verified": true, "checks": [...]}

Or programmatically:

from bobsentry import verify_bundle

report = verify_bundle("./evidence-bundle.json")
assert report.verified, report.error

Test that your integration honors BobSentry

These tests verify your integration's behavior at the authorization boundary. They do not prove that BobSentry controls the downstream system.

Conceptual flow (see tests/conformance_helpers.py and tests/test_conformance_authorization_boundary.py):

from bobsentry.models import RuntimeDecision

mutation_called = 0

def deploy() -> None:
    global mutation_called
    mutation_called += 1

def should_proceed(decision: RuntimeDecision) -> bool:
    if decision.decision == "ALLOW":
        return True
    return (
        decision.decision == "REQUIRE_APPROVAL"
        and decision.approval_status == "APPROVED"
    )

# Fixture: REQUIRE_APPROVAL + PENDING must not call deploy.
# Fixture: REQUIRE_APPROVAL + APPROVED may call deploy exactly once.
# Fixture: BLOCK must not call deploy.

if should_proceed(decision):
    deploy()

assert mutation_called in (0, 1)

Canonical product walkthrough: /docs/deploy-production-quickstart (Protect a production deployment with BobSentry).

Trust boundary

The SDK is the trust boundary. By default the runtime never receives raw payloads, raw clinical text, attachments, or model outputs. See bobsentry-runtime/docs/phi-boundary-architecture.md for the full responsibility matrix.