gate-carrier-sdk

Python SDK for the BlockIntel Gate Carrier API -- underwriting intelligence for insurance carriers.

Installation

pip install gate-carrier-sdk

Requires Python >= 3.9. Depends on httpx.

Quick Start

from gate_carrier_sdk import CarrierClient

with CarrierClient(
    api_key="carrier_abc123...",
    base_url="https://api.gate.blockintelai.net/api/v1/gate",
    carrier_id="carrier_xyz",
) as client:
    portfolio = client.get_portfolio()
    print(f"Tenants: {portfolio.aggregate_metrics.total_tenants}")

API Reference

Trial Management

Method	Description
create_underwriting_trial(prospect_name, contact_email, duration_days=14)	Create shadow-mode prospect tenant for underwriting
get_trial_status(trial_id)	Get detailed trial status with observation progress
list_trials(status=None, next_token=None, limit=None)	List all trials (paginated)
convert_trial(trial_id, policy_id)	Convert completed trial to insured tenant

Portfolio

Method	Description
get_portfolio()	Full portfolio dashboard -- risk summaries, metrics, alerts
get_portfolio_health()	High-level portfolio health status

Bot Blast Radius (BBR)

Method	Description
create_bbr_test(tenant_id, start_at=None, end_at=None, bots=None)	Create BBR test for a tenant (10/month quota)
get_bbr_report(tenant_id)	Get BBR report for a specific tenant
get_portfolio_summary()	BBR metrics across all authorized tenants
get_max_loss_estimate()	Portfolio max-loss at p50/p90/p99

Alerts

Method	Description
get_alerts(severity=None)	Alert history across portfolio
acknowledge_alert(alert_id)	Mark alert as reviewed

Evidence & Claims

Method	Description
generate_claim_pack(tenant_id, period_start, period_end, bbr_test_id=None)	Initiate async claim evidence pack generation
get_claim_pack_status(job_id)	Poll claim pack job status
get_evidence_bundle(tenant_id)	Raw evidence bundle (decisions, snapshots, Merkle root)

Underwriting Analytics

Method	Description
get_underwriting_report(tenant_id)	30-day underwriting metrics for a tenant
get_risk_score(tenant_id)	Composite risk score (0--100) with factor breakdown

Independent Evidence Verification

Method	Description
get_active_policy(tenant_id)	Active policy rules, enforcement level, snapshot hash
get_policy_history(tenant_id, next_token=None, limit=None)	Policy change history (paginated)
get_policy_at(tenant_id, timestamp)	Policy active at a specific point in time
get_kms_inventory(tenant_id)	KMS and IAM enforcement inventory
set_baseline_policy(baseline)	Set carrier's minimum policy baseline
get_baseline_policy()	Get current baseline policy
simulate_transaction(tenant_id, tx_intent=None, signing_context=None)	Simulate transaction against tenant's policy

Webhooks

Method	Description
register_policy_change_webhook(webhook_url, events=None)	Register webhook for POLICY_CHANGED events
list_policy_webhooks()	List all policy-change webhook subscriptions

Convenience Helpers

Method	Description
list_all_trials(status=None, limit=None)	Auto-paginating generator for trials
poll_claim_pack(job_id, poll_interval_s=3.0, timeout_s=600.0)	Poll claim pack until complete or timeout

Async Client

For asyncio-based frameworks (FastAPI, aiohttp, etc.), use AsyncCarrierClient. It provides the same API surface with async/await methods.

from gate_carrier_sdk import AsyncCarrierClient

async with AsyncCarrierClient(
    api_key="carrier_abc123...",
    base_url="https://api.gate.blockintelai.net/api/v1/gate",
    carrier_id="carrier_xyz",
) as client:
    score = await client.get_risk_score("tenant_xyz")
    print(score.composite_score, score.risk_label)

The async client supports auto-paginating async iterators and async polling:

