ShieldPi — Inline Guard + Async Response + Runtime EDR for AI Agents

Connect assessment evidence, runtime detection and explicitly configured response actions: attack → detect → review → respond.

• Watchtower Guard (Tier 1) — inline policy checks. Block prompt injections, redact secret leaks, inject hardened guardrails — synchronously, in your LLM request path.
• Watchtower Commands (Tier 2) — async response actions. Kill agent sessions, quarantine memory, switch system prompts, disable tools — issued by the autonomous SOC or your own automation, delivered via webhooks.
• Runtime Monitor (Tier 0) — opt-in framework instrumentation for LangChain, LangGraph, OpenAI Assistants, and the Anthropic SDK.
• Agent Endpoint Telemetry — optional phone/laptop collector that streams device posture, network, package, and explicit file-watch signals so ShieldPi can correlate agent actions with endpoint behavior.

pip install shieldpi

The SDK runtime supports Python 3.9 and later. Development tooling requires Python 3.10 or later: the dev extra uses pytest 9.0.3 or later. Use a separate development environment when testing an application that runs on Python 3.9.

---

Watchtower Guard — inline (NEW in 0.6)

Wrap each LLM turn in one synchronous call. Latency depends on the deployment and input. The client returns an allow verdict on transport failure by default; choose and handle the failure policy explicitly.

from shieldpi import ShieldPiGuard

guard = ShieldPiGuard(api_key="shpi_live_...")

def chat(user_msg: str, session_id: str) -> str:
    # 1. Pre-flight check on input.
    pre = guard.check(user_input=user_msg, session_id=session_id)
    if pre.action == "block":
        return pre.replacement_text
    if pre.action == "inject":
        user_msg = pre.guardrail_prefix + user_msg

    # 2. Call your LLM normally.
    model_output = call_llm(user_msg)

    # 3. Post-flight check on output (catches leaked credentials/PII).
    post = guard.check(model_output=model_output, session_id=session_id)
    if post.action == "redact":
        return post.replacement_text   # contains [REDACTED:...] markers
    return model_output

The four verdict actions

verdict.action	When	Your handler should
allow	No detection	Send the model output unchanged
block	High-confidence prompt injection (≥0.85)	Return verdict.replacement_text; do NOT call the LLM
redact	Output contains AWS key, JWT, Stripe key, valid Luhn CC, SSN, PEM private key	Return verdict.replacement_text (carries [REDACTED:label] markers)
inject	Medium-confidence input pattern (0.55–0.85)	Prepend verdict.guardrail_prefix to the user message before sending to LLM

Async variant

from shieldpi import AsyncShieldPiGuard

async with AsyncShieldPiGuard(api_key="shpi_live_...") as guard:
    verdict = await guard.check(user_input=msg, session_id=sid)

Tuning

ShieldPiGuard(
    api_key="shpi_live_...",
    timeout_s=0.5,           # default — fails OPEN on timeout (no DoS via slow guard)
    fail_closed=False,        # set True for compliance-grade strict deny
)

Full docs + tuning knobs: https://shieldpi.io/docs/watchtower-guard

---

Watchtower Commands — async (NEW in 0.6)

For actions that don't fit the inline path: kill an agent session after a verdict, clear poisoned memory, swap to a hardened system prompt mid-session.

Commands flow through your existing webhook URL as command.issued events. Your handler executes the action, then ACKs:

from shieldpi import ShieldPiCommands

commands = ShieldPiCommands(api_key="shpi_live_...")

# In your webhook handler:
def handle_shieldpi_webhook(event: dict) -> None:
    if event["event_type"] != "command.issued":
        return
    cmd_id = event["data"]["command_id"]
    action = event["data"]["action"]

    try:
        if action == "kill_session":
            my_app.terminate(event["data"]["session_id"])
        elif action == "quarantine_memory":
            my_app.snapshot_and_clear_memory(event["data"]["session_id"])
        elif action == "switch_system_prompt":
            my_app.set_system_prompt(event["data"]["session_id"], event["data"]["payload"]["prompt"])
        elif action == "disable_tool":
            my_app.disable_tool(event["data"]["payload"]["tool_name"])

        commands.ack(cmd_id, status="executed", reason="completed")
    except Exception as exc:
        commands.ack(cmd_id, status="failed", reason=str(exc))

Issue commands from your own automation

commands.issue(
    action="kill_session",
    session_id="user-abc-123",
    payload={"reason": "anomalous_behavior_detected"},
    ttl_seconds=600,
)

Full docs: https://shieldpi.io/docs/watchtower-commands · https://shieldpi.io/dashboard/auto-respond

---

Runtime Monitor — framework instrumentation (since 0.2)

Supported hooks emit LLM and tool events after instrumentation is enabled. Coverage depends on the framework and version; file and network visibility requires the corresponding explicit instrumentation or endpoint collector.

