Python Wrapper

The openparallax-shield Python package provides a native Python API for Shield. It wraps the Go binary using JSON-RPC over stdin/stdout -- the same protocol MCP uses -- so you get the full 4-tier pipeline with no Go toolchain required.

Install

pip install openparallax-shield

The package automatically downloads a pre-built Go binary for your platform (linux/darwin/windows, amd64/arm64) on first use. No Go compiler needed.

Requirements

• Python 3.9+
• No native dependencies (the Go binary is self-contained)

How It Works

The Python wrapper spawns a lightweight Go process that communicates over JSON-RPC via stdin/stdout:

Python process          Go binary (stdin/stdout)
┌──────────────┐        ┌────────────────────────┐
│  Shield()    │──JSON──▶│  Shield Pipeline       │
│  .evaluate() │  RPC   │  Tier 0, 1, 2          │
│  .status()   │◀──JSON──│  ONNX, heuristic, LLM │
└──────────────┘        └────────────────────────┘

The Go process is long-lived -- it starts once when you create a Shield instance and handles all subsequent evaluation calls. The ONNX model (if installed) is loaded once and reused across calls.

API Reference

Shield

class Shield:
    def __init__(
        self,
        policy_file: str = "security/shield/default.yaml",
        heuristic_enabled: bool = True,
        fail_closed: bool = True,
        onnx_threshold: float = 0.85,
        rate_limit: int = 60,
        daily_budget: int = 100,
        verdict_ttl: int = 300,
        evaluator: dict | None = None,
        canary_token: str | None = None,
    ):
        """Initialize Shield with the given configuration."""

Parameters:

Parameter	Type	Default	Description
policy_file	str	"security/shield/default.yaml"	Path to the YAML policy file
heuristic_enabled	bool	True	Enable heuristic pattern matching
fail_closed	bool	True	Block on errors (vs. allow with reduced confidence)
onnx_threshold	float	0.85	ONNX INJECTION confidence threshold for BLOCK
rate_limit	int	60	Max evaluations per minute
daily_budget	int	100	Max Tier 2 evaluations per day
verdict_ttl	int	300	Verdict cache TTL in seconds
evaluator	dict | None	None	Tier 2 LLM evaluator config
canary_token	str | None	None	Canary token for evaluator injection detection

Shield.evaluate()

def evaluate(
    self,
    action_type: str,
    payload: dict[str, Any] | None = None,
    min_tier: int = 0,
) -> Verdict:
    """Evaluate an action through the Shield pipeline."""

Parameters:

Parameter	Type	Default	Description
action_type	str	required	The action type (e.g., "execute_command", "write_file")
payload	dict	None	Action-specific parameters
min_tier	int	0	Minimum evaluation tier

Returns: Verdict

Shield.status()

def status(self) -> ShieldStatus:
    """Return the current Shield status including Tier 2 budget usage."""

Shield.close()

def close(self) -> None:
    """Shut down the Go process. Called automatically on garbage collection."""

Verdict

@dataclass
class Verdict:
    decision: str        # "ALLOW" or "BLOCK"
    tier: int            # 0, 1, or 2
    confidence: float    # 0.0 - 1.0
    reasoning: str       # Human-readable explanation
    action_hash: str     # SHA-256 hash of the evaluated action
    evaluated_at: str    # ISO 8601 timestamp
    expires_at: str      # ISO 8601 timestamp

ShieldStatus

@dataclass
class ShieldStatus:
    active: bool         # Whether Shield is running
    tier2_enabled: bool  # Whether Tier 2 evaluator is configured
    tier2_used: int      # Tier 2 evaluations used today
    tier2_budget: int    # Tier 2 daily budget

Examples

Basic Evaluation

from openparallax_shield import Shield

shield = Shield(policy_file="security/shield/default.yaml")

# Evaluate a file read.
verdict = shield.evaluate("read_file", {"path": "/home/user/.ssh/id_rsa"})
print(verdict.decision)    # BLOCK
print(verdict.tier)        # 0
print(verdict.reasoning)   # policy deny [block_sensitive_system_paths]: read_file on "/home/user/.ssh/id_rsa" matched a policy pattern

