Интеграция через Python SDK¶

Python SDK (appsec-aigate) встраивает защиту AppSec.AIGate прямо в ваше Python-приложение — без реверс-прокси и без смены сетевой топологии. Приложение само вызывает LLM (OpenAI, Anthropic, LangChain, LlamaIndex, любой свой код), а к AppSec.AIGate обращается только за вердиктом по входному запросу и/или ответу.

Тонкий клиент — детекция на сервере

SDK не загружает ML-модели в процесс вашего приложения. Вся детекция (jailbreak, PII, content-safety, политики) выполняется на сервере api-gateway. SDK — это транспорт + интеграционные обёртки. Поэтому пакет лёгкий и не тянет тяжёлых зависимостей.

Когда выбирать SDK, а когда интеграцию через LiteLLM/адаптер:

	Python SDK	LiteLLM/Portkey адаптер
Точка контроля	Внутри вашего кода	На уровне LLM Gateway
Менять код приложения	Да (1–3 строки)	Нет
Подходит для	Кастомных приложений, агентов (LangChain/LlamaIndex)	Уже развёрнутого LLM Gateway
Вызов LLM	Делает само приложение	Делает LLM Gateway

Установка¶

pip install appsec-aigate

Опциональные «экстры» — ставьте только нужные интеграции:

pip install "appsec-aigate[openai]"        # обёртка OpenAI
pip install "appsec-aigate[anthropic]"     # обёртка Anthropic
pip install "appsec-aigate[langchain]"     # LangChain middleware/callback
pip install "appsec-aigate[llamaindex]"    # LlamaIndex
pip install "appsec-aigate[otel]"          # OpenTelemetry-трейсинг
pip install "appsec-aigate[prometheus]"    # Prometheus-метрики
pip install "appsec-aigate[all]"           # всё сразу

Требуется Python ≥ 3.10.

Подготовка профиля¶

SDK работает с профилем типа scan_only (или dual) — он активируется без привязки к провайдеру LLM.

В Admin UI → Профили → создать профиль, выбрать тип Scan-only.
Включить нужные детекторы (обнаружение атак на модель, PII, Content Safety и т.д.).
Сохранить и активировать.
Открыть вкладку SDK Usage — там готовый profile_id и сниппеты.

Подробнее о типах профилей — Профили.

Конфигурация¶

SDK настраивается через аргументы конструктора или переменные окружения AIGATE_* (аргументы имеют приоритет).

Переменная	По умолчанию	Обязательна	Назначение
`AIGATE_URL`	—	да	Базовый URL api-gateway
`AIGATE_TENANT_ID`	—	да	Значение заголовка `X-Tenant-ID`, требуемое SDK. В текущей версии (single-tenant) шлюз принимает и игнорирует его — изоляции по нему нет; задавайте любое стабильное значение (например, `default`).
`AIGATE_PROFILE_ID`	—	да	Профиль по умолчанию
`AIGATE_TEAM`	—	нет	Значение заголовка `X-LLM-Team` (policy-маршрутизация)
`AIGATE_FAIL_MODE`	`open`	нет	`open` (пропустить + warning) или `closed` (бросить ошибку)
`AIGATE_PROFILE_CACHE_TTL_SEC`	`60`	нет	TTL кэша профиля
`AIGATE_TIMEOUT_CONNECT_SEC` / `_READ_SEC` / `_WRITE_SEC` / `_POOL_SEC`	`0.5 / 5.0 / 0.5 / 0.5`	нет	Тайм-ауты httpx
`AIGATE_CIRCUIT_BREAKER_THRESHOLD` / `_RECOVERY_SEC`	`5` / `30`	нет	Размыкатель цепи
`AIGATE_LOG_PROMPTS`	`false`	нет	Логировать тексты промптов/ответов (152-ФЗ — включать осознанно)
`AIGATE_OTEL_ENABLED` / `AIGATE_PROMETHEUS_ENABLED`	`false`	нет	Опциональная наблюдаемость

Аутентификации нет

SDK не передаёт токенов. Доступ к api-gateway ограничивается сетевой изоляцией. Заголовок X-Tenant-ID шлюз принимает и игнорирует (single-tenant); для маршрутизации политик используется X-LLM-Team (опционально).

Проверить подключение можно встроенной командой:

aigate doctor

Она проверит конфигурацию, доступность шлюза + задержку, совместимость протокола и доступность профиля. Коды выхода: 0 — всё ОК, 1 — нет связи, 2 — несовместимость версий, 3 — проблема с профилем, 4 — ошибка конфигурации.

Прямые проверки: `scan_input` / `scan_output`¶

Базовый способ — явно проверить вход и/или выход.