pip install "shieldpi[all]"
export SHIELDPI_SDK_KEY=shpi_live_...   # https://shieldpi.io/dashboard

# Add one import at the top of your agent process:
import shieldpi.auto

# Use LangChain / LangGraph / OpenAI Assistants / Anthropic SDK normally.
from langchain.agents import AgentExecutor
agent = AgentExecutor(...)
agent.invoke({"input": "..."})   # supported hooks emit monitoring events

Open https://shieldpi.io/dashboard/monitoring to see the live event stream and any alerts the detectors fire.

Supported frameworks

Framework	Auto-patch	Manual handler	Optional dep
LangChain (AgentExecutor)	✅	✅	pip install shieldpi[langchain]
LangChain (LCEL tools)	✅ (BaseTool hook)	✅	pip install shieldpi[langchain]
LangGraph	✅	✅	pip install shieldpi[langgraph]
OpenAI Assistants API	✅	—	pip install shieldpi[openai]
OpenAI Chat Completions w/ tools	✅	—	pip install shieldpi[openai]
Anthropic SDK (tool use)	✅	✅	pip install shieldpi[anthropic]
Custom agents (no framework)	—	✅	base install

Manual integration

from shieldpi import Monitor

monitor = Monitor(sdk_key="shpi_live_...")
with monitor.start_session(
    agent_name="invoice-bot",
    stated_goal="help users file invoices",
) as session:
    session.log_user_message("How do I file a Q1 invoice?")
    session.log_tool_call("search_docs", {"query": "Q1 invoice filing"})
    session.log_tool_result("search_docs", {"results": [...]})
    session.log_final_response("Here's how to file a Q1 invoice...")

---

Environment variables

Variable	Purpose
SHIELDPI_API_KEY	Customer API key for Guard + Commands (starts with shpi_live_).
SHIELDPI_SDK_KEY	Per-target SDK key for Runtime Monitor (starts with shpi_live_).
SHIELDPI_BASE_URL	Override the API base URL (defaults to production).
SHIELDPI_AGENT_NAME	Logical name for this agent (Monitor only).
SHIELDPI_AGENT_GOAL	Stated goal (helps detectors flag off-goal behavior).
SHIELDPI_AUTO_INSTRUMENT	Comma-separated framework allowlist. Default: all installed.
SHIELDPI_DEVICE_ID	Stable endpoint collector id. Auto-generated if omitted.
SHIELDPI_DEVICE_NAME	Human-readable endpoint name. Defaults to hostname.
SHIELDPI_WATCH_PATHS	Optional :-separated files/directories to watch for writes. No default filesystem watch.

---

Agent Endpoint Telemetry — phone/laptop collector

Run a conservative endpoint collector alongside a monitored agent:

export SHIELDPI_SDK_KEY=shpi_live_...
shieldpi-endpoint-collector --once

On Android/Termux or Linux-like hosts, the collector registers the device and sends:

• device_posture — OS, Python, hostname, ADB/root posture where available.
• network_connection — new established public IPv4 TCP connections from /proc/net/tcp.
• package_install — new third-party Android packages when pm list packages -3 is available.
• file_write — only for paths explicitly configured with --watch-path or SHIELDPI_WATCH_PATHS; file contents are never read or uploaded.

shieldpi-endpoint-collector \
  --watch-path /sdcard/Documents/ShieldPiLab \
  --interval 10

These events feed the backend correlation engine, which links device telemetry to nearby agent events such as read_file, http_request, send_email, or upload_file.

---

Failure handling and data boundaries

• Inline Guard fails open by default. Transport failures return an allow verdict with a diagnostic reason such as shieldpi_timeout. fail_closed=True raises instead; the calling application must handle that error and enforce its denial policy.
• Monitor handles delivery failures. Inspect delivery diagnostics and configure retry/queue behavior; a successful application request does not prove its telemetry reached ShieldPi.
• All Monitor patches are idempotent — double-import won't double-wrap.
• Guard decision logs omit prompt/output content. They retain SHA-256 input/output hashes and decision metadata, with no plaintext match snippets. Assessment and runtime event evidence have separate storage and retention rules.
• Commands are opt-in. Watchtower auto-respond defaults to off for every customer. Suggest mode requires human approval per command. Auto mode requires per-action explicit opt-in.

---

What's new in 0.6

• ShieldPiGuard — inline guard client with sync + async variants, fail-open by default.
• GuardVerdict — structured response with .blocked, .redacted, .needs_inject helpers.
• ShieldPiCommands — async response action client. .issue, .ack, .get, .list, .supported_actions.
• Measure engine and end-to-end latency on the deployed rules, inputs and hardware; this package does not establish a universal performance guarantee.
• Output-side scanner detects AWS / GCP / Anthropic / OpenAI / Stripe / GitHub credentials, JWTs, PEM private keys, Luhn-validated credit cards, SSN-shape strings, with severity-tiered redaction.

