buddhi-sdk 0.3.0

Python SDK for Buddhi Engine — AI Guardrails, Agent Studio, Knowledge Base, and Causal Forward Models

buddhi-sdk

Python SDK for the Buddhi Engine Platform — v0.3.0

Four dedicated clients for the full Buddhi Engine stack:

Client	What It Does	Tier
RakshaClient	AI guardrail validation — HIPAA, EU AI Act, FINRA, custom policies	Starter+
KnowledgeBaseClient	Semantic search with ~25× geometric compression	Starter+
AgentClient	Agent Studio — plan generation, memory facts, tool orchestration	Professional+
CFMClient	⚗️ Causal Forward Models — state evolution & energy-based risk	Professional+

Built for the Buddhi Engine platform — AERM architecture + Guardrail DSL + Geometric Compression.

---

Installation

pip install buddhi-sdk

Requires Python 3.8+. Dependencies: httpx>=0.24.0, pydantic>=2.0.0.

Quick Start

Guardrail Validation (Raksha M)

from buddhi_sdk import RakshaClient

client = RakshaClient(api_key="be_YOUR_KEY")

result = client.validate("Can you share patient lab results?")
print(result.decision)   # "APPROVE" or "BLOCK"
print(result.is_blocked)  # True / False

Knowledge Base (Buddhi KB)

from buddhi_sdk import KnowledgeBaseClient

kb = KnowledgeBaseClient(api_key="be_YOUR_KEY")

# Ingest a document with ~25× compression
result = kb.ingest("Company HR policy text...", doc_id="hr-policy-v2", domain="hr")
print(f"Compressed {result.compression_ratio}x")

# Semantic search (<100ms)
matches = kb.query("vacation policy for remote workers", limit=5)
for m in matches.results:
    print(f"{m.doc_id}: {m.score:.3f}")

Agent Studio (Professional+)

from buddhi_sdk import AgentClient

agent = AgentClient(api_key="be_YOUR_KEY")

# Create an agent with tools
info = agent.create_agent(
    name="HR Compliance Bot",
    domain="hr",
    commitment_level=1,
    tools=[
        {"name": "check_leave", "action_type": "DB_READ",
         "description": "Check employee leave balance"},
        {"name": "approve_leave", "action_type": "DB_WRITE",
         "description": "Approve leave request"},
    ],
)

# Generate a guardrail-validated plan
plan = agent.generate_plan("Approve 5-day leave for employee #42")
for step in plan.steps:
    if step.is_safe:
        print(f"✓ {step.action} → {step.tool}")
    else:
        print(f"✗ BLOCKED: {step.action} — {step.guardrail_result}")

# Add memory facts for grounding
agent.add_memory_fact("Employee #42 has 8 CL remaining this year")

Causal Forward Model — ⚗️ Experimental (Professional+)

from buddhi_sdk import CFMClient

cfm = CFMClient(api_key="be_YOUR_KEY")

# Predict a causal chain
result = cfm.predict_chain(
    initial_state={"dosage_mg": 50, "patient_weight_kg": 70},
    actions=["check_allergies", "administer_drug", "monitor_vitals"],
    domain="medical",
)

print(f"Peak energy: {result.max_energy:.4f}")
print(f"Critical steps: {result.critical_steps}")
for step in result.chain:
    risk = "🔴" if step.energy > 0.8 else ("🟡" if step.energy > 0.5 else "🟢")
    print(f"  {risk} {step.action} (energy={step.energy:.3f})")

Setup

1. Sign up at buddhiengine.com
2. Create an API key in Dashboard → API Keys
3. Create a guardrail policy in Dashboard → Guardrails (or use a template)
4. Install: pip install buddhi-sdk

Detailed Usage

Validate Against a Specific Policy

result = client.validate(
    "Delete all user data",
    policy_id="data-protection-policy-a1b2c3d4",
)
print(result.reason)
print(result.rule_triggered)

Use in a Chatbot / AI Pipeline

from buddhi_sdk import RakshaClient

client = RakshaClient(api_key="be_YOUR_KEY")

