eaigins-sdk 0.3.0

Enterprise AI Control Plane — Governance SDK for autonomous AI agents

eaigins-sdk

Enterprise AI Control Plane — Python Governance SDK

"Enterprise AI Control Planes govern AI decisions the way API gateways govern service calls." — Pavan Kumar, JIVA AI TECH, March 2026

---

What is Eaigins?

Eaigins (Enterprise Agentic Intelligence Governance Infrastructure Service) is a runtime governance layer that sits between your AI agents and enterprise infrastructure.

Every agent action passes through a 5-stage pipeline before it executes:

AgentAction
    ↓  [1] Runtime Interceptor    — capture & structure
    ↓  [2] AI Agent Firewall      — OWASP LLM Top 10 threat detection
    ↓  [3] Policy Enforcement Gateway — declarative governance rules
    ↓  [4] Risk Scoring Engine    — weighted multi-factor formula (paper §4.3)
    ↓  [5] Decision Provenance Engine — causal DAG + audit trail
    ↓
GovernanceDecision → ALLOW / ESCALATE / BLOCK

Zero hard dependencies. The core SDK requires only Python 3.10+. Framework integrations (LangChain, CrewAI, AutoGen) are optional extras.

---

Installation

# Core only (zero dependencies)
pip install eaigins-sdk

# With framework integrations
pip install eaigins-sdk[crewai]      # Financial Services / AML
pip install eaigins-sdk[langchain]   # Healthcare
pip install eaigins-sdk[autogen]     # Customer Care
pip install eaigins-sdk[all]         # Everything

---

60-Second Quickstart

from eaigins_sdk import ControlPlane, AgentAction
from eaigins_sdk.policies import AML_POLICIES

# 1. Create a control plane with a built-in policy set
cp = ControlPlane(policies=AML_POLICIES)

# 2. Build an AgentAction before calling any enterprise tool
action = AgentAction(
    agent_id        = "my-aml-agent-v1",
    prompt          = "File the SAR with FinCEN for case CASE-2024-001.",
    tool_name       = "file_sar",
    tool_params     = {"case_id": "CASE-2024-001", "customer_id": "C-88421"},
    context_sources = ["compliance_reports", "kyc_documents"],
    reasoning       = "Investigation complete. SAR threshold met.",
    metadata        = {"regulatory_framework": "BSA"},
)

# 3. Evaluate — runs all 5 pipeline stages
decision = cp.evaluate(action)

print(decision.decision.value)   # "BLOCK"
print(decision.reason)
# "Policies violated: 1 | Human approvals required: 1 | Risk: MEDIUM (0.64) |
#  🛑 BLOCKED — Action prevented from reaching enterprise systems."

---

Core Concepts

AgentAction

The primary input — describes what the agent wants to do:

AgentAction(
    agent_id        = "crewai::aml-crew::compliance-officer",
    prompt          = "Natural language intent",
    tool_name       = "enterprise_tool_name",      # matched against policies
    tool_params     = {"amount": 85000, ...},       # matched against policy rules
    context_sources = ["transaction_history", ...], # data sensitivity scoring
    reasoning       = "Agent's stated reason",      # captured in provenance
    metadata        = {"regulatory_framework": "FATF"},
)

GovernanceDecision

The complete pipeline output:

decision.decision        # ExecutionDecision.ALLOW / ESCALATE / BLOCK
decision.risk_level      # RiskLevel.LOW / MEDIUM / HIGH / CRITICAL
decision.risk_score      # float 0.0–1.0
decision.allowed         # bool convenience property
decision.blocked         # bool convenience property
decision.escalated       # bool convenience property
decision.reason          # human-readable explanation
decision.firewall_result # FirewallResult from Stage 2
decision.policy_result   # PolicyResult from Stage 3
decision.risk_score_detail  # RiskScore with per-factor breakdown
decision.provenance_nodes   # list[ProvenanceNode] — the causal DAG
decision.to_dict()          # JSON-serialisable audit record
decision.raise_if_blocked() # raises GovernanceViolationError if not ALLOW

Risk Formula (from the paper, §4.3)

Risk = (DataSensitivity × 0.30) + (FinancialImpact × 0.25)
     + (RegulatoryExposure × 0.25) + (PolicyViolationProb × 0.20)

Thresholds: LOW [0.00, 0.35) → ALLOW · MEDIUM [0.35, 0.65) → ESCALATE · HIGH [0.65, 0.85) → BLOCK · CRITICAL [0.85, 1.00] → BLOCK

---

Built-in Policy Sets

Domain	Import	Count	Frameworks
Financial / AML	AML_POLICIES	7 policies	BSA · FATF · FinCEN
Healthcare	HEALTHCARE_POLICIES	8 policies	HIPAA · FDA 21 CFR · Joint Commission
Customer Care	CUSTOMERCARE_POLICIES	8 policies	GDPR · CCPA · PCI-DSS

from eaigins_sdk.policies import AML_POLICIES, HEALTHCARE_POLICIES, CUSTOMERCARE_POLICIES

---

Writing Custom Policies

from eaigins_sdk.policy import Policy, PolicyGateway

no_weekend_deploys = Policy(
    policy_id   = "OPS-001",
    name        = "No Weekend Deployments",
    description = "AI agents may not trigger production deployments on weekends.",
    category    = "OPERATIONAL",
    blocks_action = True,           # hard BLOCK if violated
    requires_human_approval = False,
    evaluate    = lambda action: (
        not (action.tool_name == "deploy_to_production" and
             __import__("datetime").datetime.now().weekday() >= 5),
        "Production deployments on weekends are blocked."
    )
)

