Execlave Python SDK

Official Python SDK for the Execlave AI Governance Platform. Provides pre-execution policy enforcement, tracing, agent registration, PII scrubbing, kill-switch support, and OpenTelemetry export.

Framework integrations — drop in one callback/processor/helper for LangChain, OpenAI Agents SDK, CrewAI, LlamaIndex, AutoGen, OpenAI Chat Completions, or the Model Context Protocol. See the full docs or get an API key.

Installation

pip install execlave-sdk

With OpenTelemetry support:

pip install execlave-sdk[otel]

Quick Start

The canonical request lifecycle is register → enforce → call LLM → trace. enforce_policy is what blocks bad requests; tracing alone only logs them after the fact.

from execlave import Execlave, PolicyBlockedError, AgentPausedError

exe = Execlave(
    api_key="exe_prod_your_key_here",     # or set EXECLAVE_API_KEY env var
    base_url="https://api.execlave.com",
    environment="production",
)

# Register the agent once on startup (idempotent)
agent = exe.register_agent(
    agent_id="my-assistant",
    name="Customer Support Bot",
    description="Handles tier-1 support queries",
    type="chatbot",
    platform="custom",
    tags=["support", "production"],
)


def answer(question: str) -> str:
    trace = exe.start_trace(agent_id="my-assistant")
    trace.set_input(question)

    try:
        # Pre-execution policy enforcement. Synchronously checks every
        # policy you've configured for this agent. Raises PolicyBlockedError
        # if any policy with enforcement_mode='block' fires.
        exe.enforce_policy(agent_id="my-assistant", input=question)
    except PolicyBlockedError as e:
        trace.finish(status="error", error_type="PolicyBlockedError", error_message=str(e))
        return "Request blocked by security policy."
    except AgentPausedError:
        trace.finish(status="error", error_type="AgentPausedError")
        return "Service temporarily unavailable."

    response = llm.invoke(question)               # your LLM call
    trace.set_output(response).set_model("gpt-4").finish()
    return response


print(answer("How do I reset my password?"))

Agent Registration

Register agents to monitor them from the Execlave dashboard:

agent = exe.register_agent(
    agent_id="order-processor",
    name="Order Processor",
    description="Processes and validates customer orders",
    type="autonomous",        # chatbot | copilot | autonomous | workflow
    platform="custom",        # custom | openai | anthropic | langchain | ...
    tags=["orders", "production"],
    autonomy_level="act_with_approval",  # optional — observe | advise | act_with_approval | autonomous
)

# Check agent status (active / paused / etc.)
print(agent.status)

Tiered autonomy & registry versions (AMP)

autonomy_level (optional) maps the agent onto a tiered-governance template that auto-applies a curated policy bundle. To build a version history in the agent registry, call report_agent_metadata from your deploy pipeline on each release — it records a version snapshot (label, git commit, deploy time):

exe.report_agent_metadata(
    agent_id="order-processor",
    version_label="v2.1.0",
    git_commit="9f3c1ab",
    activate=True,        # mark this the active version (default False)
    notes="rolled out canary → 100%",
)

Both are additive and backwards-compatible — agents that omit them are unaffected.

Policy Enforcement

enforce_policy is a synchronous check against the policies you've configured in the dashboard. Call it before every LLM or tool invocation. Behavior depends on each policy's enforcement_mode:

from execlave import PolicyBlockedError

try:
    result = exe.enforce_policy(
        agent_id="my-assistant",
        input=user_message,
        environment="production",        # optional
        metadata={"user_id": "u123"},    # optional
        estimated_cost=0.02,             # optional — for cost_limit policies
        tools=["search", "email"],       # optional — for access_control policies
    )
    # result["allowed"] is True. Check result.get("warnings") for non-blocking signals.
except PolicyBlockedError as e:
    for v in e.violations:
        print(v["policyType"], v["message"])

Important: A policy must be configured with enforcement_mode = block in the dashboard to actually block. Policies in monitor or warn mode produce warnings on the result but never raise.

Tracing

Decorator

The simplest way to trace function calls:

@exe.trace
def process_order(order_data: dict) -> dict:
    result = llm.invoke(json.dumps(order_data))
    return json.loads(result)

@exe.trace only records the call — it does not run policy enforcement. To block bad inputs, call exe.enforce_policy(...) inside the function body before invoking the LLM.

Context Manager

For more control over trace metadata:

with exe.start_trace(agent_id="my-assistant", session_id="sess_abc") as trace:
    trace.set_input({"question": "What is the refund policy?"})

    result = llm.invoke("What is the refund policy?")

    trace.set_output({"answer": result})
    trace.set_model("gpt-4o")
    trace.set_tokens(input=150, output=320)
    trace.set_cost(0.0045)

