bastion-sdk

Python SDK for the Bastion trust proxy. Supports both sync and async usage.

Installation

pip install bastion-sdk

Requires Python 3.10+.

Quick Start

from bastion_sdk import BastionClient

with BastionClient("http://localhost:3000", api_key="your-project-api-key") as client:
    print(client.health())

Async

from bastion_sdk import AsyncBastionClient

async with AsyncBastionClient("http://localhost:3000", api_key="your-key") as client:
    print(await client.health())

Admin Operations

Use the admin API key (PROJECT_API_KEY) to manage agents, credentials, and policies.

Agents

# Create an agent (returns one-time agentSecret)
agent = client.create_agent("Support Bot", description="Handles refund requests")
agent_secret = agent["agentSecret"]  # save this — shown only once

# List, get, update, delete
agents = client.list_agents()
agent = client.get_agent(agent["id"])
client.update_agent(agent["id"], isActive=False)
client.delete_agent(agent["id"])  # soft-delete

Credentials

# Store a credential (encrypted at rest, raw value never returned)
credential = client.create_credential(
    name="Stripe Production",
    type="API_KEY",
    value="sk_live_...",
    agent_id=agent["id"],
)

credentials = client.list_credentials(agent_id=agent["id"])
client.revoke_credential(credential["id"])

Policies

policy = client.create_policy(
    agent_id=agent["id"],
    credential_id=credential["id"],
    allowed_actions=["charges.*"],
    denied_actions=["transfers.*"],
    constraints={
        "maxAmountPerTransaction": 1000,
        "maxDailySpend": 5000,
        "rateLimit": {"maxRequests": 100, "windowSeconds": 3600},
    },
    requires_approval_above=500,
)

# Dry-run evaluation (no side effects)
result = client.evaluate_policy(
    agent_id=agent["id"],
    credential_id=credential["id"],
    action="charges.create",
    params={"amount": 750},
)
# result["decision"] → "ESCALATE"

Agent Operations

Use an agent secret (bst_...) for proxy execution.

agent_client = BastionClient("http://localhost:3000", api_key=agent_secret)

result = agent_client.execute(
    credential_id=credential["id"],
    action="charges.create",
    target={
        "url": "https://api.stripe.com/v1/charges",
        "method": "POST",
        "body": {"amount": 5000, "currency": "usd"},
    },
    params={"amount": 50},
)

# result["upstream"]["status"] → 200
# result["upstream"]["body"] → Stripe's response
# result["meta"]["policyDecision"] → "ALLOW"

Custom Credential Injection

agent_client.execute(
    credential_id=credential["id"],
    action="test",
    target={"url": "https://api.example.com"},
    injection={"location": "header", "key": "X-Api-Key"},
)

HITL (Human-in-the-Loop)

pending = client.list_pending_requests()
client.approve_request(pending[0]["requestId"])
client.deny_request(request_id, "Too risky")

Audit

records = client.query_audit_records(
    agent["id"],
    from_="2026-01-01",
    policy_decision="DENY",
    limit=10,
)

verification = client.verify_chain(agent["id"])
# verification["valid"] → True

Error Handling

All API errors raise typed exceptions:

from bastion_sdk import BastionForbiddenError, BastionNotFoundError

try:
    agent_client.execute(...)
except BastionForbiddenError as e:
    print(f"Policy denied: {e.message}")

License

MIT