Python SDK reference

pip install promptguard-sdk==2.4.2

Drop-in security for AI applications.
Just change your base URL and add an API key.

Usage (proxy mode - existing):

    from promptguard import PromptGuard

    pg = PromptGuard(api_key="pg_live_xxx")
    response = pg.chat.completions.create(
        model="gpt-5-nano",
        messages=[{"role": "user", "content": "Hello!"}]
    )

Usage (auto-instrumentation - new):

    import promptguard
    promptguard.init(api_key="pg_live_xxx")

    # Now ALL LLM calls are secured, regardless of framework.
    from openai import OpenAI
    client = OpenAI()
    client.chat.completions.create(...)  # scanned by PromptGuard

@dataclass
class Config

``timeout`` defaults to ``30.0`` seconds here (vs ``10.0`` for the
standalone Guard scan path / ``GuardClient`` / ``init()``): the proxy fronts
the full upstream LLM call, which is inherently slower than an isolated
Guard scan, so it needs a longer ceiling.

class GuardApiError(Exception)

Only surfaced when ``fail_open=False``.  When ``fail_open=True``
(the default), callers catch this and return an allow decision.

GuardApiError(message: str, status_code: int | None = None)

class GuardClient

Provides both sync and async methods.  The client itself never
swallows errors; callers (patches, integrations) decide whether
to fail open or closed based on configuration.

``timeout`` defaults to ``10.0`` seconds — the Guard scan is a fast,
standalone call.  (The proxy client / ``Config`` default is ``30.0`` because
it fronts the full upstream LLM call.)  ``api_key`` / ``base_url`` fall back
to ``PROMPTGUARD_API_KEY`` / ``PROMPTGUARD_BASE_URL`` when omitted.

GuardClient(
    api_key: str | None = None,
    base_url: str | None = None,
    timeout: float = 10.0,
)

def scan(
    messages: list[dict[str, str]],
    direction: str = 'input',
    model: str | None = None,
    context: dict[str, Any] | None = None,
) -> GuardDecision

async def scan_async(
    messages: list[dict[str, str]],
    direction: str = 'input',
    model: str | None = None,
    context: dict[str, Any] | None = None,
) -> GuardDecision

def close()

Safe to call from sync code (e.g. ``init()`` re-init and
``shutdown()``): each async client is closed on the loop it belongs
to if that loop is still running, otherwise via a throwaway loop, and
any failure is swallowed.

async def aclose()

class GuardDecision

GuardDecision(data: dict[str, Any])

class PromptGuard

Usage::

    from promptguard import PromptGuard

    pg = PromptGuard(api_key="pg_live_xxx")
    response = pg.chat.completions.create(
        model="gpt-5-nano",
        messages=[{"role": "user", "content": "Hello!"}],
    )
    scan_result = pg.security.scan("Check this content")

PromptGuard(
    api_key: str | None = None,
    base_url: str | None = None,
    config: Config | None = None,
    timeout: float | None = None,
    max_retries: int | None = None,
    retry_delay: float | None = None,
)

def close()

class Chat

class ChatCompletions

def create(
    model: str,
    messages: list[dict[str, str]],
    temperature: float = 1.0,
    max_tokens: int | None = None,
    stream: bool = False,
    **kwargs,
) -> dict[str, Any] | Iterator[dict[str, Any]]

class Security

def scan(content: str, content_type: str = 'prompt') -> 'SecurityScanResult'

def redact(content: str, pii_types: list[str] | None = None) -> dict[str, Any]

class Agent

def validate_tool(
    agent_id: str,
    tool_name: str,
    arguments: dict[str, Any],
    session_id: str | None = None,
) -> dict[str, Any]

def stats(agent_id: str) -> dict[str, Any]

class RedTeam

Every method here used to target ``/internal/redteam``, which requires
platform-admin auth -- so all of them returned 401 for every customer who
called them. The customer-facing plane is ``/api/v1/security-testing``,
reachable with an ordinary API key (``proxy`` scope).

Two methods were removed rather than repointed, because neither has a
customer-facing equivalent:

* ``run_autonomous`` -- the LLM-powered mutation loop. It takes a
  caller-supplied budget and costs real money per run, so it stays
  back-office rather than being exposed to any API key.
* ``intelligence_stats`` -- called ``/internal/redteam/intelligence/stats``,
  which exists on no plane at all. It was a 404 for admins too.

