Metadata-Version: 2.5
Name: gebsecure
Version: 0.2.1
Summary: Official Python SDK for the GebSecure AI Agent Security & Governance Platform
Project-URL: Homepage, https://gebsecure.dev
Project-URL: Documentation, https://docs.gebsecure.dev/sdks/python
Project-URL: Source, https://github.com/gebsecure/gebsecure/tree/main/sdks/python
Project-URL: Changelog, https://github.com/gebsecure/gebsecure/blob/main/sdks/python/CHANGELOG.md
Author: GebSecure
License: Apache-2.0
Keywords: ai-agents,authorization,gebsecure,governance,sdk,security
Classifier: Development Status :: 4 - Beta
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.9
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Typing :: Typed
Requires-Python: >=3.9
Requires-Dist: cryptography>=41
Requires-Dist: httpx<1,>=0.25
Requires-Dist: pydantic<3,>=2.5
Provides-Extra: dev
Requires-Dist: mypy>=1.8; extra == 'dev'
Requires-Dist: pytest-asyncio>=0.23; extra == 'dev'
Requires-Dist: pytest>=7.4; extra == 'dev'
Requires-Dist: ruff>=0.5; extra == 'dev'
Description-Content-Type: text/markdown

# GebSecure Python SDK

Official Python SDK for the **GebSecure AI Agent Security & Governance Platform**: authorize agent
actions, execute connector actions through the secure runtime, wait for human approvals, and manage
agents, policies and delegations.

- Sync (`GebSecure`) and asyncio (`AsyncGebSecure`) clients on `httpx`
- Typed pydantic v2 models generated from the OpenAPI contract (`py.typed`, `mypy --strict` clean)
- API keys, client secrets, `private_key_jwt` (Ed25519 / ES256 / RS256) and delegation token exchange,
  with token caching, early refresh and single-flight fetching
- Retries with exponential backoff + jitter, `Retry-After`, automatic idempotency keys
- Typed errors for every problem+json code, W3C `traceparent` propagation

Requires Python 3.9+.

```bash
pip install gebsecure
```

## Quick start

```python
from gebsecure import GebSecure

client = GebSecure(api_key="gsk_...")  # or set GEBSECURE_API_KEY

result = client.guard("refunds:create", "stripe:charge/ch_123", {"amount": 250, "currency": "USD"})
if result.allowed:
    issue_refund()
else:
    print("blocked:", result.decision.decision, result.decision.reason)
```

`authorize()` returns the full decision (`ALLOW`, `DENY` or `APPROVAL_REQUIRED`) with risk, explanation
and limits. Denials are results, not exceptions:

```python
decision = client.authorize(action="refunds:create", resource="stripe:charge/ch_123", context={"amount": 250})
print(decision.decision, decision.risk.score, decision.explanation.granting_policy_id)
```

### Async

```python
import asyncio
from gebsecure import AsyncGebSecure


async def main() -> None:
    async with AsyncGebSecure(api_key="gsk_...") as client:
        decision = await client.authorize(action="files:read", resource="s3://reports/q3.pdf")
        print(decision.decision)


asyncio.run(main())
```

Cancelling the awaiting task aborts the in-flight request, a pending backoff sleep or an approval poll.

## Configuration

| Argument | Default | Notes |
| --- | --- | --- |
| `api_key` / `auth` | `GEBSECURE_API_KEY` or `GEBSECURE_CLIENT_SECRET` | exactly one |
| `gateway_url` | `GEBSECURE_GATEWAY_URL` or `https://gateway.gebsecure.dev` | data plane |
| `api_url` | `GEBSECURE_API_URL` or `https://api.gebsecure.dev` | control plane + token endpoint |
| `token_url` | `{api_url}/v1/auth/token` | |
| `max_retries` | `3` | `0` disables retries (and automatic idempotency keys) |
| `timeout` | `30.0` | per attempt, seconds |
| `base_delay` / `max_delay` | `0.5` / `8.0` | backoff, seconds |
| `max_retry_after` | `60.0` | longer `Retry-After` values raise `RateLimitError` immediately |
| `http_client` / `transport` | – | custom `httpx` client (proxies, mTLS) or transport |

`client.with_options(max_retries=0, timeout=5)` returns a copy that shares the connection pool and auth.

## Authentication

```python
from gebsecure import GebSecure, ClientSecretAuth, PrivateKeyJwtAuth, DelegationAuth

# API key (gsk_...): sent directly as the Bearer token
GebSecure(api_key="gsk_...")

# Client credentials with a client secret (gsa_...)
GebSecure(auth=ClientSecretAuth("gsa_...", scope="authorize execute"))

# private_key_jwt: the agent signs a short-lived assertion; GebSecure only stores the public key.
# Accepts PKCS#8 PEM, a private JWK (dict/JSON) or a cryptography key object. Ed25519, P-256, RSA.
GebSecure(auth=PrivateKeyJwtAuth("cred_...", open("agent-key.pem").read()))

# Delegation: exchange a gsd_ delegation token, proving identity with the delegate agent's own credentials
GebSecure(auth=DelegationAuth("gsd_...", actor=PrivateKeyJwtAuth("cred_...", key_pem), scope="authorize"))
```

