prismacore 0.2.0

Python SDK for orchestrating sing-box VPN nodes

prismacore

Python SDK для управления sing-box VPN-нодами через PrismaCore Server API.

Оркестрация нескольких серверов, выбор ноды по региону/latency, сборка подписок для клиентов (Happ, v2rayNG и т.д.).

pip install prismacore

Требования: Python 3.10+. Redis опционален (кэш подписок).

---

Быстрый старт

import asyncio
from prismacore import PrismaCore

async def main():
    async with PrismaCore.single_node(
        url="http://203.0.113.10:8000",
        token="your-api-token",
    ) as core:
        await core.node().users.create_simple("alice", inbound_tags=["vless-in"])
        sub = await core.subscriptions.build("alice")
        print(sub[:80])

asyncio.run(main())

API-токен создаётся на server: POST /api/v1/auth/token.
Server — отдельный компонент, см. репозиторий.

---

Несколько серверов

from prismacore import PrismaCore, ServerConfig

core = PrismaCore([
    ServerConfig(url="https://de.example.com", token="...", region="de", name="de-1"),
    ServerConfig(url="https://fi.example.com", token="...", region="fi", name="fi-1"),
])

await core.node("de-1").users.list()
best = await core.regions.fastest_node("de")

Поле ServerConfig	Назначение
url, token	Адрес и токен PrismaCore server
region	Логическая группа (de, fi, …)
name	Идентификатор ноды для core.node("name")
role	vpn или backup_vpn
public_host	Публичный host (метка; ссылки строит server)

---

CRUD на ноде

node = core.node()  # или core.node("de-1")

await node.users.create_simple("alice", inbound_tags=["vless-in"])
await node.inbounds.create({"type": "vless", "tag": "vless-in", "listen_port": 443})
await node.outbounds.list()
await node.routes.delete(1)
await node.system.get_current_config()

Ресурсы: users, inbounds, outbounds, routes, system.
На каждом — create, list, get, update, delete.

---

Подписки

SDK собирает подписку с нескольких нод. HTTP-эндпoинт для клиентов не входит в пакет — оборачиваете сами:

text = await core.subscriptions.build(
    "alice",
    regions=["de", "fi"],
    fastest_only=True,   # одна быстрая нода на регион
    fmt="base64",        # или "raw"
)

await core.subscriptions.invalidate("alice", regions=["de", "fi"])

Пример FastAPI:

from fastapi.responses import PlainTextResponse

@app.get("/sub/{user_name}")
async def sub(user_name: str):
    body = await core.subscriptions.build(user_name, regions=["de", "fi"])
    return PlainTextResponse(body, media_type="text/plain", headers={
        "Profile-Title": "My VPN",
    })

---

Кэш (Redis)

from prismacore import CacheConfig

core = PrismaCore(servers=[...], cache_config=CacheConfig(enabled=True))

После изменения пользователей или инбаундов вызывайте subscriptions.invalidate(), иначе клиенты получат устаревшую подписку.

---

Ошибки

Исключение	Когда
ConfigError	Неверный конфиг, несколько нод без имени
TransportError	Сеть
AuthError	403 — неверный token
NotFoundError	404
ConflictError	409

Все наследуют PrismaCoreError.

---

Импорт

from prismacore import (
    PrismaCore,
    ServerConfig, CacheConfig, ProbeConfig,
    UserCreate, InboundCreate, OutboundCreate, RouteCreate,
)

---

Документация

Полный reference: sdk.md

Server (отдельный репозиторий): PrismaCore

English: README.en.md

---

License

MIT