def list_tests() -> dict[str, Any]

def run_test(test_name: str, target_preset: str = 'default') -> dict[str, Any]

def run_all(target_preset: str = 'default') -> dict[str, Any]

def run_custom(prompt: str, target_preset: str = 'default') -> dict[str, Any]

class PromptGuardAsync

Usage::

    async with PromptGuardAsync(api_key="pg_live_xxx") as pg:
        resp = await pg.chat.completions.create(
            model="gpt-5-nano",
            messages=[{"role": "user", "content": "Hi"}],
        )

PromptGuardAsync(
    api_key: str | None = None,
    base_url: str | None = None,
    config: Config | None = None,
    timeout: float | None = None,
    max_retries: int | None = None,
    retry_delay: float | None = None,
)

async def close()

class AsyncChat

class AsyncChatCompletions

async def create(
    model: str,
    messages: list[dict[str, str]],
    temperature: float = 1.0,
    max_tokens: int | None = None,
    stream: bool = False,
    **kwargs,
) -> dict[str, Any] | AsyncIterator[dict[str, Any]]

class AsyncSecurity

async def scan(content: str, content_type: str = 'prompt') -> 'SecurityScanResult'

async def redact(content: str, pii_types: list[str] | None = None) -> dict[str, Any]

class AsyncAgent

async def validate_tool(
    agent_id: str,
    tool_name: str,
    arguments: dict[str, Any],
    session_id: str | None = None,
) -> dict[str, Any]

async def stats(agent_id: str) -> dict[str, Any]

class AsyncRedTeam

async def list_tests() -> dict[str, Any]

async def run_test(test_name: str, target_preset: str = 'default') -> dict[str, Any]

async def run_all(target_preset: str = 'default') -> dict[str, Any]

async def run_custom(prompt: str, target_preset: str = 'default') -> dict[str, Any]

class PromptGuardBlockedError(Exception)

PromptGuardBlockedError(decision: GuardDecision)

class PromptGuardError(Exception)

PromptGuardError(
    message: str,
    code: str,
    status_code: int,
    *,
    error_type: str | None = None,
    upgrade_url: str | None = None,
    current_plan: str | None = None,
    requests_used: int | None = None,
    requests_limit: int | None = None,
)

class SecurityScanResult(TypedDict)

Mirrors the Node SDK's ``SecurityScanResult`` so the documented keys are
discoverable.  Declared ``total=False`` because the API only includes
threat metadata when something is detected.

__version__ = '2.4.2'

def init(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    fail_open: bool = True,
    scan_responses: bool = False,
    timeout: float = 10.0,
) -> None

Prefer calling this **once at application startup**, before spawning worker
threads or issuing LLM calls. Re-initialising at runtime is supported and
thread-safe (the new client is swapped in before the old one is closed), but
an in-flight request already holding the previous client may see its
connection closed underneath it.

Parameters
----------
api_key:
    PromptGuard API key.  Falls back to ``PROMPTGUARD_API_KEY`` env var.
base_url:
    PromptGuard API base URL.  Falls back to ``PROMPTGUARD_BASE_URL``
    env var, then ``https://api.promptguard.co/api/v1``.
mode:
    ``"enforce"`` to block policy violations, ``"monitor"`` to log only.
fail_open:
    If ``True`` (default), allow LLM calls when the Guard API is
    unreachable.  Set to ``False`` to fail-closed.
scan_responses:
    If ``True``, also scan LLM responses through the Guard API with
    ``direction="output"``.  Defaults to ``False`` here (the zero-config
    drop-in) to avoid doubling Guard API round-trips per LLM call; the
    framework callback handlers default this to ``True`` because they
    already receive response events for free.
timeout:
    HTTP timeout (seconds) for Guard API calls.  Defaults to ``10.0``: the
    Guard scan is a fast, standalone call in the request path.  (The proxy
    client's default is ``30.0`` because it fronts the full upstream LLM
    call, which is inherently slower.)

def instrumentation_report() -> dict

A startup warning is only read by someone watching the logs at startup.
This is the same information in a form a customer can assert in their own
CI::

    assert promptguard.instrumentation_report()["detected_unpatched"] == []

which turns "we told you" into "you cannot ship without knowing".

``middleware`` names the request-edge middleware guarding this process
(``"fastapi"`` for :class:`promptguard.integrations.fastapi.PromptGuardMiddleware`)
once it is wrapping an app, which with Starlette is from startup (or the
first request when lifespan is off). It is not a patch, so it never
appears in ``patched``.

