Python SDK guide

Use the Python SDK when your customer integration lives in a backend service, worker, notebook, or Python agent framework. This page keeps one consistent example thread: payment creation and refund handling.

Install

python3 -m pip install ledgix-python

Optional extras:

pip install "ledgix-python[langchain]"
pip install "ledgix-python[llamaindex]"
pip install "ledgix-python[crewai]"
pip install "ledgix-python[yaml]"    # required for ledgix.yaml manifests

Core configuration

The Python client reads these environment variables automatically:

Field	Type	Required	Description
LEDGIX_VAULT_URL	string	Yes	Vault base URL. Default http://localhost:8000.
LEDGIX_VAULT_API_KEY	string	Yes	Tenant API key sent as X-Vault-API-Key.
LEDGIX_VAULT_TIMEOUT	float	No	HTTP timeout in seconds. Default 30.0.
LEDGIX_AGENT_ID	string	No	Default agent identity. Default default-agent.
LEDGIX_SESSION_ID	string	No	Default session grouping id.
LEDGIX_VERIFY_JWT	bool	No	Validate A-JWTs against JWKS. Default true.
LEDGIX_JWT_ISSUER	string	No	Expected iss claim. Default alcv-vault.
LEDGIX_JWT_AUDIENCE	string	No	Expected aud claim. Default ledgix-sdk.
LEDGIX_REVIEW_POLL_INTERVAL	float	No	Polling interval for processing/pending_review. Default 2.0s.
LEDGIX_REVIEW_TIMEOUT	float	No	Maximum wait before ManualReviewTimeoutError. Default 300s.
LEDGIX_MAX_RETRIES	int	No	Transport retry attempts. Default 3.
LEDGIX_RETRY_BASE_DELAY	float	No	Base delay for retry jitter, seconds. Default 0.5.

Quick start

The recommended pattern is manifest-driven: declare enforcement once, instrument at startup, and keep tool signatures unchanged.

import ledgix_python as ledgix
import tools
 
# ledgix.yaml
# enforce:
#   - tool: "stripe_*"
#     policy_id: "financial-high-risk"
#   - tool: "*"
#     policy_id: "default"
 
# 1. Configure once at startup — reads LEDGIX_* env vars for most settings
ledgix.configure(agent_id="payments-agent")
 
# 2. Auto-instrument matching functions in the module
ledgix.auto_instrument(tools)

Inside any instrumented function you can still access the A-JWT with current_token():

def stripe_charge(amount: float, customer_id: str) -> dict:
    token = ledgix.current_token()
    return stripe.PaymentIntent.create(
        amount=int(amount * 100),
        customer=customer_id,
        metadata={"vault_token": token},
    )

configure() accepts keyword arguments that map to VaultConfig fields, or you can pass a pre-built VaultConfig instance. Environment variables are applied as defaults.

current_token() returns the A-JWT token for the active call context — None if called outside an instrumented or @enforce-decorated function. Use current_clearance() if you need the full ClearanceResponse object.

Manifest format

The Python SDK auto-discovers ledgix.yaml, ledgix.yml, then ledgix.json from the current working directory:

enforce:
  - tool: "stripe_*"
    policy_id: "financial-high-risk"
  - tool: "db_write*"
    policy_id: "data-mutation"
  - tool: "*"
    policy_id: "default"

Rules are evaluated in order. The first matching glob wins.

You can also pass a path or inline manifest:

ledgix.auto_instrument(tools, manifest="config/ledgix.yaml")
ledgix.auto_instrument(
    tools,
    manifest={"enforce": [{"tool": "stripe_*", "policy_id": "financial-high-risk"}]},
)

YAML manifests require pyyaml.

Escape hatch: `@ledgix.tool`

For functions outside the scanned module, use @ledgix.tool:

@ledgix.tool
def special_refund(amount: float):
    return ledgix.current_token()
 
@ledgix.tool(policy_id="override-policy")
def stripe_charge(amount: float):
    return ledgix.current_token()

If a manifest has already been loaded, @ledgix.tool uses the matching rule first and then applies any explicit overrides.

Advanced: explicit client API

For cases that need per-call client control or direct clearance requests:

from ledgix_python import LedgixClient, VaultConfig
 
client = LedgixClient(
    config=VaultConfig(
        vault_url="https://vault.example.com",
        vault_api_key="sk_prod_example",
        agent_id="payments-agent",
        session_id="checkout-42",
    )
)

Async variant:

import asyncio
from ledgix_python import LedgixClient, ClearanceRequest
 
async def main() -> None:
    client = LedgixClient()
    result = await client.arequest_clearance(
        ClearanceRequest(
            tool_name="create_stripe_payment",
            tool_args={
                "amount": 249.99,
                "currency": "USD",
                "customer_id": "cus_123",
                "payment_method_id": "pm_123",
                "order_event_id": "ord_evt_123",
                "reasoning": "Charge matches a valid completed order event.",
            },
            agent_id="payments-agent",
            session_id="checkout-42",
            context={"policy_id": "stripe-agent-prod"},
        )
    )
    print(result.status, result.request_id)
 
asyncio.run(main())

The @ledgix.enforce(...) and @vault_enforce(client, ...) decorators remain available for advanced or compatibility cases. @vault_enforce injects a _clearance keyword argument instead of using a context variable.

Built-in status handling

The Python client mirrors the TypeScript client behavior:

Use vault_enforce when you want Ledgix to inject the approval token into the function that executes the protected action.

from ledgix_python import LedgixClient, vault_enforce
 
client = LedgixClient()
 
