Metadata-Version: 2.4
Name: bobsentry
Version: 0.4.0
Summary: BobSentry Python SDK for runtime authorization of consequential AI-agent actions
Author: BobSentry
License-Expression: LicenseRef-Proprietary
Project-URL: Homepage, https://bobsentry.com
Project-URL: Documentation, https://app.bobsentry.com/docs/python-sdk-quickstart
Project-URL: Repository, https://github.com/mdashrraf/bobsentry
Project-URL: Developer Portal, https://bobsentry.com/developers
Project-URL: Runtime Contract, https://app.bobsentry.com/docs/runtime-integration-contract
Keywords: authorization,ai-agents,runtime,policy,hipaa,evidence
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Programming Language :: Python :: 3.14
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.12
Description-Content-Type: text/markdown
Requires-Dist: httpx>=0.27.0
Requires-Dist: pydantic>=2.7.0
Requires-Dist: typing-extensions>=4.12.0
Requires-Dist: cryptography>=43.0.0
Requires-Dist: rfc8785>=0.1.0
Provides-Extra: dev
Requires-Dist: pytest>=8.2.0; extra == "dev"
Requires-Dist: pytest-asyncio>=0.23.7; extra == "dev"

# 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:

```bash
pip install bobsentry
```

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

```bash
pip install -e bobsentry-runtime/sdk/python
```

Requires Python 3.12+.

## Authorize (public client)

```python
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)

```python
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.

```python
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:

```python
@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:

```python
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

| Option                   | Default       | Description                                                                                          |
| ------------------------ | ------------- | ---------------------------------------------------------------------------------------------------- |
| `fail_mode`              | `"closed"`    | Behavior on backend failure: `closed` (raise), `open` (allow), `observe` (allow + telemetry event).  |
| `metadata_only_transport`| `True`        | Default trust contract. Setting `False` emits a `DeprecationWarning`; supported only for migrations. |
| `max_retries`            | `2`           | Retry attempts on 5xx / network failures.                                                            |
| `retry_backoff_s`        | `0.2`         | Base seconds for exponential backoff.                                                                |
| `cache_ttl_s`            | `0.0`         | Per-metadata cache TTL. Cache keys derive from metadata only, never raw payload.                    |
| `telemetry_hook`         | `None`        | Optional callable receiving sanitized telemetry events.                                              |
| `circuit_breaker`        | `5 / 30s`     | Opens after `failure_threshold` consecutive failures; recovers after `recovery_timeout_s`.           |
| `classification`         | builtin       | Local PHI key/value patterns; allowlists for context/metadata fields.                                |

## 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:

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

Or programmatically:

```python
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`):

```python
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`](https://app.bobsentry.com/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`](../../docs/phi-boundary-architecture.md)
for the full responsibility matrix.
