Plugin Collection

Official EVOID plugins on PyPI. Each is an independent package. Install only what you need. Anyone can build and publish plugins following the Plugin Standard.

Quick Install

# Via evo CLI (short names)
evo install di
evo install redis
evo install smart-storage

# Via uv
uv add evoid-di
uv add evoid-redis
uv add evoid-smart-storage

Available Plugins

evoid-base (v0.1.2): Shared Contracts

The foundation. Defines StorageEngine, CacheEngine, and LoggerEngine contracts that all other plugins implement.

from evoid_base import StorageEngine, CacheEngine, LoggerEngine

# Every storage plugin follows this interface:
# async def read(key: str, **kwargs) -> Any | None
# async def write(key: str, data: dict, **kwargs) -> bool
# async def delete(key: str, **kwargs) -> bool
# async def health() -> bool

Why it matters: You can swap evoid-sqlite for evoid-postgresql without changing a single line of business code. The contract stays the same. That’s IOP: the Intent says “store this,” and the pipeline routes to whichever engine you installed.

Storage Plugins

Package	Install	Best For
`evoid-sqlite`	`evo install sqlite`	Prototyping, single-user apps, embedded
`evoid-postgresql`	`evo install postgresql`	Production, multi-tenant, complex queries
`evoid-scylla`	`evo install scylla`	High-throughput, distributed, Cassandra-compatible

from evoid_sqlite import create_storage

storage = create_storage("my_app.db")
await storage.write("user:1", {"name": "Alice"})
user = await storage.read("user:1")

# Pipeline: validate → authorize → your handler
# Storage plugin: chosen at config time, not code time

# Switch from SQLite to PostgreSQL? Change one config line:
# storage = "postgresql"  (was "sqlite")
# Your handler code stays identical.
```

evoid-redis (v0.1.2): Redis Cache

Async Redis cache with TTL. The standard choice for ephemeral data.

evo install redis

from evoid_redis import create_cache

cache = create_cache("redis://localhost")
await cache.set("session:abc123", {"user": "Alice"}, ttl=3600)
session = await cache.get("session:abc123")

async def handle_session(ctx: Context) -> dict:
    session_id = ctx.metadata["session_id"]
    # The cache plugin handles TTL, serialization, connection pooling
    # Your code just reads and writes
    return await ctx.deps["cache"].get(f"session:{session_id}")

# Pipeline: validate (5s timeout). That's it.
# No authorize. No audit. Fast path.
```

evoid-smart-storage (v0.1.2): Multi-DB Routing

Routes data to different backends automatically. The traffic controller for your storage layer.

evo install smart-storage

# evoid.toml
[engines]
storage = "smart_storage"

[engines.smart_storage.mapping]
credentials = "postgresql"    # Sensitive data → PostgreSQL
session = "redis"             # Temporary data → Redis
logs = "memory"               # Debug data → Memory

[engines.smart_storage.level_routing]
critical = "postgresql"       # Payments → PostgreSQL (ACID)
standard = "sqlite"           # Profiles → SQLite (simple)

# This Intent is EPHEMERAL: smart-storage routes to Redis
CACHE_HIT = Intent(name="cache_check", level=Level.EPHEMERAL)