# Evaluate a shell command.
verdict = shield.evaluate("execute_command", {"command": "ls -la"})
print(verdict.decision)    # ALLOW
print(verdict.confidence)  # 0.7

Custom Policy

shield = Shield(
    policy_file="my-policy.yaml",
    heuristic_enabled=True,
    fail_closed=True,
)

# Evaluate workspace writes.
verdict = shield.evaluate("write_file", {
    "path": "/home/user/project/main.py",
    "content": "print('hello')",
})

With LLM Evaluator

shield = Shield(
    policy_file="security/shield/strict.yaml",
    evaluator={
        "provider": "anthropic",
        "model": "claude-sonnet-4-6",
        "api_key_env": "ANTHROPIC_API_KEY",
    },
    canary_token="MY-CANARY-TOKEN-12345",
    daily_budget=50,
)

# This will go through all three tiers.
verdict = shield.evaluate("execute_command", {
    "command": "curl https://api.example.com/data | jq '.secrets'",
})

Batch Evaluation

actions = [
    ("read_file", {"path": "/home/user/workspace/README.md"}),
    ("execute_command", {"command": "go test ./..."}),
    ("write_file", {"path": "/tmp/output.txt", "content": "results"}),
    ("read_file", {"path": "/etc/shadow"}),
    ("send_email", {"to": "user@example.com", "subject": "Report"}),
]

shield = Shield(policy_file="security/shield/default.yaml")

for action_type, payload in actions:
    verdict = shield.evaluate(action_type, payload)
    status = "PASS" if verdict.decision == "ALLOW" else "FAIL"
    print(f"[{status}] {action_type}: {verdict.reasoning}")

Output:

[PASS] read_file: policy allow: allow_workspace_reads
[PASS] execute_command: classifier approved
[PASS] write_file: classifier approved
[FAIL] read_file: policy deny [block_sensitive_system_paths]: read_file on "/home/user/.ssh/id_rsa" matched a policy pattern
[PASS] send_email: classifier approved

Async Usage

import asyncio
from openparallax_shield import AsyncShield

async def main():
    shield = AsyncShield(policy_file="security/shield/default.yaml")

    # Evaluate multiple actions concurrently.
    verdicts = await asyncio.gather(
        shield.evaluate("read_file", {"path": "/home/user/file1.txt"}),
        shield.evaluate("read_file", {"path": "/home/user/file2.txt"}),
        shield.evaluate("execute_command", {"command": "ls"}),
    )

    for v in verdicts:
        print(f"{v.decision}: {v.reasoning}")

    await shield.close()

asyncio.run(main())

Integration with LangChain

from langchain.agents import AgentExecutor
from openparallax_shield import Shield

shield = Shield(policy_file="security/shield/default.yaml")

def guarded_tool_executor(tool_name: str, tool_input: dict) -> str:
    """Wrap tool execution with Shield evaluation."""
    verdict = shield.evaluate(
        action_type=tool_name,
        payload=tool_input,
    )

    if verdict.decision == "BLOCK":
        return f"Action blocked by security policy: {verdict.reasoning}"

    # Execute the actual tool.
    return execute_tool(tool_name, tool_input)

Context Manager

from openparallax_shield import Shield

with Shield(policy_file="security/shield/default.yaml") as shield:
    verdict = shield.evaluate("execute_command", {"command": "whoami"})
    print(verdict.decision)
# Go process is automatically shut down on exit.

Checking Status

shield = Shield(policy_file="security/shield/default.yaml", daily_budget=100)

# After some evaluations...
status = shield.status()
print(f"Tier 2 usage: {status.tier2_used}/{status.tier2_budget}")
print(f"Tier 2 enabled: {status.tier2_enabled}")

Error Handling

from openparallax_shield import Shield, ShieldError, PolicyLoadError

try:
    shield = Shield(policy_file="nonexistent.yaml")
except PolicyLoadError as e:
    print(f"Failed to load policy: {e}")

try:
    verdict = shield.evaluate("read_file", {"path": "/tmp/test"})
except ShieldError as e:
    print(f"Evaluation failed: {e}")

Next Steps

• Node.js wrapper -- use Shield from Node.js
• Go library -- use Shield directly in Go
• Policy Syntax -- write custom policies
• Configuration -- all configuration options