def is_active() -> bool

def patched_sdks() -> list[str]

Useful in tests/health checks to assert instrumentation is live, e.g.::

    promptguard.init(api_key=...)
    assert "openai" in promptguard.patched_sdks()

Returns an empty list before ``init()`` or after ``shutdown()``, or when no
supported SDK is importable in the current environment.

def shutdown() -> None

def verify(
    api_key: str | None = None,
    base_url: str | None = None,
    timeout: float | None = None,
    max_retries: int = 1,
    retry_delay: float = 0.25,
) -> dict[str, Any]

Credentials resolve exactly as they do everywhere else — explicit argument,
then ``PROMPTGUARD_API_KEY`` / ``PROMPTGUARD_BASE_URL`` from the
environment.

``max_retries`` defaults to 1 rather than the client's 3, and
``retry_delay`` to 0.25s rather than 1s: this is a diagnostic, so answering
fast beats riding out a blip. On the client's defaults a dead host takes
about seven seconds of backoff to report, which is a long time to stare at a
pre-flight check. Raise them if you would rather tolerate a flaky network
than get a false "unreachable".

Never raises for a failed check: a check that raised would be useless in the
place this is most needed, which is a CI step or a pre-flight script that
wants to report *all* the problems at once.  A missing API key is the one
exception — that is a caller error, not a finding, and it raises
``ValueError`` like the rest of the SDK.

Returns a dict with:

``ok``
    ``True`` when no check failed.  Warnings do not clear it.
``checks``
    One entry per check: ``name``, ``status`` (``pass``/``warn``/``fail``)
    and a human-readable ``detail``.
``checks_passed`` / ``checks_failed`` / ``checks_warned``
    Counts, mirroring the CLI's JSON output.
``instrumentation``
    The full :func:`promptguard.instrumentation_report`.
``base_url``, ``sdk_version``
    What was actually checked, so a report pasted into an issue is
    self-describing.

A **warn** means the request worked but the answer was not what a protected
setup should give — an injection that came back allowed, or a PII sample
with nothing detected.  That is a policy or configuration question rather
than a broken integration, which is why it does not clear ``ok``; it is
still the thing to look at before trusting the setup.

Implements LangChain's ``BaseCallbackHandler`` to scan prompts before
LLM calls and responses after, with rich context about chains, tools,
and agent steps.

Usage::

    from promptguard.integrations.langchain import PromptGuardCallbackHandler

    handler = PromptGuardCallbackHandler(api_key="pg_live_xxx")

    # Attach to a single LLM
    llm = ChatOpenAI(model="gpt-5-nano", callbacks=[handler])

    # Or use globally with any chain / agent
    chain.invoke({"input": "..."}, config={"callbacks": [handler]})

Every scan names its Conversation in ``session_id`` with the call's own thread
id: the ``thread_id`` LangGraph puts in every callback's metadata, or a
``thread_id`` / ``session_id`` / ``conversation_id`` set in
``config["metadata"]``. A call that names no thread sends no session id, unless
the handler was given a ``session_id`` of its own.

LangChainCallbackHandler = PromptGuardCallbackHandler

class PromptGuardCallbackHandler(_BaseCallbackHandler)

A ``langchain_core`` ``BaseCallbackHandler`` when LangChain is installed;
the SDK itself still has no hard dependency on it.

``serialized`` is typed optional on every ``*_start`` callback: runnables,
and so every LangGraph graph and node, pass ``None`` there.

PromptGuardCallbackHandler(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    scan_responses: bool = True,
    fail_open: bool = True,
    timeout: float = 10.0,
    session_id: str | None = None,
)

Parameters
----------
mode:
    ``"enforce"`` to block policy violations, ``"monitor"`` to log only.
session_id:
    The Conversation to name when a call carries no thread id of its
    own -- for a handler that serves one conversation. A call's
    ``thread_id`` (or ``session_id`` / ``conversation_id``) in its
    metadata always wins. Without either, no session id is sent.
scan_responses:
    Defaults to ``True`` here — unlike ``promptguard.init()`` which
    defaults to ``False``. The callback already receives ``on_llm_end``
    events for free, so scanning responses adds no extra call surface
    and richer coverage is the expected behaviour when a user opts into
    a framework integration. ``init()`` stays conservative (off) because
    it is the zero-config drop-in and response scanning doubles the
    Guard API round-trips per LLM call.