@vault_enforce(client, tool_name="stripe_refund", policy_id="payments-prod")
def process_refund(amount: int, reason: str, order_event_id: str, **kwargs):
    clearance = kwargs["_clearance"]
    return stripe.Refund.create(
        amount=amount,
        reason=reason,
        metadata={
            "order_event_id": order_event_id,
            "ledgix_request_id": clearance.request_id,
            "ledgix_token": clearance.token,
        },
    )

LangChain example

Use the LangChain wrapper when you already have a tool object and want it to fail closed before execution.

from langchain_core.tools import StructuredTool
from ledgix_python.adapters.langchain import LedgixTool
 
refund_tool = StructuredTool.from_function(
    func=refund_customer,
    name="stripe_refund",
    description="Refund a customer payment",
)
 
guarded_tool = LedgixTool.wrap(
    client,
    refund_tool,
    policy_id="payments-prod",
)

LlamaIndex example

from llama_index.core.tools import FunctionTool
from ledgix_python.adapters.llamaindex import wrap_tool
 
refund_tool = FunctionTool.from_defaults(
    fn=refund_customer,
    name="stripe_refund",
    description="Refund a customer payment",
)
 
guarded_tool = wrap_tool(
    client,
    refund_tool,
    policy_id="payments-prod",
)

CrewAI example

from crewai.tools import BaseTool
from ledgix_python.adapters.crewai import LedgixCrewAITool
 
class StripeRefundTool(BaseTool):
    name = "stripe_refund"
    description = "Refund a customer payment"
 
    def _run(self, amount: int, reason: str, order_event_id: str):
        return refund_customer(
            amount=amount,
            reason=reason,
            order_event_id=order_event_id,
        )
 
guarded_tool = LedgixCrewAITool.wrap(
    client,
    StripeRefundTool(),
    policy_id="payments-prod",
)

Direct `request_clearance` example

Recommended: configure once, decorate each tool

import ledgix_python as ledgix
from ledgix_python.adapters.langchain import LedgixTool
from ledgix_python.adapters.llamaindex import wrap_tool
from ledgix_python.adapters.crewai import LedgixCrewAITool
 
# One call at startup
ledgix.configure(agent_id="payments-agent")
 
# LangChain — no client arg needed
guarded = LedgixTool.wrap(stripe_refund_tool, policy_id="stripe-agent-prod")
 
# LlamaIndex — no client arg needed
guarded = wrap_tool(stripe_fn_tool, policy_id="stripe-agent-prod")
 
# CrewAI — no client arg needed
guarded = LedgixCrewAITool.wrap(StripeRefundTool(), policy_id="stripe-agent-prod")

All adapter wrappers fall back to the global client when no explicit client is passed.

Advanced: explicit wrapper pattern

from ledgix_python import ClearanceRequest, LedgixClient
 
client = LedgixClient()
 
clearance = client.request_clearance(
    ClearanceRequest(
        tool_name="create_stripe_payment",
        tool_args={
            "amount": 249.99,
            "currency": "USD",
            "customer_id": "cus_123",
            "payment_method_id": "pm_123",
            "order_event_id": "ord_evt_2048",
            "reasoning": "Charge matches a completed order event.",
        },
        agent_id="payments-agent",
        session_id="checkout-42",
        context={"policy_id": "payments-prod"},
    )
)
 
if clearance.status == "pending_review":
    print("Waiting for reviewer", clearance.request_id)
elif clearance.approved and clearance.token:
    print("Approved token", clearance.token)
else:
    print("Blocked", clearance.reason)

Async example

result = await client.arequest_clearance(
    ClearanceRequest(
        tool_name="create_stripe_payment",
        tool_args={"amount": 249.99, "currency": "USD"},
        agent_id="payments-agent",
        session_id="checkout-42",
        context={"policy_id": "payments-prod"},
    )
)

Verification helpers

The Python client also supports:

fetch_jwks() and verify_token(token) for token verification
fetch_ledger(limit) and fetch_ledger_checkpoints(limit) for audit history
fetch_ledger_inclusion_proof(request_id) for a single-entry inclusion proof
fetch_ledger_consistency_proof(from_, to) for a consistency proof between checkpoints
fetch_ledger_proof_bundle(request_id) and verify_ledger_proof_bundle(bundle) for combined proof verification

Exceptions

All SDK errors inherit from LedgixError:

ClearanceDeniedError — Ledgix denied the request. Do not execute the action.
ManualReviewTimeoutError — polling hit review_timeout before a reviewer acted.
VaultConnectionError — network or transport failure reaching Vault.
TokenVerificationError — A-JWT signature, issuer, audience, or expiry check failed.
PolicyRegistrationError — register_policy(...) was rejected.

from ledgix_python import ClearanceDeniedError, ManualReviewTimeoutError
 
try:
    process_refund(4500, "duplicate charge", "ord_evt_2048")
except ClearanceDeniedError as err:
    # reviewer rejected or policy denied
    log.warning("blocked: %s", err.reason)
except ManualReviewTimeoutError as err:
    # escalate or put on a retry queue
    log.info("still pending: %s", err.request_id)

Context manager

Use VaultContext to set per-call overrides (agent id, session id, policy hints) inside a scope without rebuilding the client:

from ledgix_python import VaultContext
 
with VaultContext(agent_id="refund-worker", session_id=ticket_id):
    process_refund(4500, "duplicate charge", "ord_evt_2048")

Common Python mistakes

Wrapping helper functions instead of the real side effect.
Sending vague tool arguments that do not match the action the reviewer will later inspect.
Forgetting that pending_review is a customer workflow state, not a network failure.
Testing only the happy path and never confirming how your service behaves when Ledgix blocks or pauses the action.