# Same business code, different backends. The level decides.
# No if/else in your handler. No database imports.
```

evoid-di (v0.1.2): Dependency Injection with Fault Tolerance

Three levels of complexity plus automatic failover, health checking, and cluster integration.

evo install di

from evoid_di import di

# Level 1: Simple. Name in, instance out.
di.register("db", create_db)
db = di.resolve("db")

# Level 2: Scoped. Singleton, transient, or per-user.
di.register("db", create_db, scope="singleton")
di.register("session", create_session, scope="per_user")

# Level 3: Context-aware. Different impl based on Intent level.
di = DIEngine(rules_config=rules, implementations=impls)
instance = await di.resolve("notifier", ctx)

Fault Tolerance:

# Define fallback chain
di.set_fallback("storage.postgresql", ["storage.sqlite", "cache.redis"])

# Health checking
di.set_health_check("cache.redis", lambda: redis.ping())

# Auto-fallback on failure (never crashes)
storage = di.resolve_with_fallback("storage.postgresql")
# Tries: postgresql → sqlite → redis → cluster peers → None

# Resolve first available from list
cache = di.resolve_any("cache.redis", "cache.memory", "storage.sqlite")

Cluster Integration:

# Connect cluster registry for remote resolution
from evoid_cluster import ServiceRegistry
di.set_cluster_registry(cluster._registry)

# Remote services become fallbacks automatically
storage = di.resolve("storage.postgresql")
# If not local, checks cluster peers

evoid-auth (v0.1.2): Authentication & Authorization

Bring your own provider. No forced JWT, no forced OAuth. Just a function that takes a token and returns a role.

evo install auth

from evoid_auth import register_provider

async def my_auth(token: str) -> dict:
    user = await db.find_by_token(token)
    return {"user": user.name, "role": user.role}

register_provider("my_auth", my_auth)

# Wire to pipeline
from evoid.core.extend import before
before("GET:/users", "authenticate")

# CRITICAL: validate + authorize + audit + protect
TRANSFER_MONEY = Intent(name="transfer_money", level=Level.CRITICAL)
# Pipeline: validate → authorize → audit → protect → handler
# Same auth check, plus audit log, plus protection layer

# EPHEMERAL: validate only (no auth needed)
HEALTH_CHECK = Intent(name="health_check", level=Level.EPHEMERAL)
# Pipeline: validate. That's it. Fast, no overhead.
```

evoid-tasks (v0.1.2): Background Tasks

Godot-inspired task lifecycle with EVOID pipeline integration. Fire-and-forget, scheduled, or event-driven.

evo install tasks
uv add "evoid-tasks[loguru]"  # with structured logging

from evoid_tasks import scheduler, TaskContext

# Fire-and-forget
scheduler.run(send_email, to="alice@example.com")

# Scheduled with lifecycle
@scheduler.task(interval=60)
async def monitor(ctx: TaskContext):
    if ctx.tick:
        await check_levels()

# Event-driven
@scheduler.on("order_placed")
async def update_stats(ctx: TaskContext):
    await recalc(ctx.event_data)

# A background task becomes an IOP Intent
SYNC_INVENTORY = as_intent(
    name="sync_inventory",
    level=Level.STANDARD,
    pipeline=("validate", "authorize"),
)

# The task runs through the same pipeline as any other Intent
# Same auth, same validation, same audit. Because it's IOP.
# No special "background task" logic. Just a processor.
```

evoid-scheduler (v0.1.3): Priority-Aware Scheduling

Replaces EVOID’s built-in parallel execution with a system-aware priority scheduler. Auto-defers low-priority tasks when the system is overloaded.

evo plug install evoid-scheduler

from evoid_scheduler import SchedulerEngine, Priority

scheduler = SchedulerEngine()

# High-priority intent goes first
scheduler.submit(process_payment, priority=Priority.CRITICAL)

# Low-priority gets deferred if CPU is busy
scheduler.submit(sync_analytics, priority=Priority.LOW)

ANALYTICS = Intent(
    name="sync_analytics",
    level=Level.STANDARD,
    metadata={"priority": Priority.LOW},  # 25
)

# Scheduler reads system load. If CPU > 80%:
#   - CRITICAL intents run immediately
#   - LOW intents get deferred to a queue
# Your code doesn't know this happens. The pipeline handles it.
```

evoid-cluster (v0.1.2): Multi-Node Clustering

Connects multiple EVOID nodes into a unified distributed system via WebSocket. Nodes share Intents, not data.

evo plug install evoid-cluster

# cluster.toml
[node]
id = "node-1"
host = "0.0.0.0"
port = 9000
roles = ["api", "worker"]

[[peers]]
host = "10.0.0.2"
port = 9000

[[services]]
pattern = "chat:*"

# Node 2 handles chat
SEND_MESSAGE = Intent(name="send_message", level=Level.STANDARD)

# ClusterBridge routes automatically:
# - "process_payment" stays on Node 1 (local handler)
# - "chat:send" forwards to Node 2 (remote handler)
# - Result comes back via WebSocket

