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+.

pip install gebsecure

Quick start

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:

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

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

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

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

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

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)

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/: authorize_then_act.py, execute_with_approval.py, private_key_jwt.py, delegation_exchange.py, inspect_and_trace.py.

Development

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 for versioning and compatibility.