OAuth tokens are cached and refreshed `min(60 s, expires_in / 2)` before they expire. Concurrent callers
share a single token request. If an API call returns 401, the cached token is dropped, a new one is
fetched and the request is replayed once. Each token request uses a fresh assertion (unique `jti`).
The assertion `aud` defaults to the token URL. Pass `audience=` if the server's `PUBLIC_API_URL` differs.

## Executing actions and approvals

```python
from gebsecure import ApprovalNotGrantedError

result = client.execute(connection="stripe-prod", action="refunds.create", input={"charge": "ch_123", "amount": 250})
# result.status: succeeded | failed | denied | awaiting_approval | running | cancelled | simulated

try:
    result = client.execute_with_approval(
        connection="stripe-prod",
        action="refunds.create",
        input={"charge": "ch_123", "amount": 5000},
        timeout=900,
        poll_interval=5,
    )
except ApprovalNotGrantedError as e:
    print("approval", e.approval.status)
```

`execute_with_approval` runs the action. If it is `awaiting_approval`, it waits for the approval
(`wait_for_approval`). The GebSecure worker then resumes the stored execution with the approval, so the
helper polls `get_execution` until the execution is terminal and returns it; it never executes a second
time. A rejected, expired or cancelled approval raises `ApprovalNotGrantedError`. Other helpers: `get_execution`, `cancel_execution`, `get_approval`,
`wait_for_approval`, `authorize_batch`.

Control plane: `list_agents`, `get_agent`, `list_policies`, `list_approvals`, `create_delegation`.

`client.with_delegation("gsd_...", scope="authorize")` returns a client that acts under a delegation token,
using this client's credentials as the actor. `client.get_access_token()` returns the current (cached or
refreshed) access token for APIs the SDK does not wrap.

## Inspecting untrusted content, DLP and agent spans

```python
import time
from gebsecure import build_span, traceparent_of

# Before the model reads a web page, email, RAG chunk or tool output:
r = client.inspect(content=page_text, source="web", origin="https://example.com/page")
if r.action == "BLOCK":
    raise RuntimeError(f"prompt injection ({r.injection.level}, score {r.injection.score})")
safe_text = r.content  # sanitized / masked; hand this to the model instead of page_text

# Outbound DLP (dry_run=True evaluates without recording a DLP event):
scan = client.dlp_scan(content={"reply": draft}, direction="outbound", action="zendesk:tickets.reply")

# Report your own spans (LLM calls, planning, tool selection) into the shared trace:
t0 = time.time() * 1000
plan = llm.plan(safe_text)
span = build_span("llm.plan", t0, time.time() * 1000, traceparent=incoming_traceparent, attributes={"model": "x"})
client.report_spans([span])
client.authorize(action="stripe:refunds.create", resource="stripe:charge/ch_1", traceparent=traceparent_of(span))
```

`build_span` takes the trace id of `traceparent` (and its span id as the parent) or starts a new trace;
`traceparent_of(span)` makes later SDK calls children of the span, so the platform's authorization and
execution spans appear under it in `get_trace(trace_id)`.

## Simulation

```python
from gebsecure import SimulationFailedError

try:
    run = client.run_simulation(
        environment_id="env_...",
        fail_on_mismatch=True,  # CI: a failed run is raised instead of returned
        scenarios=[
            {
                "label": "support refunds a small charge",
                "principal": {"synthetic": {"type": "agent", "tags": ["support"]}},
                "request": {
                    "action": "stripe:refunds.create",
                    "resource": "stripe:charge/ch_1",
                    "context": {"amount": 50},
                },
                "expect": {"decision": "ALLOW"},
            }
        ],
    )
except SimulationFailedError as e:
    for r in e.run.results:
        print(r.label, r.failures)
```

## User sessions (CLIs)

```python
from gebsecure import GebSecure, StaticTokenAuth, Unauthenticated

session = GebSecure(auth=Unauthenticated()).login(email="me@example.com", password=password)
store(session.refresh_token)  # rotates: always keep the newest one
admin = GebSecure(auth=StaticTokenAuth(session.access_token))
print(admin.me().permissions)
tokens = GebSecure(auth=Unauthenticated()).refresh_session(load_refresh_token())
admin.logout()
```

`login` always asks for the refresh token in the body (`refreshTokenDelivery=body`); `refresh_session` is never
retried (except on 429) because a replayed refresh token revokes the whole session family. An
`Unauthenticated` client raises `ConfigurationError` for every other call, before sending anything.

## Management

Every method accepts a request model or a mapping (wire or Python names) and/or keyword fields, plus the
per-call options `traceparent=` and `idempotency_key=`.