cp = ControlPlane(policies=[no_weekend_deploys])

---

@governed_tool Decorator

Zero-boilerplate governance for any Python function:

from eaigins_sdk.decorators import governed_tool
from eaigins_sdk import ControlPlane, GovernanceViolationError
from eaigins_sdk.policies import AML_POLICIES

cp = ControlPlane(policies=AML_POLICIES)

@governed_tool(
    cp,
    tool_name       = "process_wire_transfer",
    context_sources = ["banking_records", "wire_transfer_details"],
    agent_id        = "my-financial-agent",
)
def process_wire_transfer(account_id: str, amount: float) -> dict:
    # Only executed if Eaigins returns ALLOW
    return call_payment_api(account_id, amount)

try:
    result = process_wire_transfer(account_id="ACC-001", amount=150_000)
except GovernanceViolationError as e:
    print(e.decision.decision.value)  # "BLOCK" or "ESCALATE"
    print(e.decision.reason)

---

Framework Integrations

LangChain

from langchain.agents import AgentExecutor, create_react_agent
from eaigins_sdk.integrations import EaignsLangChainCallback
from eaigins_sdk import ControlPlane
from eaigins_sdk.policies import HEALTHCARE_POLICIES

cp       = ControlPlane(policies=HEALTHCARE_POLICIES)
callback = EaignsLangChainCallback(cp, agent_id="langchain::clinical-copilot")

executor = AgentExecutor(
    agent=agent, tools=tools, verbose=True,
    callbacks=[callback],
    handle_parsing_errors=True,
)

CrewAI

from crewai.tools import tool as crewai_tool
from eaigins_sdk.integrations import governed_crewai_tool
from eaigins_sdk import ControlPlane
from eaigins_sdk.policies import AML_POLICIES

cp = ControlPlane(policies=AML_POLICIES)

@crewai_tool("File SAR with Regulator")
@governed_crewai_tool(cp, tool_name="file_sar",
                      context_sources=["compliance_reports"])
def file_sar(case_id: str, customer_id: str) -> str:
    """Submit SAR to FinCEN. Requires human authorization."""
    return submit_to_fincen(case_id, customer_id)

AutoGen

from eaigins_sdk.integrations import EaignsAutoGenProxy
from eaigins_sdk import ControlPlane
from eaigins_sdk.policies import CUSTOMERCARE_POLICIES

cp = ControlPlane(policies=CUSTOMERCARE_POLICIES)

proxy = EaignsAutoGenProxy.create(
    cp,
    function_map={
        "process_refund": process_refund_fn,
        "export_data":    export_data_fn,
    },
    tool_context_map={
        "process_refund": ["payment_records"],
        "export_data":    ["crm_customer_profile"],
    },
    agent_id = "autogen::cx-support-agent",
)

---

Deployment Models (from the paper, §10)

Gateway  — one shared ControlPlane for all agents
           cp = ControlPlane(policies=ALL_POLICIES)

Sidecar  — one ControlPlane per service / container
           cp = ControlPlane(policies=DOMAIN_POLICIES, name="service-A")

SDK      — embedded in application code via @governed_tool or GovernedMixin

---

Audit Trail Export

# JSON audit trail — SIEM / regulatory ingestion
trail = cp.export_audit_trail()
# [{"action_id": "...", "decision": "BLOCK", "risk_score": 0.71, ...}, ...]

# Full JSON string
print(cp.export_audit_json())

# NetworkX graph for visualisation (requires networkx)
G = cp.provenance.to_networkx(action_id)

---

CLI

# Self-test (all domains, all pipeline stages)
eaigins validate

# Evaluate a JSON action file
eaigins evaluate action.json --domain financial

# Print a sample audit trail
eaigins export

# Version
eaigins version

action.json example:

{
    "agent_id": "my-agent",
    "prompt": "Export all customer records to analytics partner.",
    "tool_name": "export_customer_data",
    "tool_params": {"scope": "ALL", "include_pii": true, "destination": "external"},
    "context_sources": ["crm_customer_profile", "payment_records"],
    "reasoning": "Marketing team requested full export.",
    "metadata": {}
}

---

Configuring the Risk Engine

from eaigins_sdk.risk import RiskEngine

engine = RiskEngine(
    # Custom weights (must sum to 1.0)
    weights={
        "data_sensitivity":      0.40,
        "financial_impact":      0.20,
        "regulatory_exposure":   0.20,
        "policy_violation_prob": 0.20,
    },
    # Extend built-in data sensitivity map
    extra_data_sensitivity={
        "my_proprietary_db":   0.85,
        "internal_ip_store":   0.90,
    },
    # Extend financial impact map
    extra_financial_impact={
        "my_payment_gateway": 0.80,
    },
)

cp = ControlPlane(risk_engine=engine, policies=AML_POLICIES)

---

Session Statistics

stats = cp.stats
# {
#   "name": "my-control-plane",
#   "intercepted": 42,
#   "allowed": 28,
#   "escalated": 8,
#   "blocked": 6,
#   "policies": 7,
# }

---

Reference

Based on: Enterprise AI Control Plane — Governing Autonomous AI Decision Systems Pavan Kumar, JIVA AI TECH, March 2026

Aligned with:

• EU AI Act (transparency and traceability requirements)
• NIST AI Risk Management Framework (AI RMF)
• ISO/IEC 42001 (AI management systems)
• OWASP Top 10 for LLM Applications (threat detection)

---

License

MIT License © 2026 Pavan Kumar, JIVA AI TECH