Prompt Injection Detection with Python

Integrate prompt injection detection in Python by calling IdenticAPI's Prompt Injection Shield from your backend — FastAPI, Django, Flask, or Celery workers — before forwarding user text, RAG chunks, or fetched web content to an LLM. Keep API keys on the server; never embed them in client applications or notebooks committed to source control.

This guide provides a typed client, exception handling, verdict policies, and FastAPI middleware-style integration.

API contract

POST /api/v1/security/prompt-injection
Authorization: Bearer idapi_test_your_key_here
Content-Type: application/json

{"text": "your input", "source": "optional", "context": "optional"}

Response shape:

{
  "request_id": "req_py_001",
  "api": "prompt-injection-shield",
  "verdict": "unsafe",
  "risk": "high",
  "findings": [
    {"category": "instruction_override", "reason": "Attempt to override prior instructions detected"}
  ],
  "reasons": ["Attempt to override prior instructions detected"],
  "usage_units": 1
}

Verdicts: safe, suspicious, unsafe. Documentation: Prompt Injection Shield. Product: Prompt Injection Shield.

Install dependencies

pip install httpx pydantic

httpx for HTTP/2-ready async calls; pydantic for response validation.

Client module

identicapi/prompt_injection.py:

from __future__ import annotations

import os
from enum import Enum
from typing import Any, Optional

import httpx
from pydantic import BaseModel, Field


class InjectionVerdict(str, Enum):
    SAFE = "safe"
    SUSPICIOUS = "suspicious"
    UNSAFE = "unsafe"


class InjectionFinding(BaseModel):
    category: str
    reason: str
    confidence: Optional[float] = None
    start: Optional[int] = None
    end: Optional[int] = None


class PromptInjectionResponse(BaseModel):
    request_id: str
    api: str
    verdict: InjectionVerdict
    risk: str
    findings: list[InjectionFinding] = Field(default_factory=list)
    reasons: list[str] = Field(default_factory=list)
    usage_units: int
    processing_time_ms: Optional[int] = None
    detector_version: Optional[str] = None


class PromptInjectionApiError(Exception):
    def __init__(self, message: str, status_code: int, code: Optional[str] = None):
        super().__init__(message)
        self.status_code = status_code
        self.code = code


class GuardAction(str, Enum):
    ALLOW = "allow"
    REVIEW = "review"
    BLOCK = "block"


def verdict_to_action(verdict: InjectionVerdict) -> GuardAction:
    if verdict == InjectionVerdict.UNSAFE:
        return GuardAction.BLOCK
    if verdict == InjectionVerdict.SUSPICIOUS:
        return GuardAction.REVIEW
    return GuardAction.ALLOW


class PromptInjectionClient:
    def __init__(
        self,
        api_key: Optional[str] = None,
        base_url: Optional[str] = None,
        timeout: float = 10.0,
    ):
        self.api_key = api_key or os.environ.get("IDENTICAPI_API_KEY")
        if not self.api_key:
            raise PromptInjectionApiError(
                "IDENTICAPI_API_KEY is not configured", status_code=500, code="missing_api_key"
            )
        self.base_url = (base_url or os.environ.get("IDENTICAPI_BASE_URL") or "https://www.identicapi.com").rstrip("/")
        self.timeout = timeout

    def screen(
        self,
        text: str,
        *,
        source: Optional[str] = None,
        context: Optional[str] = None,
    ) -> PromptInjectionResponse:
        if not text or len(text) > 32_000:
            raise PromptInjectionApiError(
                "text must be between 1 and 32,000 characters",
                status_code=400,
                code="invalid_text_length",
            )

        payload: dict[str, Any] = {"text": text}
        if source:
            payload["source"] = source
        if context:
            payload["context"] = context

        url = f"{self.base_url}/api/v1/security/prompt-injection"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        try:
            with httpx.Client(timeout=self.timeout) as client:
                response = client.post(url, json=payload, headers=headers)
        except httpx.RequestError as exc:
            raise PromptInjectionApiError(
                f"Network request failed: {exc}", status_code=503, code="network_error"
            ) from exc

        try:
            data = response.json()
        except ValueError as exc:
            raise PromptInjectionApiError(
                "Invalid JSON response from API",
                status_code=response.status_code,
                code="invalid_json",
            ) from exc

        if response.status_code >= 400:
            error = data.get("error", {}) if isinstance(data, dict) else {}
            raise PromptInjectionApiError(
                error.get("message", f"API returned {response.status_code}"),
                status_code=response.status_code,
                code=error.get("code"),
            )

        return PromptInjectionResponse.model_validate(data)

Async variant for FastAPI

Add to the same module:

    async def screen_async(
        self,
        text: str,
        *,
        source: Optional[str] = None,
        context: Optional[str] = None,
    ) -> PromptInjectionResponse:
        if not text or len(text) > 32_000:
            raise PromptInjectionApiError(
                "text must be between 1 and 32,000 characters",
                status_code=400,
                code="invalid_text_length",
            )

        payload: dict[str, Any] = {"text": text}
        if source:
            payload["source"] = source
        if context:
            payload["context"] = context

        url = f"{self.base_url}/api/v1/security/prompt-injection"
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json",
        }

        try:
            async with httpx.AsyncClient(timeout=self.timeout) as client:
                response = await client.post(url, json=payload, headers=headers)
        except httpx.RequestError as exc:
            raise PromptInjectionApiError(
                f"Network request failed: {exc}", status_code=503, code="network_error"
            ) from exc

        try:
            data = response.json()
        except ValueError as exc:
            raise PromptInjectionApiError(
                "Invalid JSON response from API",
                status_code=response.status_code,
                code="invalid_json",
            ) from exc

        if response.status_code >= 400:
            error = data.get("error", {}) if isinstance(data, dict) else {}
            raise PromptInjectionApiError(
                error.get("message", f"API returned {response.status_code}"),
                status_code=response.status_code,
                code=error.get("code"),
            )

        return PromptInjectionResponse.model_validate(data)