async with AsyncCarrierClient(api_key=key, base_url=url, carrier_id=cid) as client:
    # Auto-paginate all observing trials
    async for trial in client.list_all_trials("OBSERVING"):
        print(trial.trial_id, trial.decisions_observed)

    # Poll claim pack until complete
    result = await client.poll_claim_pack(job_id, poll_interval_s=5.0)
    print(result.download_url)

Client-Side Utilities

These are pure functions -- no API calls required.

verify_policy_compliance(active_policy, baseline)

Check a tenant's active policy against the carrier's baseline specification. Returns a PolicyComplianceDiff with is_compliant, missing_rules, weaker_rules, satisfied_rules, and enforcement level comparison.

from gate_carrier_sdk import CarrierClient, verify_policy_compliance

with CarrierClient(api_key=key, base_url=url, carrier_id=cid) as client:
    active_policy = client.get_active_policy("tenant_xyz")
    baseline_resp = client.get_baseline_policy()

    diff = verify_policy_compliance(active_policy, baseline_resp.baseline)

    if not diff.is_compliant:
        print("Missing rules:", diff.missing_rules)
        print("Weaker rules:", diff.weaker_rules)

verify_webhook_signature(payload, signature, secret)

Verify HMAC-SHA256 webhook signatures on incoming payloads. Uses timing-safe comparison to prevent timing oracle attacks.

from gate_carrier_sdk import verify_webhook_signature

@app.route("/webhooks/gate", methods=["POST"])
def handle_webhook():
    signature = request.headers.get("X-Gate-Webhook-Signature", "")
    if not verify_webhook_signature(request.data, signature, WEBHOOK_SECRET):
        return "Invalid signature", 401
    # Process event...

Error Handling

All errors extend CarrierError so callers can catch either the base class or a specific subclass.

Error Class	HTTP Status	Description
CarrierError	--	Base error (all others extend this)
CarrierAuthError	401/403	Invalid or missing carrier API key
CarrierNotFoundError	404	Resource not found
CarrierConflictError	409	Conflict (e.g. duplicate trial, already converted)
CarrierQuotaExceededError	429	Monthly quota exhausted (has reset_at property)
CarrierServerError	5xx	Server-side error

from gate_carrier_sdk import (
    CarrierClient,
    CarrierAuthError,
    CarrierQuotaExceededError,
    CarrierError,
)

try:
    client.create_bbr_test("tenant_xyz")
except CarrierAuthError:
    print("Invalid API key")
except CarrierQuotaExceededError as e:
    print(f"Quota exceeded, resets at: {e.reset_at}")
except CarrierError as e:
    print(f"API error: {e} (HTTP {e.status_code})")

Environment Variables

Variable	Required	Description
GATE_BASE_URL	Yes	Gate Control Plane base URL
CARRIER_API_KEY	Yes	Carrier API key (format: carrier_<hex>)
CARRIER_ID	Yes	Carrier identifier

import os
from gate_carrier_sdk import CarrierClient

client = CarrierClient(
    api_key=os.environ["CARRIER_API_KEY"],
    base_url=os.environ["GATE_BASE_URL"],
    carrier_id=os.environ["CARRIER_ID"],
)

Configuration Options

Option	Type	Default	Description
api_key	str	--	Carrier API key (required)
base_url	str	--	Gate Control Plane base URL (required)
carrier_id	str	--	Carrier identifier (required)
timeout_s	float	30.0	Request timeout in seconds
max_retries	int	3	Max retry attempts on transient failures
user_agent_suffix	str	None	Optional suffix for the User-Agent header

Notes

• Requires Python >= 3.9.
• Depends on httpx for HTTP requests.
• All response types are dataclasses with typed fields.
• The synchronous CarrierClient is not thread-safe. Create one client per thread, or use AsyncCarrierClient for concurrent async workflows.
• All methods automatically retry on transient failures (5xx, network errors) with exponential backoff.
• URL path parameters are encoded with urllib.parse.quote to prevent injection.

License

MIT

---

Looking for TypeScript? See gate-carrier-sdk.