| Area | Methods |
| --- | --- |
| Policies | `get_policy`, `create_policy`, `update_policy` (PUT, new version), `delete_policy`, `validate_policy`, `get_policy_version`, `diff_policy_versions`, `list_policy_deployments`, `deploy_policy`, `rollback_policy`, `add_policy_test`, `replace_policy_tests` (PUT), `delete_policy_test`, `run_policy_tests`, `simulate_policies` |
| Agents | `create_agent`, `update_agent`, `set_agent_status`, `delete_agent`, `list_agent_credentials`, `create_agent_credential`, `rotate_credential`, `revoke_credential` |
| Secrets | `create_secret`, `list_secrets`, `get_secret`, `update_secret`, `rotate_secret`, `revoke_secret`, `delete_secret` |
| Connectors | `list_connectors`, `get_connector`, `list_connections`, `get_connection`, `create_connection`, `update_connection`, `delete_connection`, `check_connection_health`, `start_connection_oauth`, `complete_connection_oauth`, `install_client_credentials`, `install_jwt_bearer` |
| Audit | `search_audit` (`after_seq=` tails new events, oldest first), `get_audit_event`, `export_audit` (raw text), `verify_audit` |
| Approvals | `get_approval_details`, `decide_approval` (with the reviewed `snapshot_hash`) |
| Alerts | `list_alerts`, `get_alert`, `update_alert` |
| Traces | `search_traces`, `get_trace`, `replay_session` |

`deploy_policy` returns the deploy report with `deployed=False` when tests or conflicts block it (HTTP 412)
instead of raising. Secret values are write-only: the API never returns them and the SDK never logs them
or puts request bodies in error messages. Deletes return `None`.

## Errors

All errors derive from `GebSecureError` and expose `status`, `code`, `detail`, `request_id`, `trace_id`,
`details`:

| Exception | Codes |
| --- | --- |
| `AuthenticationError` | `unauthenticated` (401) |
| `PermissionDeniedError` | `forbidden` (403) |
| `PolicyDeniedError` | `policy_denied` |
| `ApprovalRequiredError` | `approval_required` |
| `ValidationError` | `validation_failed`, `bad_request` |
| `RateLimitError` | `rate_limited`, `quota_exceeded` (`retry_after` seconds) |
| `NotFoundError` | `not_found` |
| `ConflictError` | `conflict`, `precondition_failed` |
| `UpstreamError` | `upstream_error` (502) |
| `GatewayTimeoutError` | `timeout` (504 or client-side timeout) |
| `UnavailableError` | `unavailable` (503) |
| `InternalError` | `internal`, unknown 5xx |
| `APIError` | base of the above, and any unknown code |
| `APIConnectionError` | network failures (status 0) |
| `ApprovalTimeoutError` / `ApprovalNotGrantedError` | approval helpers |
| `SimulationFailedError` | `simulation_failed` (422 with `fail_on_mismatch`; carries `run`) |
| `ConfigurationError` | `invalid_configuration` (e.g. an authenticated call on an `Unauthenticated` client) |

## Retries

429 responses are always retried, honouring `Retry-After`. Network errors, timeouts and 502/503/504 are
retried only for idempotent requests: GETs, PUTs and DELETEs (except `update_policy`, which appends a
version), token requests, `dry_run` authorizations and DLP scans, span reports, `persist=False`
simulations, evaluation-only calls, cancellations, executions that carry an idempotency key, and any call
made with `idempotency_key=` (sent as the `Idempotency-Key` header on every attempt). Other POSTs and
PATCHes are not retried. When retries are enabled, `execute` generates an
`idempotencyKey` (`sdk-<uuid>`) and reuses it for every attempt. Backoff is exponential
(`0.5 s × 2ⁿ`, capped at 8 s) with equal jitter. A 502/504 whose body is an `ExecuteResult` is a final
`failed` result and is not retried.

## Tracing

Every call sends a W3C `traceparent`. Pass `traceparent=` on a call to continue your own trace.
`request_id` and `trace_id` on errors identify the request in GebSecure logs and traces.

## Examples

See [`examples/`](examples): `authorize_then_act.py`, `execute_with_approval.py`, `private_key_jwt.py`,
`delegation_exchange.py`, `inspect_and_trace.py`.

## Development

```bash
cd sdks/python
python -m venv .venv && . .venv/bin/activate   # Windows: .venv\Scripts\activate
pip install -e ".[dev]"
pytest                      # unit + cross-SDK conformance (sdks/conformance/fixtures)
mypy --strict src/gebsecure
ruff check . && ruff format --check .
GEBSECURE_LIVE_TEST=1 GEBSECURE_API_KEY=gsk_... GEBSECURE_GATEWAY_URL=http://localhost:4001 pytest tests/test_live.py
```

Models in `src/gebsecure/_generated_models.py` are generated. Regenerate them with
`node sdks/scripts/generate-models.mjs` from the repository root and do not edit that file by hand.
`_async.py` mirrors `_sync.py`, so keep the two in step.

See [`sdks/VERSIONING.md`](../VERSIONING.md) for versioning and compatibility.
