gate-sdk 0.4.0

Production-grade Python SDK for BlockIntel Gate Hot Path API

BlockIntel Gate Python SDK

Production-grade Python SDK for BlockIntel Gate Hot Path API.

Installation

pip install gate-sdk

Or from source:

cd sdk/python
pip install -e ".[dev]"

Requirements

• Python >= 3.9
• httpx >= 0.25.0

Quick Start

HMAC Authentication (Recommended for Production)

from gate_sdk import GateClient, GateClientConfig, HmacAuth

# Initialize client with HMAC auth
gate = GateClient(GateClientConfig(
    base_url="https://gate.blockintelai.com",
    tenant_id="your-tenant-id",
    auth=HmacAuth(
        mode="hmac",
        key_id="your-key-id",
        secret="your-hmac-secret",
    ),
    enable_stepup=True,
))

# Evaluate a transaction
response = gate.evaluate({
    "txIntent": {
        "from": "0x1234567890123456789012345678901234567890",
        "to": "0x0987654321098765432109876543210987654321",
        "value": "1000000000000000000",  # 1 ETH in wei
        "data": "0x...",
        "nonce": 42,
        "gasPrice": "20000000000",
        "gasLimit": "21000",
        "chainId": 1,
    },
    "signingContext": {
        "signerId": "my-signer-id",
        "source": {
            "repo": "myorg/myrepo",
            "workflow": "deploy-production",
            "environment": "production",
        },
        "wallet": {
            "address": "0x1234...",
            "type": "hardware",
        },
    },
})

if response["decision"] == "ALLOW":
    # Proceed with transaction
    print(f"Transaction approved: {response.get('correlationId')}")
elif response["decision"] == "REQUIRE_STEP_UP":
    # Poll for step-up decision
    final = gate.await_stepup_decision(
        request_id=response["stepUp"]["requestId"]
    )

    if final["status"] == "APPROVED":
        # Proceed with transaction
        print(f"Step-up approved: {final.get('correlationId')}")
    else:
        # Block transaction
        print(f"Step-up denied or expired: {final['status']}")
else:
    # BLOCK
    print(f"Transaction blocked: {response['reasonCodes']}")

API Key Authentication

from gate_sdk import GateClient, GateClientConfig, ApiKeyAuth

# Initialize client with API key
gate = GateClient(GateClientConfig(
    base_url="https://gate.blockintelai.com",
    tenant_id="your-tenant-id",
    auth=ApiKeyAuth(
        mode="apiKey",
        api_key="your-api-key",
    ),
))

response = gate.evaluate({
    "txIntent": {
        "from": "0x123...",
        "to": "0x456...",
        "value": "1000000000000000000",
    },
})

Local Development with Gate Local

For local development and testing, use Gate Local - a Docker container that emulates the Gate Hot Path:

# Start Gate Local
docker pull blockintelai/gate-local:latest
docker run -d --name gate-local -p 3000:3000 blockintelai/gate-local:latest

Then configure your client for local mode:

from gate_sdk import GateClient, GateClientConfig, ApiKeyAuth

# Local development configuration
gate = GateClient(GateClientConfig(
    base_url="http://localhost:3000",  # Gate Local endpoint
    tenant_id="local-dev",              # Any tenant ID (ignored in local mode)
    local=True,                          # Enable local mode (disables auth/heartbeat)
    auth=ApiKeyAuth(
        mode="apiKey",
        api_key="local-dev-key"          # Any API key (ignored in local mode)
    ),
))

# Evaluate transactions locally
response = gate.evaluate({
    "txIntent": {
        "toAddress": "0x742d35Cc6634C0532925a3b844Bc9e7595f0bEb",
        "value": "1000000000000000000",  # 1 ETH in wei
        "valueUsd": 2500.0,
        "chainId": 1,
    },
})

print(f"Decision: {response['decision']}")

📚 Full Local Development Guide: See Gate Local Quick Start Guide for complete setup instructions, trading bot integration examples, and troubleshooting.

Step-Up Polling

# Manual polling
status = gate.get_stepup_status(request_id="stepup-request-id")
print(f"Status: {status['status']}")  # PENDING | APPROVED | DENIED | EXPIRED

