gate-sdk 0.3.1

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.net",
    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.net",
    tenant_id="your-tenant-id",
    auth=ApiKeyAuth(
        mode="apiKey",
        api_key="your-api-key",
    ),
))

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

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.net",  # 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

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. Automatic Token Acquisition: The SDK automatically starts a background heartbeat refresher when the GateClient is initialized
2. Token Refresh: Heartbeat tokens are refreshed every 10 seconds (configurable) to maintain a valid token
3. Signing Enforcement: Before any evaluate() call, the SDK checks for a valid heartbeat token. If missing or expired, it throws HEARTBEAT_MISSING error
4. Token Inclusion: The heartbeat token is automatically included in the signingContext of every evaluation request

Configuration

The heartbeat manager is automatically configured based on your GateClientConfig:

gate = GateClient(GateClientConfig(
    base_url="https://gate.blockintelai.net",  # 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.net",  # Optional
    default_signer_id="my-signer-id",  # Optional: default signerId for heartbeat
))

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, GateErrorCode

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

Heartbeat Manager API

The heartbeat manager is internal to the SDK, but you can access it if needed:

# Check if heartbeat is valid
is_valid = gate._heartbeat_manager.is_valid()

# Get current heartbeat token (if valid)
token = gate._heartbeat_manager.get_token()

# Update signer ID (called automatically when signer is known)
gate._heartbeat_manager.update_signer_id("new-signer-id")

# Stop heartbeat refresher (e.g., on shutdown)
gate._heartbeat_manager.stop()

Note: The heartbeat manager automatically updates the signerId when using the KMS wrapper, so manual updates are typically not needed.

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.blockintel.ai
• Issues: https://github.com/4KInc/blockintel-ai/issues
• Email: [e-mail]