def chatbot(user_message: str) -> str:
    check = client.validate(user_message)

    if check.is_blocked:
        return f"I can't help with that. ({check.reason})"

    # Safe — call your AI model
    return my_llm.generate(user_message)

Rich Context (Dict Input)

result = client.validate({
    "user_message": "Book a meeting at 3am",
    "user_role": "intern",
    "department": "finance",
})

Evaluation Mode Selection

from buddhi_sdk import RakshaClient, MODE_HYBRID, MODE_BM_ONLY, MODE_DSL_ONLY

# Hybrid (default) — BM semantic + AERM YAML rules
result = client.validate(text, mode=MODE_HYBRID)

# BM Only — semantic pipeline only, faster
result = client.validate(text, mode=MODE_BM_ONLY)

# DSL Only — YAML rules only, fastest
result = client.validate(text, mode=MODE_DSL_ONLY)

Batch Evaluation (Professional+)

results = client.validate_batch([
    {"input_context": "message 1"},
    {"input_context": "message 2", "policy_id": "my-policy"},
])
for r in results.results:
    print(r.decision)

KB Operations — Full API

from buddhi_sdk import KnowledgeBaseClient

kb = KnowledgeBaseClient(api_key="be_YOUR_KEY")

# Ingest
r = kb.ingest("Full policy text...", doc_id="doc-001", domain="hr")
print(f"Compressed {r.compression_ratio}x — {r.storage_bytes} bytes")

# Query
results = kb.query("maternity leave rules", limit=5)
for m in results.results:
    print(f"  {m.doc_id}: {m.score:.3f}")

# Stats
stats = kb.stats()
print(f"{stats.document_count} docs, {stats.total_storage_bytes} bytes")

# Remove
kb.remove("doc-001")

# Export/Import
kb.export("backup.buddhi")
kb.import_file("backup.buddhi", merge=True)

Agent Memory Management

from buddhi_sdk import AgentClient

agent = AgentClient(api_key="be_YOUR_KEY", agent_id="my-agent")

# Add facts
agent.add_memory_fact("Customer credit limit is $5,000", source="crm")
agent.add_memory_fact("Last order was placed on 2026-03-15", source="orders")

# List facts
for fact in agent.list_memory_facts():
    print(f"  [{fact.source}] {fact.content}")

# Remove a fact
agent.remove_memory_fact(fact_id="abc-123")

Agent with CFM Integration

from buddhi_sdk import AgentClient

agent = AgentClient(api_key="be_YOUR_KEY")

# Create agent with CFM model attached
info = agent.create_agent(
    name="Risk-Aware Agent",
    domain="financial",
    cfm_id="cfm-financial",  # Attach CFM model
    tools=[
        {"name": "check_balance", "action_type": "DB_READ"},
        {"name": "execute_trade", "action_type": "DB_WRITE"},
    ],
)

# Plan will include energy-based risk from CFM
plan = agent.generate_plan(
    goal="Execute $1M trade on NIFTY futures",
    use_cfm=True,  # Enable CFM simulation on plan steps
)

WebSocket Agent Integration

import asyncio
import json
from buddhi_sdk import AsyncAgentClient

async def main():
    async with AsyncAgentClient(api_key="be_...", agent_id="abc") as agent:
        plan = await agent.generate_plan("Monitor reactor temperature")

        if plan.overall_decision == "APPROVE":
            import websockets
            async with websockets.connect("wss://your-instance/ws") as ws:
                for step in plan.steps:
                    if step.is_safe:
                        await ws.send(json.dumps({
                            "type": "agent_action",
                            "action": step.action,
                            "tool": step.tool,
                        }))
                        response = await ws.recv()
                        print(f"Result: {json.loads(response)}")

asyncio.run(main())

List Policies / Instances / Usage

# Policies
for policy in client.list_policies():
    print(f"{policy.name} (rules={policy.rule_count})")

# Usage
usage = client.get_usage()
print(f"Used: {usage.monthly_usage}/{usage.monthly_limit}")

# Instances
for inst in client.list_instances():
    print(f"{inst.instance_id}: {inst.status}")