timeout:
    HTTP timeout (seconds) for Guard API calls (default 10s, matching
    the Guard scan path; the proxy client uses 30s because it fronts the
    full upstream LLM call).

def on_llm_start(
    serialized: dict[str, Any] | None,
    prompts: list[str],
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None

def on_chat_model_start(
    serialized: dict[str, Any] | None,
    messages: list[list[Any]],
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None

def on_llm_end(
    response: Any,
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> None

def on_llm_error(
    error: BaseException,
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> None

def on_chain_start(
    serialized: dict[str, Any] | None,
    inputs: dict[str, Any],
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None

def on_chain_end(
    outputs: dict[str, Any],
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> None

def on_chain_error(
    error: BaseException,
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> None

def on_tool_start(
    serialized: dict[str, Any] | None,
    input_str: str,
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    tags: list[str] | None = None,
    metadata: dict[str, Any] | None = None,
    **kwargs: Any,
) -> None

def on_tool_end(
    output: Any,
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> None

def on_tool_error(
    error: BaseException,
    *,
    run_id: UUID | None = None,
    parent_run_id: UUID | None = None,
    **kwargs: Any,
) -> None

class PromptGuardRunnable

Unlike :class:`PromptGuardCallbackHandler`, which only *observes* LLM I/O
and therefore can block but never rewrite it, this guard sits **in** the
data flow. That lets it honor a ``redact`` decision by returning the
redacted value, in addition to raising :class:`PromptGuardBlockedError` on a
``block`` decision. Compose it into any LCEL chain with ``|``::

    from promptguard.integrations.langchain import PromptGuardRunnable

    guard = PromptGuardRunnable(api_key="pg_live_xxx").as_runnable()
    chain = guard | prompt | llm            # scan the user input
    chain.invoke("user question")

Use ``direction="output"`` to guard an LLM/parser result at the tail of a
chain::

    output_guard = PromptGuardRunnable(direction="output").as_runnable()
    chain = prompt | llm | StrOutputParser() | output_guard

Redaction is applied in place for ``str``, ``dict`` (the first recognized
text field), and ``list[dict]`` message inputs. For any other value type a
``redact`` decision cannot be safely rewritten, so — mirroring the callback
posture — enforce mode blocks and monitor mode warns and passes through.

PromptGuardRunnable(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    direction: str = 'input',
    fail_open: bool = True,
    timeout: float = 10.0,
    session_id: str | None = None,
)

def guard_value(value: Any, config: dict[str, Any] | None = None) -> Any

This is the plain callable behind :meth:`as_runnable` — usable directly
(e.g. in a ``RunnableLambda`` you build yourself) without importing
LangChain. ``config`` is the call's ``RunnableConfig``, which
``RunnableLambda`` passes in: its ``configurable["thread_id"]`` -- or a
``thread_id`` / ``session_id`` / ``conversation_id`` in its
``metadata`` -- names the Conversation, before the guard's own
``session_id``.

def as_runnable() -> Any

Hooks into LlamaIndex's callback system to scan LLM prompts and responses
with rich pipeline context.

Usage::

    from promptguard.integrations.llamaindex import PromptGuardCallbackHandler
    from llama_index.core.callbacks import CallbackManager

    pg_handler = PromptGuardCallbackHandler(api_key="pg_live_xxx")
    callback_manager = CallbackManager([pg_handler])

    from llama_index.core import Settings
    Settings.callback_manager = callback_manager

LlamaIndex's callback events name no conversation -- each carries only its own
event id -- so no session id is sent unless one is passed: give a chat
engine's handler (or query guard) the ``session_id`` of the conversation it
serves, and every scan it makes names that Conversation.

LlamaIndexCallbackHandler = PromptGuardCallbackHandler

class PromptGuardCallbackHandler

Implements the LlamaIndex ``BaseCallbackHandler`` interface without
importing LlamaIndex at module level.

PromptGuardCallbackHandler(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    scan_responses: bool = True,
    fail_open: bool = True,
    timeout: float = 10.0,
    event_starts_to_ignore: list[str] | None = None,
    event_ends_to_ignore: list[str] | None = None,
    session_id: str | None = None,
)

Parameters
----------
mode:
    ``"enforce"`` to block policy violations, ``"monitor"`` to log only.
session_id:
    The Conversation this handler's scans belong to. LlamaIndex events
    name none, so without it no session id is sent.
scan_responses:
    Defaults to ``True`` here — unlike ``promptguard.init()`` which
    defaults to ``False``. The callback already receives ``on_event_end``
    events for free, so scanning responses adds no extra call surface and
    richer coverage is the expected behaviour when opting into a
    framework integration. ``init()`` stays conservative (off) because it
    is the zero-config drop-in and response scanning doubles the Guard
    API round-trips per LLM call.
timeout:
    HTTP timeout (seconds) for Guard API calls (default 10s, matching the
    Guard scan path; the proxy client uses 30s because it fronts the full
    upstream LLM call).

def on_event_start(
    event_type: str,
    payload: dict[str, Any] | None = None,
    event_id: str = '',
    parent_id: str = '',
    **kwargs: Any,
) -> str

def on_event_end(
    event_type: str,
    payload: dict[str, Any] | None = None,
    event_id: str = '',
    **kwargs: Any,
) -> None

def start_trace(trace_id: str | None = None) -> None

def end_trace(
    trace_id: str | None = None,
    trace_map: dict[str, list[str]] | None = None,
) -> None

class PromptGuardQueryGuard

Unlike :class:`PromptGuardCallbackHandler`, which only *observes* events and
can therefore block but never rewrite them, this guard is a pipeline
component that sits **in** the data flow. It scans the query before
retrieval/synthesis, raising :class:`PromptGuardBlockedError` on a ``block``
decision and returning the redacted query on a ``redact`` decision. Drop it
at the head of a ``QueryPipeline``::

    from promptguard.integrations.llamaindex import PromptGuardQueryGuard
    from llama_index.core.query_pipeline import QueryPipeline

    guard = PromptGuardQueryGuard(api_key="pg_live_xxx")
    pipeline = QueryPipeline(chain=[guard.as_query_component(), retriever])
    pipeline.run(input="user question")

``guard_query`` is also usable directly as a plain preprocessor function
(no LlamaIndex import required) if you are not building a ``QueryPipeline``.

PromptGuardQueryGuard(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    fail_open: bool = True,
    timeout: float = 10.0,
    session_id: str | None = None,
)

def guard_query(query_str: str) -> str

def as_query_component() -> Any

Usage::

    from crewai import Crew
    from promptguard.integrations.crewai import PromptGuardGuardrail

    pg = PromptGuardGuardrail(api_key="pg_live_xxx")

    crew = Crew(
        agents=[...],
        tasks=[...],
        before_kickoff_callbacks=[pg.before_kickoff],
        after_kickoff_callbacks=[pg.after_kickoff],
    )

    # The fields are lists. ``Crew`` ignores unknown keyword arguments, so
    # ``before_kickoff=...`` would be accepted and register nothing.

    # Or wrap individual tools:
    from promptguard.integrations.crewai import secure_tool

    @secure_tool(api_key="pg_live_xxx")
    class MyTool(BaseTool):
        ...

class PromptGuardGuardrail

PromptGuardGuardrail(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    fail_open: bool = True,
    timeout: float = 10.0,
    session_id: str | None = None,
)

def before_kickoff(inputs: dict[str, Any]) -> dict[str, Any]

def after_kickoff(result: Any) -> Any

def scan_task_output(output: str, task_name: str = 'unknown') -> str

def secure_tool(
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    fail_open: bool = True,
) -> Callable

Works with FastAPI, Starlette, or any ASGI app. It imports neither framework.

Usage::

    from fastapi import FastAPI
    from promptguard.integrations.fastapi import PromptGuardMiddleware

    app = FastAPI()
    app.add_middleware(
        PromptGuardMiddleware,
        api_key="pg_live_xxx",          # or PROMPTGUARD_API_KEY
        paths=["/v1/chat"],             # omit to scan every path
        exclude_paths=["/v1/chat/feedback"],
    )

For each request with a body on a selected path, the middleware buffers the
body, pulls out its prompt-shaped fields (see :data:`DEFAULT_PROMPT_FIELDS`),
scans them with one ``direction="input"`` Guard API call, and either answers
before the app runs or replays the original bytes to the app unchanged.

This is the choice for an app that **accepts** prompts but calls the LLM
somewhere auto-instrumentation does not patch. With ``promptguard.init()`` also
on, the same text is scanned twice — once here, once at the provider call.

The scan is the async Guard client's ``scan_async``, and a body larger than
:data:`OFF_LOOP_BODY_BYTES` is decoded and parsed in a worker thread, so
nothing here holds the event loop while a request is inspected.

def extract_scan(body: Any, fields: Sequence[str]) -> _Scan

class PromptGuardMiddleware

Parameters
----------
app:
    The ASGI app to wrap. ``app.add_middleware(PromptGuardMiddleware, ...)``
    passes it for you.
api_key, base_url:
    Fall back to ``PROMPTGUARD_API_KEY`` / ``PROMPTGUARD_BASE_URL``.
mode:
    ``"enforce"`` (default) answers a block with ``403``; ``"monitor"`` logs
    it and lets the request through.
fail_open:
    Governs a Guard API that cannot answer: a transport error, a timeout, a
    5xx or a 429 (and a 401/403, which is also logged at ``ERROR`` once).
    ``True`` (default, ADR-0002) lets the request through unscanned and logs
    a ``WARNING`` with ``extra={"promptguard_event":
    "guard_unavailable_fail_open"}``; ``False`` answers ``503``. A Guard
    ``400``/``413``/``422`` is caused by the request itself and is refused
    in every mode: ``413`` for a size error, ``400`` otherwise.
timeout:
    Guard API timeout in seconds (default ``10.0``).
paths:
    Path prefixes to scan; ``None`` (default) scans every path, and an empty
    list is refused. Matching is on whole segments of the app's route path
    (``scope["path"]`` less ``root_path``, as Starlette routes it):
    ``"/v1/chat"`` covers ``/v1/chat`` and ``/v1/chat/completions`` but not
    ``/v1/chatter``.
exclude_paths:
    Path prefixes never scanned, matched the same way. An exclusion wins
    over ``paths``. An excluded request's body is not even buffered.
methods:
    HTTP methods to scan; ``None`` (default) scans any method that carries
    a body, ``GET`` and ``DELETE`` included.
prompt_fields:
    Top-level fields to scan; see :data:`DEFAULT_PROMPT_FIELDS`.
max_body_bytes:
    The largest body buffered for a scan, raw or decompressed; see
    :data:`DEFAULT_MAX_BODY_BYTES`. Over it the request is answered
    ``413`` and neither scanned nor passed to the app.
session_id:
    A function of the ASGI ``scope`` returning the Conversation the request
    is a turn of, or ``None``. Read it from the authenticated session your
    auth middleware put in the scope (``scope["user"]``,
    ``scope["session"]``, ``scope["state"]``), not from anything the
    client controls. A function that raises is logged and the request is
    scanned without a session id.
session_id_header:
    Shorthand for a ``session_id`` that reads one request header, e.g.
    ``"X-Conversation-Id"``. Name only a header your own auth layer or a
    trusted proxy sets and strips from client requests: a value a browser
    can choose lets a client pick which Conversation its turns land in.
    Not both at once.

A request is not a conversation by itself, so with neither (the default),
or when they name none, no session id is sent. An id over 255 characters
is dropped.

Bodies are read by ``Content-Type``: JSON (in its declared charset),
``text/plain`` (scanned whole, as one user message) and
``application/x-www-form-urlencoded`` (its fields). ``gzip`` and
``deflate`` bodies are decompressed first. ``multipart/form-data`` and
other media are passed on unscanned. In enforce mode a body that cannot
be read — an unknown ``Content-Encoding`` (``415``) or charset (``415``),
undecodable text or invalid JSON (``400``) — is refused; monitor mode logs
it and passes it on.

A redact decision cannot be applied to an arbitrary body, so enforce mode
refuses it with ``403`` and monitor mode logs it. That is the rule ADR-0003
sets for a patched call auto-instrumentation cannot rewrite, applied here
by analogy.

PromptGuardMiddleware(
    app: ASGIApp,
    *,
    api_key: str | None = None,
    base_url: str | None = None,
    mode: str = 'enforce',
    fail_open: bool = True,
    timeout: float = 10.0,
    paths: Iterable[str] | None = None,
    exclude_paths: Iterable[str] | None = None,
    methods: Iterable[str] | None = None,
    prompt_fields: Sequence[str] = DEFAULT_PROMPT_FIELDS,
    max_body_bytes: int = DEFAULT_MAX_BODY_BYTES,
    session_id: Callable[[Scope], str | None] | None = None,
    session_id_header: str | None = None,
)