SDK Documentation

"text-purple-400">pip install astra-bastion
"text-gray-500"># Or with uv (recommended)
uv add astra-bastion

"text-purple-400">from astra_bastion "text-purple-400">import AstraBastion
# API Key authentication (recommended "text-purple-400">for services)
client = AstraBastion(
    api_key="ab_live_sk_...",
    base_url="https://api.astrabastion.com"  # optional, defaults to cloud
)
# JWT authentication ("text-purple-400">for user sessions)
client = AstraBastion(
    access_token="eyJhbGci...",
    refresh_token="eyJhbGci...",  # auto-refreshes on 401
    base_url="https://api.astrabastion.com"
)

# Synchronous usage
response = client.gateway.chat(
    provider="openai",
    model="gpt-4o",
    messages=[
        {"role": "system", "content": "You are a helpful assistant."},
        {"role": "user", "content": "Explain zero trust architecture."}
    ],
    policy_id="prod-policy-001",
    temperature=0.7,
    max_tokens=2048
)
print(response.choices[0].message.content)
print(f"Injection score: {response.security.injection_score}")
print(f"PII detected: {response.security.pii_detected}")
# Async usage
"text-purple-400">async "text-purple-400">def main():
    response = "text-purple-400">await client.gateway.async_chat(
        provider="anthropic",
        model="claude-sonnet-4-20250514",
        messages=[{"role": "user", "content": "Hello"}]
    )
    "text-purple-400">return response

# Get current trust score
score = client.trust.get_current_score()
print(f"Composite: {score.composite_score} ({score.grade})")
"text-purple-400">for pillar "text-purple-400">in score.pillars:
    print(f"  {pillar.name}: {pillar.score} (weight: {pillar.weight})")
# Get score history
history = client.trust.get_score_history(
    start_date="2026-03-01",
    end_date="2026-03-15",
    interval="daily"
)
# Configure decay model
client.trust.update_decay_config(
    model="exponential",
    half_life_days=30,
    min_score=0.0
)

# List all registered agents
agents = client.aegis.list_agents(status="active")
# Activate kill switch
client.aegis.activate_kill_switch(
    scope="agent",
    target_id="agent-uuid",
    reason="Anomalous behavior detected",
    duration_minutes=60
)
# Get agent telemetry
telemetry = client.aegis.get_agent_telemetry(
    agent_id="agent-uuid",
    metric="request_count",
    interval="1h",
    lookback="24h"
)

"text-purple-400">from astra_bastion.exceptions "text-purple-400">import (
    AstraError,
    AuthenticationError,
    RateLimitError,
    GatewayBlockedError
)
"text-purple-400">try:
    response = client.gateway.chat(
        provider="openai",
        model="gpt-4o",
        messages=[{"role": "user", "content": prompt}]
    )
"text-purple-400">except GatewayBlockedError "text-purple-400">as e:
    print(f"Blocked: {e.error_code} - {e.detail}")
    print(f"Patterns matched: {e.patterns}")
"text-purple-400">except RateLimitError "text-purple-400">as e:
    print(f"Rate limited. Retry after {e.retry_after}s")
"text-purple-400">except AuthenticationError:
    print("Token expired or invalid")
"text-purple-400">except AstraError "text-purple-400">as e:
    print(f"API error: {e.status} {e.title}")

"text-purple-400">npm install @astra-bastion/sdk
"text-gray-500"># Or with pnpm
pnpm add @astra-bastion/sdk

import { AstraBastion } from '@astra-bastion/sdk'
// API Key authentication
const client = new AstraBastion({
  apiKey: 'ab_live_sk_...',
  baseUrl: 'https://api.astrabastion.com' // optional
})
// JWT authentication with auto-refresh
const client = new AstraBastion({
  accessToken: 'eyJhbGci...',
  refreshToken: 'eyJhbGci...',
  onTokenRefresh: (tokens) => {
    // Persist new tokens
    localStorage.setItem('access_token', tokens.accessToken)
  }
})

const response = await client.gateway.chat({
  provider: 'openai',
  model: 'gpt-4o',
  messages: [
    { role: 'system', content: 'You are a helpful assistant.' },
    { role: 'user', content: 'Explain prompt injection attacks.' }
  ],
  policyId: 'prod-policy-001',
  temperature: 0.7
})
console.log(response.choices[0].message.content)
console.log('Security:', response.security)
// Streaming support
const stream = await client.gateway.chatStream({
  provider: 'anthropic',
  model: 'claude-sonnet-4-20250514',
  messages: [{ role: 'user', content: 'Hello' }]
})
for await (const chunk of stream) {
  process.stdout.write(chunk.delta.content ?? '')
}

// Get current trust score
const score = await client.trust.getCurrentScore()
console.log(`Trust: ${score.compositeScore} (${score.grade})`)
// List compliance frameworks
const frameworks = await client.compliance.listFrameworks()
for (const fw of frameworks.data) {
  console.log(`${fw.name}: ${fw.controlsCompleted}/${fw.controlsTotal}`)
}
// Run risk simulation
const simulation = await client.risk.runMonteCarloSimulation({
  riskId: 'risk-uuid',
  iterations: 10000,
  confidenceLevel: 0.95
})
console.log(`VaR (95%): $${simulation.var.toLocaleString()}`)
console.log(`CVaR: $${simulation.cvar.toLocaleString()}`)

import {
  AstraError,
  AuthenticationError,
  RateLimitError,
  GatewayBlockedError
} from '@astra-bastion/sdk'
try {
  const response = await client.gateway.chat({ ... })
} catch (error) {
  if (error instanceof GatewayBlockedError) {
    console.error(`Blocked: ${error.errorCode} - ${error.detail}`)
  } else if (error instanceof RateLimitError) {
    console.error(`Rate limited. Retry after ${error.retryAfter}s`)
  } else if (error instanceof AstraError) {
    console.error(`API error: ${error.status} ${error.title}`)
  }
}