# Health
health = client.health()
print(f"Status: {health.status}, Tier: {health.tier}")

Async Clients

Every client has an async counterpart:

import asyncio
from buddhi_sdk import AsyncRakshaClient, AsyncAgentClient, AsyncCFMClient, AsyncKnowledgeBaseClient

async def main():
    async with AsyncRakshaClient(api_key="be_...") as client:
        result = await client.validate("Check this message")
        print(result.decision)

    async with AsyncKnowledgeBaseClient(api_key="be_...") as kb:
        await kb.ingest("Document text", domain="hr")

asyncio.run(main())

Self-Hosted / Custom Base URL

client = RakshaClient(
    api_key="be_YOUR_KEY",
    base_url="https://guardrails.yourcompany.internal/api/v1",
)

Response Objects

EvaluateResult (RakshaClient)

result = client.validate("some text")

result.decision          # str: APPROVE, BLOCK, REJECT, CONTINUE, WARN
result.reason            # str or None
result.severity          # str or None
result.rule_triggered    # str or None
result.policy_id         # str or None
result.evaluated_rules   # int
result.execution_time_ms # float
result.is_allowed        # True if APPROVE or CONTINUE
result.is_blocked        # True if BLOCK, BLOCKED, or REJECT
result.needs_review      # True if REVIEW or WARN

AgentPlan (AgentClient)

plan = agent.generate_plan("Approve leave")

plan.agent_id            # str
plan.goal                # str
plan.steps               # List[PlanStep]
plan.overall_decision    # "APPROVE" or "BLOCK"
plan.execution_time_ms   # float

# Each step:
step.action              # str: TOOL_CALL, EVALUATE, etc.
step.tool                # str or None
step.is_safe             # bool
step.guardrail_result    # str or None
step.energy              # float or None (if CFM attached)

CausalChainResult (CFMClient)

result = cfm.predict_chain(...)

result.chain             # List[CausalChainStep]
result.total_energy      # float
result.max_energy        # float
result.critical_steps    # int (steps with energy > 0.8)

# Each chain step:
step.step                # int
step.action              # str
step.state               # Dict[str, Any]
step.energy              # float
step.state_change        # float

Error Handling

from buddhi_sdk import (
    RakshaClient,
    AuthenticationError,       # Bad API key (401)
    QuotaExceededError,        # Monthly limit reached (429)
    PolicyNotFoundError,       # Policy doesn't exist (404)
    InstanceUnavailableError,  # No running AERM instance (503)
    TierRestrictionError,      # Feature requires higher tier (403)
    ExperimentalFeatureError,  # CFM not available on current tier (403)
    ServerError,               # Unexpected server error (5xx)
)

try:
    result = client.validate("test")
except AuthenticationError:
    print("Bad API key")
except TierRestrictionError as e:
    print(f"Requires {e.required_tier}+ tier")
except ExperimentalFeatureError as e:
    print(f"{e.feature} is experimental — upgrade to {e.required_tier}")

Examples

File	What It Demonstrates
example.py	Guardrail validation, policies, mode selection, KB ingest/query
example_kb.py	Full KB lifecycle: ingest, query, stats, export, remove
example_agent.py	Agent CRUD, memory facts, plan generation, blocked goals
example_cfm.py	CFM chain prediction across medical, financial, satellite domains
example_agent_websocket.py	Agent + WebSocket streaming, async patterns, LLM integration

Pricing Tiers

Plan	Monthly Calls	Overage	Batch	Agent	CFM	KB Storage
Starter/Free	500	Blocked (429)	No	No	No	10 MB
Professional	10,000	$0.012/call	Yes	Yes	⚗️ Yes	1 GB
Business	100,000	$0.006/call	Yes	Yes	Yes	10 GB
Enterprise	Unlimited	Negotiated	Yes	Yes	Yes	Unlimited

KB storage is compressed ~25× using geometric signatures.

Requirements

• Python >= 3.8
• httpx >= 0.24.0
• pydantic >= 2.0.0
• Optional: websockets for WebSocket examples

License

MIT — Buddhi Engineering