Metadata-Version: 2.4
Name: sernixa
Version: 0.6.3
Summary: Sernixa Python SDK — governed tool and agent execution with MCP call-gate support
Author: Sernixa Team
License-Expression: MIT
Project-URL: Repository, https://github.com/abhishekdhull63/Sernixa.ai-Web
Project-URL: Documentation, https://github.com/abhishekdhull63/Sernixa.ai-Web/tree/main/docs
Project-URL: Changelog, https://github.com/abhishekdhull63/Sernixa.ai-Web/tree/main/packages/sernixa/CHANGELOG.md
Keywords: sernixa,mcp,governance,agent,ai-safety,sdk,tool-governance
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.10
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Security
Classifier: Topic :: Software Development :: Libraries
Classifier: Typing :: Typed
Requires-Python: >=3.10
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: httpx>=0.28.0
Provides-Extra: langchain
Requires-Dist: langchain>=1.0.0; extra == "langchain"
Provides-Extra: crewai
Requires-Dist: crewai>=0.80.0; extra == "crewai"
Provides-Extra: frameworks
Requires-Dist: langchain>=1.0.0; extra == "frameworks"
Requires-Dist: crewai>=0.80.0; extra == "frameworks"
Dynamic: license-file

# Sernixa Python SDK

Sernixa gates Python function execution through a deterministic approval engine.
The backend decides whether an action is allowed, auto-approved, blocked, or
pending human review; the SDK executes the wrapped business function locally
only after an approved-style decision.

The SDK is intentionally small: it does not execute business logic in the
Sernixa backend, and it does not pretend to be a full agent framework. It gives
you low-friction interception points for plain Python, LangChain, CrewAI, and
MCP-style tool boundaries.

## Before You Start

You need:

- A Sernixa workspace.
- A Sernixa API base URL, for example `https://api.sernixa.com` for hosted
  environments or `http://localhost:8000` for local development.
- A browser-issued access token or an organization API key for automation.
- One function, tool, or agent action to protect first.

The SDK sends governance metadata to Sernixa. Your function still runs in your
process, and only after Sernixa returns an approved-style decision.

## Quickstart

Install from PyPI:

```bash
pip install sernixa
```

For local development from this checkout:

```bash
pip install -e packages/sernixa
```

Configure the API endpoint and key:

```bash
export SERNIXA_BASE_URL=https://api.sernixa.com
export SERNIXA_API_KEY="<your-org-api-key>"
```

For local development, use the local FastAPI backend instead:

```bash
export SERNIXA_BASE_URL=http://localhost:8000
```

```python
import sernixa

@sernixa.intercept(
    intent_id="customer-notes",
    risk_level="LOW",
    operation_class="read",
    data_sensitivity="internal",
    systems_touched=["postgres"],
    metadata={"ticket": "DEMO-1"},
)
def summarize_customer(customer_id: str) -> str:
    return f"summary for {customer_id}"

print(summarize_customer("cus_123"))
```

If Sernixa returns `approved`, `auto_approved`, or `executed`, the function runs
locally. If it returns `pending_review`, the SDK polls until a reviewer approves
or rejects the action, or until the configured timeout expires.

## First Run Checklist

1. Create or choose a Sernixa workspace.
2. Sign in interactively or create an organization API key for automation.
3. Export `SERNIXA_BASE_URL` and the selected credential.
4. Wrap one low-risk function with `@sernixa.intercept(...)`.
5. Run the function and confirm the decision appears in the Command Center.
6. Increase risk levels only after the low-risk path is working.

High and critical risk actions are not silently auto-approved. Expect them to
wait for human review unless policy blocks them outright.

## Configuration

- `SERNIXA_BASE_URL`: Sernixa API URL. Defaults to `http://localhost:8000`.
- `SERNIXA_ACCESS_TOKEN`: revocable browser session token. Takes precedence over an API key.
- `SERNIXA_API_KEY`: bearer token for the API.
- `SERNIXA_POLL_INTERVAL_SECONDS`: approval poll interval. Defaults to `2`.
- `SERNIXA_POLL_TIMEOUT_SECONDS`: approval wait timeout. Defaults to `600`.
- `SERNIXA_TIMEOUT_SECONDS`: per-request HTTP timeout. Defaults to `10`.
- `SERNIXA_MAX_RETRIES`: retries for transient network errors, `429`, and `5xx`. Defaults to `2`.
- `SERNIXA_MAX_ASYNC_IN_FLIGHT`: maximum concurrent async SDK HTTP requests per event loop. Defaults to `20`.
- `SERNIXA_SDK_DLP_MODE`: metadata DLP mode. Defaults to `mask`; use `warn`, `block`, or `off` intentionally.
- `SERNIXA_CAPTURE_ARGUMENTS`: set to `true` to include function argument values in submitted metadata. Defaults to `false`.
- `SERNIXA_LOG_LEVEL`: optional SDK logger level such as `INFO` or `DEBUG`.
- `SERNIXA_DEBUG`: set to `true` for verbose SDK request/decision logging.

