sakshi-sdk

Register your AI agents and witness their decisions. Python SDK for the Sakshi governance platform.

pip install sakshi-sdk   # imports as `sakshi`

from sakshi import SakshiClient

sakshi = SakshiClient("https://console.example.bank", api_key="...")

agent_id = sakshi.register(
    "loan-decision-agent",
    owner_name="Priya Sharma",
    owner_email="priya@example.bank",
    autonomy_tier="L2",
    blast_radius={"customer_facing": True, "spend_limit_inr": 500_000},
)

with sakshi.witness(
    agent_id,
    context={"application_id": "4471", "retrieved": chunks},
    model={"provider": "openai", "model": "gpt-5.2", "version": "2026-05"},
    client_ref="app-4471",  # idempotency key
) as w:
    w.step("plan", goal="assess repayment capacity")
    w.tool("bureau_pull", output=bureau_response)
    w.human("async_review", "arun@example.bank", verdict="approved")
    w.action(decision="approve", limit_inr=200_000)

Middlewares (all duck-typed — the SDK carries zero provider or framework dependencies):

from sakshi.middleware import watch_openai_compatible, witness_node, watch_langgraph, watch_mcp

llm = watch_openai_compatible(openai_client)   # also: watch_anthropic,
                                               # watch_gemini, watch_bedrock

# LangGraph (or any graph framework): per-node + per-run evidence
graph_builder.add_node("assess", witness_node(assess))
app = watch_langgraph(graph_builder.compile(), name="loan-flow")
# MCP (Model Context Protocol): witness tool calls + detect
# tool-manifest poisoning (OWASP MCP Top 10)
session = watch_mcp(mcp_client_session, server="tools.example")

# Agent frameworks: one object per framework, witnessing (and, for ADK and
# Strands, tool-path enforcement) across the whole agent run
from sakshi.middleware import SakshiAdkPlugin, SakshiCrewListener, SakshiStrandsHooks
runner = Runner(agent=root_agent, ...,          # Google ADK
                plugins=[SakshiAdkPlugin(client, "loan-decision-agent", enforce=True)])
SakshiCrewListener()                            # CrewAI (registers on its event bus)
agent = Agent(model=model, tools=[...],         # AWS Strands
              hooks=[SakshiStrandsHooks(client, "loan-decision-agent", enforce=True)])

Inside an active witness block, every LLM call becomes an llm_call step with latency, token usage, and the served model identity (RBI draft para 56), and every graph node becomes a graph_node step with the state keys it updated. Works unchanged with self-hosted Ollama/vLLM via the OpenAI-compatible client. No active session — calls pass through untouched.

Regulatory helpers (RBI draft MRM 59(ii)-(iii), SEBI CP-P2):

with sakshi.witness(agent_id) as w:
    w.disclosure(channel="app", method="banner")   # told the customer it's AI
    ...

# customer asked for a human — synchronous, raises on failure
sakshi.request_human_handoff(agent_id, client_ref="app-4471",
                             reason="customer requested a human")

disclosure() records an ai_disclosure touchpoint (coverage becomes measurable evidence); request_human_handoff() parks a review item in the agent's team queue. A lost handoff is a compliance failure, so it never buffers or drops.

Delivery & retry semantics (v0.2):

• register() is idempotent by name — rerunning your startup script returns the existing agent, never a duplicate.
• witness() capture retries with jittered exponential backoff and is idempotent by client_ref — retries can never double-record.
• enforce() never auto-retries: an evaluation is evidence and a routed action must not be double-enqueued. Unreachable platform raises SakshiEnforcementUnavailable (fail-closed) — your code decides.

Properties you can rely on:

• Fail-open by default — if the platform is unreachable, your agent keeps running; records are buffered, retried, and dropped with a warning as the last resort. Governance must never take production down. Use fail_open=False for synchronous capture that raises.
• Failures are evidence — an exception inside a witness block is captured as the decision outcome and re-raised.
• PII-safe by design — Aadhaar/PAN/mobile and other Indian identifiers are detected and tokenized server-side at ingest, before storage or hashing.
• register() is always synchronous: an unregistered agent should not run (RBI draft MRM guidance, para 21).

Call sakshi.flush() before shutdown in batch jobs; long-running services can rely on the atexit hook.