Python SDK¶

Официальный Python SDK для интеграции с AegisAI: aegisai.sdk.

Установка¶

SDK входит в основной пакет AegisAI:

poetry install
# или
pip install aegisai

Требования: Python 3.11–3.12, httpx.

AegisAIClient¶

Инициализация¶

from aegisai.sdk import AegisAIClient
from aegisai.sdk.types import AgentType

client = AegisAIClient(
    api_key="aegis_your_key",
    base_url="http://localhost:8000",  # по умолчанию
    agent_id=None,                      # задаётся после register()
    agent_name="My Agent",
    agent_type=AgentType.LLM,           # llm, tool, workflow, custom
    timeout=30,
)

Async context manager¶

async with AegisAIClient(api_key="...", agent_name="Bot") as client:
    await client.register()
    # ... работа с агентом
# client.close() вызывается автоматически

Методы¶

register()¶

Регистрирует агента в платформе.

reg = await client.register(
    capabilities=["observe", "track"],
    permissions={"max_risk": 0.5},
)
print(reg.id, reg.status)

Возвращает AgentRegistration:

Поле	Тип	Описание
`id`	`str`	UUID агента
`name`	`str`	Имя
`type`	`str`	Тип агента
`status`	`str`	`active`
`capabilities`	`list[str]`	Возможности

check_policy()¶

Проверяет действие по политикам OPA.

decision = await client.check_policy(
    action="delete_data",
    resource="personal_data",
    context={"approved": False},
)

if not decision.allowed:
    print(decision.reason)
if decision.required_approval:
    # запросить HITL
    ...

PolicyDecision:

Поле	Тип	Описание
`allowed`	`bool`	Разрешено
`required_approval`	`bool`	Нужен HITL
`risk_score`	`float`	0.0–1.0
`risk`	`str`	low / medium / high / critical
`anomaly`	`bool`	Аномалия
`reason`	`str`	Обоснование

guard.validate() / guard.is_safe()¶

Проверка промпта на injection / jailbreak / data leakage:

if await client.guard.is_safe(prompt):
    response = await llm.generate(prompt)
else:
    result = await client.guard.validate(prompt)
    print(result["threat_level"], result["reasons"])

См. Prompt Guard.

observe()¶

Context manager для автоматического логирования действия.

async with client.observe("analyze_data", "sales_db") as ctx:
    ctx.add_reasoning("Шаг 1: Загрузка данных")
    result = await do_analysis()
    ctx.add_reasoning("Шаг 2: Анализ завершён")
    ctx.set_output({"rows": 150})

При check_policy=True (по умолчанию) перед выполнением проверяется политика. При отказе выбрасывается PolicyDeniedError.

log_action()¶

Прямая запись в журнал аудита.

entry = await client.log_action(
    action="read_data",
    resource="customer_db",
    input_data={"query": "SELECT ..."},
    output={"count": 42},
    reasoning=["Запрос к БД"],
    risk_score=0.1,
    is_approved=True,
)
print(entry.id, entry.signature)

heartbeat()¶

Сигнал жизни агента.

await client.heartbeat(status="active")

ObserverContext¶

Метод	Описание
`add_reasoning(step)`	Добавить шаг рассуждения
`set_output(data)`	Установить результат
`set_risk_score(score)`	Задать risk score

При выходе из async with контекст автоматически вызывает log_action().

Исключения¶

from aegisai.sdk.exceptions import (
    APIError,              # Ошибка API (4xx, 5xx)
    AgentNotRegisteredError,  # Вызов без register()
    PolicyDeniedError,     # Политика отклонила действие
)

try:
    await client.log_action("test", "res")
except AgentNotRegisteredError:
    await client.register()
except APIError as e:
    print(f"HTTP {e.status_code}: {e}")

Типы агентов¶

from aegisai.sdk.types import AgentType

AgentType.LLM        # "llm"
AgentType.TOOL       # "tool"
AgentType.WORKFLOW   # "workflow"
AgentType.CUSTOM     # "custom"

Полный пример¶

import asyncio
from aegisai.sdk import AegisAIClient

async def main():
    async with AegisAIClient(
        api_key="aegis_acme_demo_seed_key_001",
        agent_name="Demo Agent",
    ) as client:
        await client.register()

        decision = await client.check_policy("read_data", "customer_pii", {
            "consent": True,
            "purpose": "analytics",
        })
        print(f"Allowed: {decision.allowed}, Risk: {decision.risk_score}")

        if decision.allowed:
            async with client.observe("read_data", "customer_pii") as ctx:
                ctx.add_reasoning("Чтение с согласия пользователя")
                ctx.set_output({"records": 100})

        await client.heartbeat()

asyncio.run(main())

Примеры

См. examples/simple_agent.py, examples/langchain_agent.py, examples/openai_agent.py.