# Your code doesn't know if the handler is local or remote.
# The Intent declares what. The cluster decides where.
```

evoid-godot (v0.1.3): Game Integration

Server-side adapter for connecting Godot games to EVOID. Works with the GDScript client plugin.

evo plug install evoid-godot

from evoid_godot import setup_game_hosting, game_intent_handler

# Setup default handlers for game events
setup_game_hosting("my-game")

PLAYER_MOVE = Intent(
    name="game:my-game:player_move",
    level=Level.EPHEMERAL,  # Game state is temporary
    metadata={"player_id": "abc", "x": 10, "y": 20},
)

# Pipeline: validate. That's it. Fast, no auth for movement.
# But a "purchase_item" intent? That's CRITICAL:

PURCHASE_ITEM = Intent(
    name="game:my-game:purchase_item",
    level=Level.CRITICAL,  # Real money involved
    metadata={"player_id": "abc", "item": "sword", "price": 9.99},
)
# Pipeline: validate → authorize → audit → protect → handler
```

evoid-maubot (v0.2.0): Matrix Bot Adapter

Maubot plugin for Matrix messaging. Converts Matrix events to Intents.

evo plug install evoid-maubot

from evoid_maubot import EvoidMaubot

adapter = EvoidMaubot("my-matrix-bot")
# Matrix messages become Intents, routed through the pipeline

evoid-transport (v0.1.2): Low-Latency UDP

Binary UDP protocol for game state synchronization. ~0.5ms overhead vs ~2-5ms for WebSocket.

evo plug install evoid-transport

from evoid_transport import EvoidUDPPort

transport = EvoidUDPPort()
await transport.start("my-game")

# Broadcast state to all players every tick
await transport.broadcast_state_sync(game_state, tick=60)

# Measure latency per client
latency = await transport.measure_latency("player-123")

# This maps directly to IOP levels:
# RELIABLE + CRITICAL = payments, legal moves
# RELIABLE + STANDARD = card plays, chat
# UNRELIABLE + EPHEMERAL = position, animation frames

# The transport doesn't know your game logic.
# Your Intent's level determines the channel.
```

evoid-dashboard (v0.1.2): Monitoring Dashboard

ASGI-based web dashboard. Service map, intent registry, message bus history, DB viewer.

evo install dashboard
uv add "evoid-dashboard[full]"  # with jinja2 + uvicorn

from evoid_dashboard import create_dashboard

# Run on port 8001
create_dashboard(port=8001)

Open http://localhost:8001 to see:

Service map with connections
All registered Intents
Message bus history
Database connections
System info

Combining Plugins

Plugins are Lego blocks. Snap together what you need:

# evoid.toml: full stack
[engines]
storage = "smart_storage"
cache = "redis"
di = "di"

[adapter]
type = "asgi"
port = 8000

from evoid_di import DIEngine
from evoid_auth import register_provider
from evoid_tasks import scheduler

# Smart Storage routes by Intent level
di = DIEngine(rules_config=rules, implementations={
    "sqlite": lambda: create_sqlite("app.db"),
    "redis": lambda: create_redis("redis://localhost"),
    "postgresql": lambda: create_postgres("postgres://..."),
})

# Auth with custom provider
register_provider("jwt", jwt_auth)

# Background task through the same pipeline
@scheduler.task(interval=300)
async def sync_inventory(ctx):
    await sync_all_locations()

# Pipeline executes:
# 1. validate      → schema check (built-in)
# 2. authorize     → evoid-auth checks role (plugin)
# 3. audit         → logs to PostgreSQL via smart-storage (plugin)
# 4. protect       → rate limit + circuit breaker (built-in)
# 5. handler       → your payment code
#
# Infrastructure chosen by:
#   - Level (CRITICAL → full pipeline)
#   - smart-storage config (critical → PostgreSQL)
#   - DI rules (CRITICAL → email notifier)
#
# Your handler? Just processes the payment. That's IOP.
```

Plugin Standard

Every plugin follows the EVOID plugin standard:

pyproject.toml with evoid>=0.4.0 dependency
evoid_plugin.json manifest
register_plugin() entry point
IOP-compliant code (data + functions)

See Plugin Standard for details.