# Automatic polling with timeout
result = gate.await_stepup_decision(
    request_id="stepup-request-id",
    max_wait_ms=15000,  # 15 seconds
    interval_ms=250,  # Poll every 250ms
)

print(f"Final status: {result['status']}")
print(f"Elapsed time: {result['elapsedMs']} ms")

Polling behavior:

• 404 NOT_FOUND → request ID does not exist OR does not belong to the tenant
• EXPIRED → TTL exceeded (deterministic), even if DynamoDB TTL has not deleted the item yet
• PENDING → waiting for external approval
• APPROVED | DENIED → terminal states

Configuration

GateClientConfig

from dataclasses import dataclass
from gate_sdk import GateClientConfig, HmacAuth, StepUpConfig

config = GateClientConfig(
    base_url="https://gate.blockintelai.com",  # Gate Hot Path API base URL
    tenant_id="your-tenant-id",                # Your tenant ID
    auth=HmacAuth(                              # Authentication
        mode="hmac",
        key_id="your-key-id",
        secret="your-secret",
    ),
    timeout_ms=15000,                          # Request timeout (default: 15000ms)
    user_agent="my-app/1.0",                   # User agent (default: blockintel-gate-sdk/0.1.0)
    clock_skew_ms=120000,                      # Clock skew tolerance (default: 120000ms)
    enable_stepup=False,                       # Enable step-up support (default: False)
    stepup=StepUpConfig(                       # Step-up configuration
        polling_interval_ms=250,               # Polling interval (default: 250ms)
        max_wait_ms=15000,                     # Max wait time (default: 15000ms)
        treat_require_stepup_as_block_when_disabled=True,  # Transform REQUIRE_STEP_UP to BLOCK (default: True)
    ),
)

When step-up is disabled, the SDK treats REQUIRE_STEP_UP as BLOCK by default to preserve Gate-only safety, unless the caller explicitly overrides this behavior.

gate = GateClient(config)

## API Reference

Responses are returned in the same JSON shape as the Gate Hot Path API (no automatic key renaming).

### `GateClient.evaluate(req, request_id=None)`

Evaluate a transaction defense request.

**Parameters:**
- `req: Dict[str, Any]` - Request dictionary with `txIntent` and optional `signingContext`
- `request_id: Optional[str]` - Optional request ID (auto-generated if not provided)

**Returns:** `DefenseEvaluateResponseV2`