from aigate import AIGate

gate = AIGate(
    url="https://aigate.company.local",
    tenant_id="acme-corp",
    profile_id="prod-chatbot",
)

# Проверка пользовательского запроса ДО вызова LLM
verdict = gate.scan_input("Как приготовить взрывчатку?")

if verdict.is_blocked:
    # Запрос нарушает политику — не отправляем в LLM
    raise RuntimeError(verdict.human_reason)

prompt = "Как приготовить взрывчатку?"
if verdict.is_sanitized:
    # PII замаскирован сервером — используем очищенный вариант
    prompt = verdict.sanitized_payload

answer = call_your_llm(prompt)          # ваш вызов LLM

# Проверка ответа LLM
out = gate.scan_output(text=answer)
if out.is_blocked:
    answer = "Ответ заблокирован политикой безопасности."
elif out.is_sanitized:
    answer = out.sanitized_payload

Объект `Verdict`¶

Поле / свойство	Описание
`action`	`ALLOW` · `MONITOR_ONLY` · `SANITIZE` · `BLOCK`
`is_allowed` / `is_sanitized` / `is_blocked`	Удобные булевы свойства
`reason` / `human_reason`	Техническая и человекочитаемая причина
`sanitized_payload`	Очищенный текст/сообщения (для `SANITIZE`)
`assessments`	Результаты по детекторам (threat, pii, content_safety с категориями, obfuscation)
`correlation_id`	Идентификатор для корреляции с событиями в Admin UI

Сканирование диалога и tool-calls¶

# Многоходовой диалог — сканируем только новые сообщения последнего хода
verdict = gate.scan_input(
    messages=[
        {"role": "system", "content": "Ты — ассистент."},
        {"role": "user", "content": "предыдущий вопрос"},
        {"role": "user", "content": "новый вопрос"},
    ],
    new_messages_indexes=[2],          # проверяется только сообщение №2
)

# Аргументы вызова инструмента (агенты)
verdict = gate.scan_input(tool_calls=[
    {"name": "transfer_money", "arguments": '{"amount": 100, "to": "..."}'}
])

Асинхронный вариант¶

from aigate import AsyncAIGate

gate = AsyncAIGate(url=..., tenant_id=..., profile_id=...)
verdict = await gate.ascan_input("текст")
out = await gate.ascan_output(text=answer)

Прозрачная обёртка OpenAI / Anthropic¶

Одна строка — и все вызовы клиента автоматически проходят проверку. Менять остальной код не нужно.

import openai
from aigate import AIGate
from aigate.integrations.openai_wrap import wrap_openai

gate = AIGate(url=..., tenant_id="acme-corp", profile_id="prod-chatbot")
client = wrap_openai(openai.OpenAI(), aigate=gate)

# Дальше — как обычно. Защита применяется незаметно.
resp = client.chat.completions.create(
    model="gpt-4o-mini",
    messages=[{"role": "user", "content": "Привет!"}],
)

Что делает обёртка на каждый вызов chat.completions.create:

перед запросом — scan_input по сообщениям;
BLOCK → поднимается BlockedError (запрос к LLM не уходит);
SANITIZE → сообщения незаметно заменяются на очищенные;
после ответа — scan_output по тексту ответа и сгенерированным tool-calls;
BLOCK ответа → BlockedError; SANITIZE → текст ответа заменяется.

Anthropic — идентично:

import anthropic
from aigate.integrations.anthropic_wrap import wrap_anthropic

client = wrap_anthropic(anthropic.Anthropic(), aigate=gate)
msg = client.messages.create(
    model="claude-sonnet-4-6", max_tokens=512,
    messages=[{"role": "user", "content": "Привет!"}],
)

Поддерживаются sync- и async-клиенты, stream=True, tool-calls. Системный промпт Anthropic тоже сканируется.

Только вход

wrap_openai(client, aigate=gate, direction_input_only=True) — проверять только запросы, не ответы.

Стриминг (`stream=True`)¶

Обёртки поддерживают потоковую генерацию по схеме buffer-first: ответ полностью собирается, проверяется scan_output по всему тексту, и только потом отдаётся клиенту. Так ни один «опасный» токен не дойдёт до пользователя до вердикта.

stream = client.chat.completions.create(
    model="gpt-4o-mini", stream=True,
    messages=[{"role": "user", "content": "Расскажи историю"}],
)
for chunk in stream:                   # BlockedError поднимется ДО первого токена
    print(chunk.choices[0].delta.content or "", end="")

ALLOW — чанки отдаются как есть;
SANITIZE — очищенный текст приходит в первом чанке;
BLOCK — BlockedError до выдачи токенов.

