eva-sdk (Python)

Модуль ядра Евы — отдельный процесс, который говорит с ядром двумя каналами:

• внутрь ядра — модуль поднимает MCP-сервер, ядро подключается и добавляет его инструменты в реестр. Так модуль даёт Еве новые руки;
• наружу из ядра — модуль ходит в HTTP API ядра: заводит чаты, ведёт туры и читает их поток, дёргает модель одноразовым вызовом. Так модуль приводит Еве новые поверхности (мессенджер, почта, голос).

Инструментальному модулю хватает первого канала. Модулю-поверхности нужны оба: он ловит входящее у себя, ведёт тур в ядре и доставляет ответ обратно.

Rust-версия живёт в репозитории ядра (crates/eva-sdk); протокол общий, описан в книге ядра — глава «Модули и SDK».

Установка

pip install git+https://git.desu.church/eva/sdk-py.git

Модули, которые Ева пишет себе сама, и раннер python-записей middleware пользуются тем же пакетом — в их окружении он стоит заранее.

Инструментальный модуль

import asyncio
from dataclasses import dataclass

from eva_sdk import Module, ToolCtx

# about — чем модуль представляется Еве, когда набор инструментов свернул
# его в строку: по этой фразе она решает, догружать ли семейство
module = Module("weather", about="Weather forecasts by city")

@dataclass
class Forecast:
    city: str
    days: int | None = None

@module.tool("forecast", "Прогноз на завтра в указанном городе", icon="🌧", brief="{city}")
async def forecast(ctx: ToolCtx, input: Forecast) -> str:
    return f"в {input.city} дождь"

asyncio.run(module.serve_uds("/run/eva-mcp/weather.sock"))

Ядро подключится к сокету, если он лежит в modules.socket_dir (тогда модуль виден в module_list и включается инструментом module_approve) или объявлен в modules.servers явно. Перезапуск модуля ядро переживает само — сокет вернётся, инструменты переподключатся.

Инструмент бросает ToolError — это ошибка инструмента: её прочтёт модель и решит, что делать. Любое другое исключение модуль не роняет, но уедет к модели тем же способом.

Вход инструмента

Аргументы объявляются датаклассом — типом второго параметра обработчика. Из него adaptix рождает JSON Schema для модели (| None — необязательное поле, значение по умолчанию едет в схему), он же приезжает в обработчик разобранным. Схема и разбор идут от одного типа, так что разъехаться не могут: обещанное модели — ровно то, что инструмент прочтёт. Вход не по схеме — ошибка инструмента, а не KeyError в середине. Инструменту без аргументов есть NoInput.

Описания полей и прочая правка схемы — рецептом реторты; своя уезжает в Module(..., retort=...):

from adaptix import P, Retort
from adaptix.json_schema import JSONSchema, JSONSchemaPatch, json_schema

retort = Retort(recipe=[
    json_schema(
        P[Forecast].city,
        JSONSchemaPatch().merge_with(JSONSchema(description="Город, для которого нужен прогноз")),
    ),
])
module = Module("weather", retort=retort)

Модуль-поверхность

from eva_sdk import Kernel

async with Kernel("http://127.0.0.1:8090") as kernel:
    chat = await kernel.create_chat(title="tg:чат")
    person = await kernel.resolve_person(alias="tg:5551234", name="Пикаро")

    # дешёвый вопрос мимо тура: истории и инструментов у него нет
    verdict = await kernel.complete(
        system="Отвечай одним словом: да или нет.",
        user="Это обращено к Еве?",
        max_tokens=8,
    )

    turn = (
        kernel.turn(chat.id, "Ева, привет!")
        .sender(person.id)
        .anchor(chat_id=-1001234567890, message_id=4471)
        .chat_title("Любители ВН")
        .source("telegram")
    )
    async for event in turn.send():
        if event.type == "text_delta":
            print(event.text, end="")

anchor — сообщение поверхности, на которое отвечает тур: инструменты модуля увидят его в своём ToolCtx и поймут, к чему цеплять реакцию или картинку.

.effort("high") — уровень рассуждения на один тур: "off", "low", "mid", "high" по возрастанию. Не задан — решает уровень ниже: настройка чата, ступень самой модели, конфиг ядра. Ставит его поверхность, когда сама знает, что вопрос тяжёлый или, наоборот, декоративный.

Тип события — строка, и список у ядра открытый: поверхность показывает события, которые знает, а незнакомые пропускает.

Тулы на один тур

Модульные инструменты живут в реестре ядра и видны всем турам. Поверхности же часто нужен тул, который существует только в её турах и который она исполняет сама: телеграмный «промолчать», кнопка на устройстве, диалог с тем, кто сейчас у экрана. Такой тул объявляется прямо на тур:

from dataclasses import dataclass

from eva_sdk import ExtraTool

@dataclass
class Alarm:
    at: str

alarm = ExtraTool("set_alarm", "Поставить будильник на устройстве", input=Alarm)

async for event in turn.extra_tools([alarm]).send():
    if (input := alarm.parse(event)) is not None:
        event.response.respond(f"будильник на {input.at} поставлен")

Схема для модели рождается из датакласса, им же parse разбирает вызов — как у модульных тулов. Вызов чужого тула и реплейное событие parse отдаёт None: накопленный хвост рисуют, но не исполняют, иначе переподключение исполнило бы вызов дважды.

Отвечает респондер из event.response — одноразовый: доставку он ведёт фоновой задачей, так что цикл по стриму не запирается на HTTP-запросе. Ответ строкой — один блок без шапки; [Block(title, content)] — секции, которые ядро отрендерит модели канонично. Провал — fail(текст), его модель прочтёт ошибкой вызова и решит, что делать. Вызов, брошенный без ответа, поток закроет сам, когда кончится или оборвётся: ядро стоит и ждёт именно этого ответа, и молчание стоило бы ему полного таймаута.

ExtraTool(..., ends_turn=True) делает тул вердиктом — успешный вызов закрывает тур (так живёт телеграмное «молчу»). timeout_secs — сколько ядру ждать ответа; не задан — 300 секунд, потолок 900.

Middleware: перехват до действия

Тот же сервер обслуживает стадии middleware — точки, где ядро само собирается что-то сделать и ждёт решения:

from eva_sdk import Cancel, Continue, Module, Patch, Stage

module = Module("guard")

@module.stage(Stage.TOOL_CALL)
async def no_shell_at_night(stage: str, payload: dict):
    if payload.get("name") == "shell" and _is_night():
        return Cancel("ночью shell закрыт")
    return Continue()

Решение — Continue(), Cancel(reason) или Patch(payload). Стадии объявляются в handshake (_meta["dev.eva/middleware"]), так что ядро зовёт только тех, кто их объявил. Молчание, падение и таймаут трактуются как Continue: сломанный перехватчик тормозит своё действие, а не всю Еву.

Стадии v1: turn.start, tool.call, message.out, memory.save, triage.decision.

На этом же SDK живёт раннер python-записей middleware: Ева пишет обработчик текстом, раннер забирает включённые записи из ядра, собирает цепочку и отвечает на middleware/handle как обычный модуль.

Что чьё

Граница простая: ядро отвечает за Еву, модуль — за поверхность.

• в ядре: личности, память, история чата, туры, реестр инструментов, политика доступа к опасным инструментам;
• в модуле: токены поверхности, свой allowlist, дебаунсы, потолки и краткость на чат и на человека, кэш вложений, привязка «внешний чат → чат ядра». Модуль держит свою базу и в базу ядра не лезет.

Правило рабочее, а не эстетическое: модуль обновляется и перезапускается отдельно от ядра, а общая база сделала бы их одним целым.