What's new in 0.7

• Agent Endpoint Telemetry — shieldpi-endpoint-collector CLI plus EndpointTelemetryClient.
• Conservative phone/laptop collector for posture, network, package, and explicit file-watch signals.
• Designed to correlate endpoint behavior with live agent events in ShieldPi Watchtower.

Links

• Inline Guard docs — https://shieldpi.io/docs/watchtower-guard
• SIEM integration formats (OCSF / CEF / Syslog) — https://shieldpi.io/docs/siem-formats
• Webhooks — https://shieldpi.io/docs/webhooks
• Dashboard — https://shieldpi.io/dashboard
• Methodology (open) — https://api.shieldpi.io/api/intelligence/methodology
• Support — support@shieldpi.io

License

Apache-2.0

Saved-target CI scanning

The CI CLI is implemented in this canonical checkout. Install this reviewed source with python -m pip install ./sdks/shieldpi-python from the repository root; public-package availability is a separate release step. No dotenv file is loaded.

export SHIELDPI_API_KEY='your-customer-api-key'
export SHIELDPI_TARGET_ID='existing-configured-target-uuid'
shieldpi ci --target-id "$SHIELDPI_TARGET_ID" --fail-severity high --timeout 600 --output shieldpi-result.json

--min-security-score 70 adds a score gate distinct from the severity gate. Exit 0 means passed, 1 means failed, and 2 means an operational/contract error. Partial or unverified assessments fail. The deadline stops waiting, not the server scan. Reuse the returned scan ID/history before retrying an uncertain launch. Set SHIELDPI_BASE_URL for another deployment (origin or origin/api); redirects and environment proxies are disabled.

import os
from shieldpi.ci import CIClient

with CIClient(os.environ["SHIELDPI_API_KEY"]) as client:
    result = client.run(os.environ["SHIELDPI_TARGET_ID"], fail_threshold="high", timeout=600)
    if not result["passed"]:
        raise SystemExit(1)

Targets must already exist, belong to the caller, and satisfy account/plan requirements. Model targets require saved BYOK. For agent targets supply on_agent_session(session) to connect your agent using the returned session_id, token and chat_url while polling continues. CLI users provide --agent-session-file /new/private/path.json and an agent process that reads it. The file is exclusive-created with 0600 permissions; it contains a control token. Never upload it or include it in logs. Clean it up after the agent session ends. The result JSON contains no handshake token.

CI migration from the retired prototype

The sdk/python 0.1 distribution is retired and its installer refuses to overwrite the canonical package. Use shieldpi.ci.CIClient; the runtime Monitor/Guard/Commands imports remain available. The old URL/model inference and numeric fail_threshold must become a configured target ID and the separate min_security_score field. The CLI's --fail-threshold alias accepts severity strings only.

The backend accepts severity fail_threshold and optional integer min_security_score. Poll /api/ci/scan/{id}/result: a successful HTTP response is not a security pass. block=true and wait_timeout_seconds are rejected instead of ignored; block=false remains accepted for compatibility. Use a client deadline. The same rule applies to the action in sdk/github-action, whose source installation uses the canonical package from its own reviewed revision.

Assessment profiles and finding review

The reviewed source also provides shieldpi.assessments.Assessments. Configure and authorize the saved target first; secrets stay on that target, not in profile JSON. Profile versions are immutable and tenant/target-bound. Business policies record context; they do not create a custom policy grader or authorize tool use.

import os
from shieldpi.assessments import Assessments

with Assessments(os.environ["SHIELDPI_API_KEY"]) as assessments:
    versions = assessments.profile_versions(os.environ["SHIELDPI_TARGET_ID"])
    preview = assessments.preview(os.environ["SHIELDPI_TARGET_ID"])
    # Structural only. An actual provider request requires explicit True and may cost money.
    check = assessments.connection_check(os.environ["SHIELDPI_TARGET_ID"])

create_profile_version(target_id, name, definition, previous_version_id=None) creates a version; the definition schema is documented in /docs/api-reference and the dashboard profile editor. Use --profile-version-id VERSION_UUID in CI (or action input profile-version-id). Explicit controls that conflict with a profile are rejected. The processing preview makes no network request and does not imply end-to-end local processing or a regional processing guarantee.

review_history(finding_id) returns the original assessment context and ordered review events. append_review(finding_id, decision) accepts disposition, owner, reason, optional expiration and retest scan UUID. Risk acceptance requires a future timezone-aware expiry. verified_fixed requires a linked, compatible, complete retest with exact replay identity and authoritative safe evidence. An absent finding, partial scan or legacy missing provenance does not establish that a vulnerability is fixed. Reviews do not overwrite original scan evidence.