You can also configure a client directly:

```python
import os
from sernixa import Client

client = Client(
    base_url="http://localhost:8000",
    access_token=os.environ.get("SERNIXA_ACCESS_TOKEN"),
    api_key=os.environ.get("SERNIXA_API_KEY"),
    poll_interval=1,
    poll_timeout=120,
    timeout_seconds=10,
    max_retries=2,
    max_async_in_flight=20,
)
```

Validate the configured key and evaluate a policy input directly:

```python
identity = client.whoami()
decision = client.governance_test({"operation_class": "read"})
```

Backend-authoritative approval continuity and evidence export are also available
without constructing raw HTTP requests:

```python
approval = client.approval_detail("approval_123")

bundle = client.export_evidence_bundle(approval_id="approval_123")
verification = client.verify_evidence_bundle(bundle)
if verification.get("valid") is not True:
    raise RuntimeError("Evidence bundle verification failed")
```

For org or time-range evidence, omit `approval_id` or provide both `date_from`
and `date_to`. Ambiguous approval/team/date scopes and limits outside
`1..5000` fail locally before a request is sent. These methods return backend
records; they do not reinterpret a cached local status as an approval or a valid
evidence artifact.

For an interactive application, use the browser device flow and then create a
client from its revocable session token:

```python
pending = client.start_cli_device_login(device_name="Developer workstation")
print(pending.verification_uri, pending.user_code)
login = client.poll_cli_device_login(pending.device_code)

session = Client(access_token=login.access_token)
plan = session.current_plan()
flight = session.feature_availability("flight_recorder")
```

The backend remains authoritative for roles, plans, and feature availability.
Call `logout_cli_session()` to revoke a browser session. API keys remain the
recommended credential for unattended automation.

If `whoami()` fails, fix the base URL or API key before wrapping production
tools.

## Intercepting Functions

```python
@client.intercept(
    intent_id="billing-adjustment",
    risk_level="HIGH",
    operation_class="financial",
    data_sensitivity="financial",
    systems_touched=["stripe", "postgres"],
)
def adjust_invoice(invoice_id: str, amount_cents: int) -> None:
    ...
```

High and critical risk actions are never auto-approved by Sernixa. They remain pending review even when similar low-risk actions have a strong approval history.

## Async Functions

The same decorator works on coroutines:

```python
@client.intercept(
    intent_id="agent-read",
    risk_level="LOW",
    operation_class="read",
    data_sensitivity="internal",
    systems_touched=["vectordb"],
)
async def run_agent(query: str) -> str:
    return await agent.ainvoke(query)
```

Async clients reuse one `httpx.AsyncClient` per event loop and cap concurrent SDK
requests with `SERNIXA_MAX_ASYNC_IN_FLIGHT`. Long-running async services should
close reusable sessions during shutdown:

```python
async with Client() as client:
    await protected_tool()

# or
client = Client()
...
await client.aclose()
```

## Decision Handling

```python
from sernixa import is_approved_status, is_terminal_status

assert is_approved_status("auto_approved")
assert is_terminal_status("rejected")
```

The SDK raises explicit exceptions for reviewer and policy outcomes:

```python
from sernixa.exceptions import (
    SernixaBlockedError,
    SernixaConfigurationError,
    SernixaExpiredError,
    SernixaRateLimitError,
    SernixaRejectedError,
    SernixaTimeoutError,
    SernixaValidationError,
)

try:
    adjust_invoice("inv_123", 1000)
except SernixaBlockedError as exc:
    print(f"Blocked by policy: {exc.reason}")
except SernixaRejectedError as exc:
    print(f"Rejected by reviewer: {exc.reason}")
except SernixaExpiredError:
    print("Approval expired before a reviewer decided.")
except SernixaTimeoutError:
    print("SDK timed out waiting for a decision.")
```

Configuration and payload mistakes fail before the business function runs:

- `SernixaConfigurationError`: invalid base URL, timeout, poll, or retry settings.
- `SernixaValidationError`: malformed action metadata such as an empty `intent_id`, invalid `risk_level`, or non-JSON metadata.
- `SernixaRateLimitError`: the backend kept returning `429` after retries.

## Extra Review Metadata

Use `metadata` for non-sensitive context that helps the approver understand the action:

```python
@sernixa.intercept(
    intent_id="support-note",
    risk_level="LOW",
    operation_class="update",
    data_sensitivity="internal",
    systems_touched=["postgres"],
    metadata={"change_ticket": "SUP-1842", "env": "local-demo"},
)
def write_support_note(...):
    ...
```

Core governance fields such as `risk_level` and `idempotency_key` cannot be overridden by extra metadata.

By default, the SDK does not upload function argument values. It sends argument
counts and keyword names only, and masks sensitive-looking metadata before the
request leaves the process. To include full argument values for a trusted local
debugging session, set `SERNIXA_CAPTURE_ARGUMENTS=true`; keep the default for
shared development and production-like runs.

## V3 Delegation Context

For local multi-agent workflows, create a delegation token through Sernixa and
attach it while any delegatee agent runs protected tools:

```python
from sernixa import Client, delegation_scope, with_delegation

client = Client()
token = client.create_delegation_token(
    delegator_agent_id="orchestrator-agent",
    delegatee_agent_id="worker-agent",
    scope=delegation_scope(
        max_risk_level="low",
        allowed_operation_classes=["read"],
        allowed_data_sensitivities=["internal"],
        resources={"repo": ["sernixa"]},
    ),
    tool_subset=["view-details"],
)

with with_delegation(
    agent_id="worker-agent",
    token_id=token["token_id"],
    signing_secret=os.environ["SERNIXA_REQUEST_SIGNING_SECRET"],
    chain_id=token["chain_id"],
    runtime_id="local-worker-runtime",
    service_identity="spiffe://local/agent/worker-agent",
):
    view_details()
```

The SDK signs each delegated request with a canonical envelope, timestamp,
nonce, request body hash, key ID, and runtime identity metadata. Sernixa
verifies the request signature, replay status, token signature, hash chain,
expiry, delegatee identity, and scope before the existing approval logic runs.

## LangChain Adapter

For current LangChain agents, add one middleware instance instead of wrapping every tool:

```python
from langchain.agents import create_agent
from sernixa import Client
from sernixa.adapters import sernixa_middleware

agent = create_agent(
    model="openai:gpt-5",
    tools=[search_customers, update_customer],
    middleware=[
        sernixa_middleware(
            client=Client(),
            environment="production",
            data_sensitivity="internal",
            systems_touched=["crm"],
        )
    ],
)
```

The middleware uses LangChain's current `wrap_tool_call` and `awrap_tool_call` lifecycle. It calls the handler exactly once for `allow`, raises `SernixaReviewRequiredError` for `review`, and raises `SernixaBlockedError` for `deny`. Tool arguments and adapter metadata are redacted before evaluation.

Use the decorator when you control the tool function:

```python
from langchain.tools import tool
from sernixa.adapters import langchain_tool

@tool
@langchain_tool(
    intent_id="finance-tool",
    risk_level="HIGH",
    operation_class="financial",
    data_sensitivity="financial",
    systems_touched=["stripe"],
)
def transfer_funds_tool(amount_cents: int, destination_account: str) -> str:
    """Transfer funds after Sernixa approval."""
    ...
```

Use the object proxy when a tool object already exists:

```python
from sernixa.adapters import secure_langchain_tool

protected_tool = secure_langchain_tool(
    existing_tool,
    intent_id="customer-lookup",
    risk_level="LOW",
    operation_class="read",
    data_sensitivity="internal",
    systems_touched=["crm"],
)

result = protected_tool.invoke({"customer_id": "cus_123"})
```

Install LangChain separately if you use the adapter:

```bash
pip install "sernixa[langchain]"
```

The dependency-free proxy remains available for older or custom tool pipelines. It guards `invoke`, `ainvoke`, `run`, `arun`, and direct calls where the underlying tool exposes them, and forwards unknown attributes to the wrapped tool.

## CrewAI Adapter

Use the decorator for plain functions that become CrewAI tools:

```python
from sernixa.adapters import crewai_tool

@crewai_tool(
    intent_id="ticket-update",
    risk_level="HIGH",
    operation_class="update",
    data_sensitivity="internal",
    systems_touched=["ticketing"],
)
def update_ticket(ticket_id: str, note: str) -> str:
    return "updated"
```