Manual Trace

trace = exe.start_trace(agent_id="my-assistant")
trace.set_input(user_query)

try:
    response = llm.invoke(user_query)
    trace.set_output(response)
    trace.finish(status="success")
except Exception as e:
    trace.finish(status="error", error_message=str(e), error_type=type(e).__name__)
    raise

Trace Fields

status values: "success" (default), "error", "timeout". All setter methods are chainable.

Privacy & PII Scrubbing

Built-in client-side PII scrubbing before data leaves your infrastructure:

exe = Execlave(
    api_key="exe_prod_xxx",
    privacy={
        "enabled": True,                       # turn the feature on
        "scrub_fields": ["input", "output"],   # fields to scan
        "hash_pii": True,                      # include short SHA-256 hashes in metadata
    },
)

Detected PII types: email addresses, SSNs, credit card numbers, US phone numbers, IP addresses, API keys.

Client-side Injection Scoring

When enable_injection_scan=True (the default), the SDK runs a regex-based prompt-injection check on the trace's input and annotates the trace with the detected risk level and matched patterns:

exe = Execlave(
    api_key="exe_prod_xxx",
    enable_injection_scan=True,
)

The scan attaches a metadata.injection_scan block to the trace (with risk_level and patterns_matched) so detections show up in the dashboard.

This option does not block LLM calls — it is a tagging/telemetry feature. To actually prevent execution when injection is detected, configure an injection_scan policy with enforcement_mode = block in the dashboard and call exe.enforce_policy(...) before your LLM call. See Policy Enforcement.

Detected patterns include "ignore previous instructions", jailbreak attempts, system-prompt extraction, and other common prefixes.

Kill Switch / Pause Support

Execlave supports remote agent pausing via the dashboard. Once paused, every new trace or enforce call raises AgentPausedError:

from execlave import AgentPausedError

try:
    result = answer("Process this order")
except AgentPausedError:
    return "Service temporarily unavailable — agent paused by admin."

The SDK polls for status changes in the background (configurable interval) and connects via Socket.IO when available for sub-second propagation.

OpenTelemetry Integration

Export Execlave traces as OpenTelemetry spans for unified observability:

from execlave import Execlave

exe = Execlave(
    api_key="exe_prod_xxx",
    mode="otlp",
    otlp_endpoint="http://localhost:4318",   # your OTel collector
)

Requires the otel extra: pip install execlave-sdk[otel].

Configuration

Constructor Options

Environment Variables

Error Handling

from execlave import (
    ExeclaveError,
    ExeclaveAuthError,
    PolicyBlockedError,
    ValidatorDeniedError,
    AgentPausedError,
    EnforcementUnavailableError,
)

try:
    exe.enforce_policy(agent_id="my-assistant", input=user_message)
    # ... LLM call + tracing ...
except ValidatorDeniedError as e:
    # A Custom Validator (BYOV) denied the call. Subclass of PolicyBlockedError,
    # so the broader `except PolicyBlockedError` below would also catch it —
    # list this first only if you need validator-specific handling.
    return "Blocked by a custom validator."
except PolicyBlockedError as e:
    # A block-mode policy fired. e.violations is a list of dicts with policyType, message, severity.
    return "Blocked by security policy."
except AgentPausedError:
    # Agent paused via kill switch.
    return "Service temporarily unavailable."
except EnforcementUnavailableError:
    # Only raised when enforcement_on_outage='fail_closed' AND the API
    # is unreachable for 3+ consecutive attempts (circuit breaker open).
    return "Governance system unavailable."
except ExeclaveAuthError:
    raise  # Misconfigured API key — fail loud.
except ExeclaveError as e:
    print(f"SDK error: {e}")
    raise

Async Trace Buffer

The SDK uses a non-blocking circular buffer (max 10,000 traces) with a background flush thread. Traces are batched and sent to the Execlave API automatically.

# Manual flush (e.g., before shutdown)
exe.flush()

# Graceful shutdown — flushes remaining traces and joins background threads
exe.shutdown()

Development

# Clone the repo
git clone https://github.com/execlave/sdk-python.git
cd execlave/sdk-python

# Install dev dependencies
pip install -e ".[test]"

# Run tests
pytest                    # 130 tests
pytest --cov=execlave     # With coverage

# Type checking
mypy execlave/

Legal

By using this SDK, you agree to the Execlave Terms of Service.

• Privacy Policy
• Acceptable Use Policy
• Responsible AI
• Security

License

MIT — see LICENSE for details.