FastAPI route example

main.py:

import logging
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, Field

from identicapi.prompt_injection import (
    PromptInjectionClient,
    PromptInjectionApiError,
    verdict_to_action,
    GuardAction,
)

logger = logging.getLogger(__name__)
app = FastAPI()
client = PromptInjectionClient()


class ChatRequest(BaseModel):
    message: str = Field(min_length=1, max_length=8000)


class ChatResponse(BaseModel):
    reply: str
    request_id: str


@app.post("/chat", response_model=ChatResponse)
async def chat(body: ChatRequest):
    try:
        screen = await client.screen_async(
            body.message,
            source="chat_input",
            context="route=/chat",
        )
    except PromptInjectionApiError as exc:
        logger.error(
            "injection_screen_failed",
            extra={"status": exc.status_code, "code": exc.code},
        )
        raise HTTPException(
            status_code=503,
            detail="Security screening temporarily unavailable",
        ) from exc

    action = verdict_to_action(screen.verdict)
    logger.info(
        "prompt_injection_screen",
        extra={
            "request_id": screen.request_id,
            "verdict": screen.verdict.value,
            "risk": screen.risk,
            "categories": [f.category for f in screen.findings],
            "action": action.value,
        },
    )

    if action == GuardAction.BLOCK:
        raise HTTPException(
            status_code=400,
            detail={"error": "Message blocked", "request_id": screen.request_id},
        )

    if action == GuardAction.REVIEW:
        raise HTTPException(
            status_code=422,
            detail={"error": "Message flagged for review", "request_id": screen.request_id},
        )

    reply = await call_llm(body.message)
    return ChatResponse(reply=reply, request_id=screen.request_id)


async def call_llm(message: str) -> str:
    return f"Echo: {message[:100]}"

RAG chunk filtering

async def filter_chunks(chunks: list[dict]) -> list[dict]:
    safe = []
    for chunk in chunks:
        result = await client.screen_async(
            chunk["text"],
            source="rag_chunk",
            context=f"chunk_id={chunk['id']}",
        )
        if verdict_to_action(result.verdict) == GuardAction.ALLOW:
            safe.append(chunk)
        else:
            logger.warning(
                "rag_chunk_blocked",
                extra={
                    "chunk_id": chunk["id"],
                    "request_id": result.request_id,
                    "verdict": result.verdict.value,
                },
            )
    return safe

Environment configuration

export IDENTICAPI_API_KEY=idapi_test_your_key_here
export IDENTICAPI_BASE_URL=https://www.identicapi.com

Load via .env in development with python-dotenv — add .env to .gitignore.

Pytest example

tests/test_prompt_injection_client.py:

import pytest
from identicapi.prompt_injection import verdict_to_action, InjectionVerdict, GuardAction


def test_verdict_to_action_unsafe():
    assert verdict_to_action(InjectionVerdict.UNSAFE) == GuardAction.BLOCK


def test_verdict_to_action_safe():
    assert verdict_to_action(InjectionVerdict.SAFE) == GuardAction.ALLOW

Add integration tests with fixtures from Prompt Injection Testing.

Django view sketch

from django.http import JsonResponse
from identicapi.prompt_injection import PromptInjectionClient, verdict_to_action, GuardAction

client = PromptInjectionClient()

def chat_view(request):
    message = request.POST.get("message", "").strip()
    if not message:
        return JsonResponse({"error": "message required"}, status=400)

    try:
        screen = client.screen(message, source="chat_input")
    except Exception:
        return JsonResponse({"error": "screening unavailable"}, status=503)

    if verdict_to_action(screen.verdict) != GuardAction.ALLOW:
        return JsonResponse(
            {"error": "blocked", "request_id": screen.request_id},
            status=400,
        )

    return JsonResponse({"reply": "...", "request_id": screen.request_id})

Related guides

• TypeScript integration
• Detection strategy
• Prevention architecture
• Manual testing: Prompt Injection Checker

Limitations

• Sync vs async — do not block ASGI workers with sync screen() in hot paths
• Large documents — split text over 32,000 characters
• Rate limits — batch chunk screening may need throttling (rate limits in docs)
• Fail-closed on outage — impacts availability; document tradeoff
• Detection scope — addresses injection patterns, not all misuse (keyword limits)

IdenticAPI returns structured risk signals; your application decides block vs review vs allow.

Practical checklist

• Wrap API calls in PromptInjectionClient with explicit exceptions
• Validate responses with Pydantic models
• Map verdicts to GuardAction consistently across routes
• Log request_id and finding categories, not user secrets
• Screen RAG chunks in retrieval pipeline
• Store API key in environment/secrets manager only
• Add pytest fixtures for expected verdicts on known payloads
• Complete security checklist before launch

Python backends integrate Prompt Injection Shield with a thin HTTP client and strict verdict policy — place it on every path where untrusted text becomes LLM context.