**Response:**
```python
{
    "decision": "ALLOW" | "BLOCK" | "REQUIRE_STEP_UP",
    "reasonCodes": List[str],
    "policyVersion": Optional[str],
    "correlationId": Optional[str],
    "stepUp": Optional[{
        "requestId": str,
        "ttlSeconds": Optional[int],
    }],
}

GateClient.get_stepup_status(request_id, tenant_id=None)

Get current step-up status.

Parameters:

• request_id: str - Step-up request ID
• tenant_id: Optional[str] - Optional tenant ID (default: from config)

Returns: StepUpStatusResponse

Status Types:

• PENDING - Waiting for decision
• APPROVED - Step-up approved
• DENIED - Step-up denied
• EXPIRED - Step-up expired (TTL exceeded)

Polling behavior:

• Returns 404 NOT_FOUND if request ID does not exist OR does not belong to the tenant
• Returns EXPIRED deterministically if TTL exceeded, even if DynamoDB TTL has not deleted the item yet

GateClient.await_stepup_decision(request_id, max_wait_ms=None, interval_ms=None)

Poll step-up status until decision is reached or timeout.

Parameters:

• request_id: str - Step-up request ID
• max_wait_ms: Optional[int] - Maximum wait time in milliseconds
• interval_ms: Optional[int] - Polling interval in milliseconds

Returns: StepUpFinalResult

Error Handling

The SDK provides custom exception types:

from gate_sdk import (
    GateError,
    GateNetworkError,
    GateTimeoutError,
    GateNotFoundError,
    GateAuthError,
    GateRateLimitError,
    StepUpNotConfiguredError,
)

try:
    response = gate.evaluate({...})
except GateAuthError as e:
    print(f"Auth failed: {e}")
    print(f"Status: {e.status_code}")
    print(f"Request ID: {e.request_id}")
except GateRateLimitError as e:
    print(f"Rate limited: {e.retry_after}")
except StepUpNotConfiguredError as e:
    print(f"Step-up not configured: {e}")
except GateError as e:
    print(f"Error: {e.code} - {e.message}")

Error Codes:

• NETWORK_ERROR - Network connection failed
• TIMEOUT - Request timeout
• NOT_FOUND - Resource not found (404)
• UNAUTHORIZED - Authentication failed (401)
• FORBIDDEN - Access denied (403)
• RATE_LIMITED - Rate limit exceeded (429)
• SERVER_ERROR - Server error (5xx)
• INVALID_RESPONSE - Invalid response format
• STEP_UP_NOT_CONFIGURED - Step-up required but not enabled
• STEP_UP_TIMEOUT - Step-up polling timeout
• HEARTBEAT_MISSING - Heartbeat token is missing or expired
• HEARTBEAT_EXPIRED - Heartbeat token has expired
• HEARTBEAT_INVALID - Heartbeat token is invalid
• HEARTBEAT_MISMATCH - Heartbeat token does not match expected parameters

Authentication

HMAC v1 Signing

The SDK implements HMAC v1 signing for secure authentication:

Signing String:

v1\n
<HTTP_METHOD>\n
<PATH>\n
<TENANT_ID>\n
<KEY_ID>\n
<TIMESTAMP_MS>\n
<REQUEST_ID>\n
<SHA256_HEX_OF_BODY>\n

Signature:

HMAC-SHA256(secret, signingString) as hex

Headers:

• X-GATE-TENANT-ID
• X-GATE-KEY-ID
• X-GATE-TIMESTAMP-MS
• X-GATE-REQUEST-ID
• X-GATE-SIGNATURE

API Key

For simpler onboarding, use API key authentication:

Headers:

• X-API-KEY
• X-GATE-TENANT-ID
• X-GATE-REQUEST-ID
• X-GATE-TIMESTAMP-MS

Step-Up Flow

Step-up is a feature-flagged capability that allows Gate to defer decisions to an external approval system.

Flow:

1. SDK calls evaluate() → Gate returns REQUIRE_STEP_UP
2. SDK polls /defense/stepup/status until decision is reached
3. External system (Control Plane) approves/denies via separate API
4. SDK receives final decision: APPROVED, DENIED, or EXPIRED

Important:

• Hot Path never approves/denies step-up
• Approve/deny happens only on Control Plane
• SDK only polls status from Hot Path
• The SDK never performs approve/deny actions. Step-up resolution is handled exclusively by the Control Plane.

Gate-only deployments should leave step-up disabled; the SDK will never "wait" unless step-up is enabled.

TTL Guardrails:

• Default: 600 seconds
• Min: 300 seconds
• Max: 900 seconds

Retry Logic

The SDK automatically retries failed requests:

• Max Attempts: 3
• Retry On: Network errors, timeouts, 429, 5xx
• Never Retry On: 4xx (except 429)
• Backoff: Exponential with jitter (100ms base, 2x factor, 800ms max)

Request ID Stability:

• Same request_id is used across all retries
• Ensures idempotency on Gate server

Degraded Mode / X-BlockIntel-Degraded

When the SDK is in a degraded situation, it logs X-BlockIntel-Degraded: true with a reason for logs and telemetry only. This is never sent as an HTTP request header to the Gate server.

Reasons: retry, 429, fail_open, fail_safe_allow.

Example (one line):
[GATE SDK] X-BlockIntel-Degraded: true (reason=retry) attempt=1/3 status=503 exc=GateServerError requestId=abc-123

How to observe:

• Logs: [GATE SDK] X-BlockIntel-Degraded: true (reason: <reason>) at WARNING level. Ensure logging captures gate_sdk (e.g. logging.getLogger('gate_sdk') or root).
• Metrics: Use on_metrics; metrics include timeouts, errors, failOpen etc. Correlate with log lines if you ship both.

Manual check (retry): Point the SDK at an endpoint that returns 5xx; confirm one degraded log per retry attempt including attempt, max, and status/exc.

Heartbeat System

The SDK includes a Heartbeat Manager that automatically acquires and refreshes heartbeat tokens from the Gate Control Plane. Heartbeat tokens are required for all signing operations and ensure that Gate is alive and enforcing policy.

How It Works

1. Per-Signer Token Cache: The heartbeat manager maintains a Dict[str, SignerHeartbeatEntry] so each signer gets its own cached token, refresh timer, and backoff state. Switching between signers never invalidates other signers' tokens.
2. Automatic Token Acquisition: When get_token_for_signer(signer_id) is called, the manager returns the cached token immediately if valid, or acquires a new one synchronously. The initial default signer is pre-warmed on start().
3. Per-Signer Refresh: Each signer entry has its own daemon threading.Timer that refreshes the token before expiry (default every 10 seconds + jitter + backoff on failure).
4. LRU Eviction: When the number of concurrent signer entries exceeds max_signers (default: 20), the least-recently-used entry is evicted.
5. Idle TTL Eviction: A background timer (every 60s) evicts signer entries not used within signer_idle_ttl_seconds (default: 5 minutes).
6. Local Rate Limiting: Per-signer guard prevents re-requesting within local_rate_limit_seconds (default: 2.1s) to avoid hammering the Control Plane.
7. Token Inclusion: The heartbeat token is automatically included in the signingContext of every evaluation request.

Multi-Signer Support

Trading desks running 3+ bot profiles no longer experience HEARTBEAT_MISSING errors when switching signers. Each signer's token is cached independently:

# KMS wrapper automatically uses the correct signer from KeyId
protected_kms = wrap_kms_client(kms, gate)

# These calls use independent heartbeat tokens — no cross-invalidation
protected_kms.sign(KeyId="alias/bot-1", Message=tx1, ...)
protected_kms.sign(KeyId="alias/bot-2", Message=tx2, ...)
protected_kms.sign(KeyId="alias/bot-1", Message=tx3, ...)  # cache hit

Configuration

The heartbeat manager is automatically configured based on your GateClientConfig:

gate = GateClient(GateClientConfig(
    base_url="https://gate.blockintelai.com",  # Hot Path URL
    tenant_id="your-tenant-id",
    auth=HmacAuth(...),
    # Heartbeat manager uses base_url to infer Control Plane URL
    # Or explicitly set control_plane_url if different
    control_plane_url="https://control-plane.blockintelai.com",  # Optional
    signer_id="my-signer-id",  # Optional: default signerId for heartbeat (if known upfront)
    heartbeat_refresh_interval_seconds=10,  # Optional: heartbeat refresh interval (default: 10s)
))

The HeartbeatManager constructor accepts additional tuning parameters:

Parameter	Default	Description
max_signers	20	Max concurrent signer entries before LRU eviction
signer_idle_ttl_seconds	300 (5 min)	Evict signers not used within this window
local_rate_limit_seconds	2.1	Min seconds between acquire attempts per signer

Heartbeat Token Properties

• TTL: 15-30 seconds (short-lived for security)
• Scope: Scoped to tenantId, signerId, environment, and policyVersion
• Validation: Hot Path validates heartbeat tokens before processing any transaction
• Enforcement: "No valid heartbeat → NO SIGNATURE" - transactions are blocked if heartbeat is missing or expired

Error Handling

from gate_sdk import GateError

try:
    response = gate.evaluate({...})
except GateError as e:
    if e.code == "HEARTBEAT_MISSING":
        print("Heartbeat token missing - Gate may be down or unreachable")

Heartbeat Manager API

The primary API is get_token_for_signer(), which handles cache lookup, on-demand acquisition, and waiting:

# Get token for a specific signer
token = gate._heartbeat_manager.get_token_for_signer("my-signer-id", max_wait_seconds=2.0)

# Stop heartbeat manager and clean up all timers (e.g., on shutdown)
gate._heartbeat_manager.stop()

Note: The KMS wrapper automatically calls get_token_for_signer() with the correct signer ID extracted from KeyId, so manual token management is typically not needed.

KMS Wrapper

The SDK provides a KMS wrapper that automatically intercepts boto3 KMS signing operations and enforces Gate policies. This allows you to protect your KMS keys without modifying your existing code.

Basic Usage

from gate_sdk import (
    GateClient,
    GateClientConfig,
    HmacAuth,
    wrap_kms_client,
    BlockIntelBlockedError,
    BlockIntelStepUpRequiredError,
)
import boto3

# Initialize Gate client
gate = GateClient(GateClientConfig(
    base_url="https://gate.blockintelai.com",
    tenant_id="your-tenant-id",
    auth=HmacAuth(
        mode="hmac",
        key_id="your-key-id",
        secret="your-hmac-secret",
    ),
))

# Initialize boto3 KMS client
kms = boto3.client('kms', region_name='us-east-1')

# Wrap KMS client with Gate protection
protected_kms = wrap_kms_client(kms, gate, {
    'mode': 'enforce',  # 'enforce' or 'dry-run'
    'extract_tx_intent': lambda **kwargs: {
        'toAddress': '0x...',  # Extract from message if possible
        'networkFamily': 'EVM',
        'chainId': 1,
    },
    'on_decision': lambda decision, details: print(f'Gate decision: {decision}', details),
})

# Use wrapped KMS client - Gate will intercept automatically
try:
    response = protected_kms.sign(
        KeyId='alias/my-key',
        Message=b'transaction-data',
        MessageType='RAW',
        SigningAlgorithm='ECDSA_SHA_256',
    )
    print('Signature:', response['Signature'])
except BlockIntelBlockedError as e:
    print(f'Transaction blocked by Gate: {e.message}')
except BlockIntelStepUpRequiredError as e:
    print(f'Step-up required: {e.request_id}')
    # Handle step-up flow

Wrapper Modes

• enforce (default): Gate policies are enforced. Transactions are blocked if Gate denies.
• dry-run: Gate evaluates transactions but always allows KMS calls. Useful for testing and monitoring.

Automatic Per-Signer Heartbeat

The KMS wrapper automatically extracts the signer ID from the KMS KeyId and calls get_token_for_signer(signer_id) to get a cached (or freshly acquired) heartbeat token for that specific signer. Switching between KMS keys does not invalidate other signers' tokens.

Advanced Configuration

from gate_sdk.kms import WrapKmsClientOptions

options = WrapKmsClientOptions(
    mode='enforce',
    extract_tx_intent=lambda **kwargs: {
        # Custom extraction logic
        'toAddress': extract_address_from_message(kwargs.get('Message')),
        'networkFamily': 'EVM',
        'chainId': 1,
    },
    on_decision=lambda decision, details: log_decision(decision, details),
)

protected_kms = wrap_kms_client(kms, gate, options)

Security

• HTTPS Required: SDK validates HTTPS in production (localhost exception)
• Secret Protection: Never logs secrets or API keys
• Clock Skew: Configurable tolerance for timestamp validation
• Replay Protection: Request ID + timestamp prevent replay attacks
• Heartbeat Enforcement: All signing operations require valid heartbeat tokens

Testing

Run the test suite:

cd sdk/python
pytest -v

Run with coverage:

pytest --cov=gate_sdk --cov-report=term-missing

Building

Build the package:

cd sdk/python
python -m build

This creates:

• dist/blockintel_gate_sdk-*.whl (wheel)
• dist/blockintel-gate-sdk-*.tar.gz (source distribution)

Publishing

• Package versions are immutable once published (PyPI does not allow overwriting a released version). Always bump the version before tagging a release.

See PUBLISHING.md for detailed publishing instructions.

Quick steps:

1. Update version in pyproject.toml
2. Create GitHub release tag
3. GitHub Actions publishes to PyPI automatically

License

MIT License - see LICENSE file.

Support

• Documentation: https://docs.blockintelai.com
• Issues: https://github.com/4KInc/blockintel-ai/issues
• Email: [e-mail]

Keywords

blockintel, gate, sdk, defense, crypto, security