Use the object proxy for existing CrewAI-style tool objects:

```python
from sernixa.adapters import secure_crewai_tool

protected_tool = secure_crewai_tool(
    existing_tool,
    intent_id="crew-ticket-update",
    risk_level="HIGH",
    operation_class="update",
    data_sensitivity="internal",
    systems_touched=["ticketing"],
)

protected_tool.run("SEC-1842")
```

Install CrewAI separately if you use CrewAI itself:

```bash
pip install "sernixa[crewai]"
```

The SDK proxy supports the common `run`, `_run`, and direct-call execution
shapes without requiring CrewAI as a hard dependency.

## MCP Boundary Helpers

The local SDK helper governs an in-process MCP-style tool boundary:

```python
from sernixa.adapters import McpToolBoundary

boundary = McpToolBoundary(server_name="workspace-mcp", toolset_id="toolset-prod")

def read_file(path: str) -> str:
    return open(path).read()

result = boundary.invoke(
    tool_name="read_file",
    arguments={"path": "/workspace/report.md"},
    handler=read_file,
    intent_id="mcp-read-file",
    risk_level="LOW",
    operation_class="read",
    data_sensitivity="internal",
    client_name="local-agent-host",
)
```

Place this at the host/router boundary before dispatching a tool call to the
underlying MCP server/tool implementation. For registered remote and stdio MCP
servers, the backend's strict gateway owns protocol initialization, the
initialized notification, session/version continuity, and `tools/call`.

## Gateway Run Wrapper

`SernixaGateway` is the target high-level developer surface for governing an
existing agent or MCP/tool run call without rewriting the agent. It wraps the
current approval oracle and records gateway metadata on the action request.

```python
import os

from sernixa import RuntimeSensorConfig, SernixaGateway

sensor = RuntimeSensorConfig(
    # Explicit argv only; the SDK never invokes a shell.
    command=("/opt/sernixa/bin/sernixa-host-sensor",),
    api_url=os.environ["SERNIXA_BASE_URL"],
    org_id=os.environ["SERNIXA_ORG_ID"],
    collector_id=os.environ["SERNIXA_COLLECTOR_ID"],
    ingest_secret=os.environ["KERNEL_COLLECTOR_INGEST_SECRET"],
)

with SernixaGateway(
    api_key=os.environ["SERNIXA_API_KEY"],
    mcp_profile="prod-tools",
    enforce_ebpf=True,
    auto_start_runtime_sensor=True,
    runtime_sensor_config=sensor,
) as gateway:
    result = gateway.run(agent.run, input=user_task)
```

With `auto_start_runtime_sensor=True`, the SDK validates and starts the installed
host sensor, consumes its JSONL stream, signs Flight Recorder batches, tracks
backend acceptance, privately spools failed non-empty batches, and terminates
the process on `close()` or context exit. Required evidence fails closed unless
the backend accepts a sensor event or heartbeat before the bounded startup
timeout. The ingest secret remains in the SDK process and is never copied into
the privileged sensor environment.

The SDK does not silently install or privilege a kernel component. Linux eBPF,
macOS Endpoint Security, and Windows ETW/eBPF sensors still require the
platform's explicit installation, entitlement, signing, and privilege steps.
If automatic lifecycle is not requested, `enforce_ebpf=True` only checks
backend readiness, preserving the existing deployment contract.

## Agent Plan Guards

Use the plan guard when a chat agent, Claude-style hook runner, or Codex-style
wrapper can declare intended tool calls before execution:

```python
from sernixa import Client
from sernixa.adapters import SernixaAgentGuard, chat_plan_step

guard = SernixaAgentGuard(
    Client(),
    agent_type="chat",
    user_identity={"email": "operator@example.com"},
    context={"workspace": "support", "channel": "slack"},
)
guard.declare_plan([
    chat_plan_step(
        tool_name="calendar.create",
        action="calendar.write",
        arguments={"title": "Review"},
        resource="calendar:primary",
    )
])
guard.execute_tool(
    calendar_create,
    tool_name="calendar.create",
    action="calendar.write",
    arguments={"title": "Review"},
    resource="calendar:primary",
)
```

`declare_plan()` calls the shared backend endpoint
`/api/controls/governance/agent-plan/evaluate`. Every step is evaluated through
the same governance engine and audit chain as `governance_test()`. Execution is
blocked if no plan was declared, the plan is denied/review-only, the requested
tool is unplanned, or a plan step is reused.

For a host integration that performs execution itself, explicitly consume the
backend-signed step proof immediately before dispatch:

```python
client = Client(plan_session_file=".sernixa/passport.json")
intent = client.governance_issue_intent_proof(
    provider="codex",
    session_id="session-123",
    turn_id="turn-123",
    cwd_digest=sha256_of_cwd,
    prompt=direct_user_prompt,
)
passport = client.governance_evaluate_plan(
    [{
        "step_id": "read-1",
        "tool_name": "Read",
        "action": "file.read",
        "arguments": {"path": "README.md"},
        "resource": "README.md",
    }],
    agent_type="codex",
    intent_proof=intent["intent_proof"],
)
authorization = client.governance_authorize_plan_step(
    tool_name="Read",
    action="file.read",
    arguments={"path": "README.md"},
    resource="README.md",
    provider="codex",
)
```

The API verifies the signed plan and exact hashes, checks organization/principal
binding and expiry, and atomically consumes the proof. A changed argument,
undeclared tool, or replay raises `SernixaBlockedError`. Authorization does not
execute the tool and does not replace host permissions.

Any exact action, target, destination, argument, credential use, privilege, or
destructive operation in a verified direct human prompt can proceed without
redundant review while retaining its risk classification and evidence. Scope
introduced by the agent instead returns `review`; retry the unchanged `plan_id`
and plan with its `approval_id` after approval. Hard organization policy,
failed security infrastructure, invalid or expired proof, and proof replay
remain blocked.

## State Model

- `executed`: Sernixa allowed the action and the SDK executed the local function.
- `auto_approved`: Sernixa policy allowed execution without human review.
- `pending_review`: the SDK is waiting for a reviewer and polling the approval.
- `rejected`: a reviewer denied the action; the SDK raises `SernixaRejectedError`.
- `blocked`: policy/security denied the action; the SDK raises `SernixaBlockedError`.
- `expired`: approval TTL elapsed; the SDK raises `SernixaExpiredError`.
- `failed`: backend replay/execution evidence failed; the SDK raises `SernixaError`.

## Troubleshooting

- `Action blocked`: inspect `exc.reason` and the approval/audit page. Dangerous primitives and invalid signatures fail closed.
- `SernixaValidationError`: fix empty IDs, invalid risk levels, empty `systems_touched`, or non-JSON metadata.
- `SernixaConfigurationError`: check `SERNIXA_BASE_URL`, timeout, poll, and retry values.
- `SernixaRateLimitError`: reduce agent concurrency or increase backend limits for the environment.
- `SernixaTimeoutError`: the approval is still pending after `SERNIXA_POLL_TIMEOUT_SECONDS`.
- Browser works but SDK fails: verify `SERNIXA_API_KEY` and that the backend URL is reachable from the Python process.

## Local And Hosted Configuration

Local demo:

```bash
export SERNIXA_BASE_URL=http://localhost:8000
export SERNIXA_API_KEY=e2e-admin
export SERNIXA_POLL_INTERVAL_SECONDS=1
```

Hosted or shared environment:

```bash
export SERNIXA_BASE_URL=https://sernixa.example.com
export SERNIXA_API_KEY="$SERNIXA_SERVICE_TOKEN"
export SERNIXA_TIMEOUT_SECONDS=10
export SERNIXA_MAX_RETRIES=2
```

Use real bearer tokens, HTTPS, shared nonce/rate-limit storage, and production
KMS/HSM signing before treating a hosted environment as production.

Current backend handoff note: local SDK examples target a configured API. A
production-like Sernixa backend must run behind the Auth.js/Google web session
boundary, accept only the short-lived API JWT minted by the web app, and use
Convex plus shared Redis for production runtime state.

TypeScript handoff note: the monorepo now includes `packages/sdk` as a real
Node client for action gating, approval polling, typed errors, and gateway
readiness checks, and the package is published on npm as `@sernixa/sdk`.
Python remains the fuller SDK for delegation request-envelope signing and
framework adapters.

## Release

The package is published on PyPI as `sernixa`. Use editable install from this
repository only when developing the SDK itself.

For local repository development, use editable install from the repo root:

```bash
pip install -e packages/sernixa
```

## Local Examples

- `examples/basic_auto_approval/`: local walkthrough showing low-risk, repeated approval memory, and high-risk pending behavior.
- `examples/multi_agent_delegation/`: generic orchestrator/worker delegation token flow.
- `examples/sernixa-sdk/`: lower-level sync, async, LangChain, CrewAI, MCP boundary, Node gateway, and compose smoke examples.