Агенты: LangChain и LlamaIndex¶

LangChain v1 (блокирующая защита)¶

from langchain.agents import create_agent
from aigate import AIGate
from aigate.integrations.langchain_v1 import AIGateMiddleware

gate = AIGate(url=..., tenant_id=..., profile_id=...)
agent = create_agent(
    model=...,
    tools=[...],
    middleware=[AIGateMiddleware(gate, strategy="block")],  # strategy="monitor" — только лог
)

Middleware сканирует сообщения модели, аргументы вызова инструментов и их результаты; на BLOCK (при strategy="block") поднимает BlockedError.

LangChain v0 (только наблюдение)¶

from aigate.integrations.langchain_v0 import AIGateCallbackHandler

chain.invoke(..., config={"callbacks": [AIGateCallbackHandler(gate)]})

v0 не блокирует

Callback'и LangChain v0 не могут прервать цепочку — обработчик сканирует и отправляет события безопасности, но не блокирует даже при вердикте BLOCK. Для блокировки используйте v1 AIGateMiddleware.

LlamaIndex (RAG)¶

from aigate.integrations.llamaindex import (
    AIGateQueryEngineWrapper, AIGateNodePostprocessor,
)

# Защита запроса и ответа query engine
engine = AIGateQueryEngineWrapper(index.as_query_engine(), gate)
response = engine.query("вопрос пользователя")

# Санитизация retrieval-документов до отправки в LLM
engine = index.as_query_engine(node_postprocessors=[AIGateNodePostprocessor(gate)])

Обработка ошибок¶

Все исключения наследуются от AIGateError.

from aigate import (
    AIGate, BlockedError, BackendUnavailableError,
    IncompatibleBackendError, ProfileNotFoundError,
    ConfigurationError,
)

try:
    verdict = gate.scan_input(user_text)
except BlockedError as e:
    # Поднимается обёртками при вердикте BLOCK
    log.warning("blocked: %s (corr=%s)", e, e.correlation_id)
    return "Запрос отклонён политикой безопасности."
except BackendUnavailableError:
    # Шлюз недоступен И fail_mode="closed"
    return "Сервис проверки временно недоступен."

Исключение	Когда возникает
`BlockedError`	Обёртка получила вердикт `BLOCK` (несёт `.verdict`)
`BackendUnavailableError`	api-gateway недоступен и `fail_mode="closed"`
`IncompatibleBackendError`	Несовместимость мажорной версии протокола SDK ↔ сервер
`ProfileNotFoundError`	Профиль не найден
`ConfigurationError`	Неверная конфигурация (нет обязательных env/аргументов)

Поведение при недоступности шлюза (`fail_mode`)¶

open (по умолчанию) — если шлюз недоступен, scan_* возвращает вердикт ALLOW и пишет предупреждение fail_open_invoked. Приложение продолжает работать (доступность важнее). Факт деградации сохраняется в локальном буфере (только метаданные, без текстов — 152-ФЗ).
closed — поднимается BackendUnavailableError. Используйте, когда пропускать непроверенный трафик недопустимо.

gate = AIGate(url=..., tenant_id=..., profile_id=..., fail_mode="closed")

Размыкатель цепи (circuit breaker) не даёт приложению «висеть» на недоступном шлюзе: после серии ошибок вызовы мгновенно завершаются по выбранному fail_mode.

Аудит и наблюдаемость¶

Все проверки SDK видны в Admin UI → События безопасности с источником transport = sdk и атрибуцией приложения (имя/версия приложения, версия SDK, хост). Фильтр по источнику — выпадающий список «Источник».
С экстрами [prometheus] / [otel] SDK экспортирует метрики (aigate_sdk_scans_total, aigate_sdk_scan_duration_seconds, aigate_sdk_backend_errors_total и др.) и OTel-спан на каждую проверку.

Безопасность и 152-ФЗ¶

Тексты промптов и ответов не логируются по умолчанию (AIGATE_LOG_PROMPTS=false); буфер деградации хранит только метаданные.
Импорт aigate не «патчит» сторонние библиотеки — обёртки активируются только явным вызовом.
В зависимостях SDK запрещены ML-фреймворки (детекция строго серверная).

Персональные данные

Если приложение обрабатывает ПДн, помните о требованиях 152-ФЗ: не включайте AIGATE_LOG_PROMPTS в проде без необходимости и согласования, и полагайтесь на серверную маскировку PII (SANITIZE).

Дополнительная информация¶

Профили — создание scan_only-профиля.
События безопасности — просмотр и фильтрация событий SDK.
Интеграция с LiteLLM — альтернатива через LLM Gateway.