NeuAIs Documentation

Nice constraint game, love it. Let’s do a lean Python spine that can sit in:

• a CLI tool
• a web app
• a daemon / worker
• or your weird mesh of agents

All in one file, no framework assumptions.

Below is one Python module (eventkit.py) with roughly:

• ~50 lines: event router (track/page/identify)
• ~30 lines: session/identity storage
• ~60 lines: plugin engine

I’ve kept imports minimal and logic “generic” (no hard dependency on HTTP, web, etc.).

# eventkit.py
from __future__ import annotations
from dataclasses import dataclass, asdict
from typing import Any, Dict, Protocol, List, Optional, Callable
import time, uuid, platform

# ---------- Session / Identity (≈30 lines) ----------

@dataclass
class SessionState:
    anonymous_id: str
    user_id: Optional[str]
    session_id: str
    traits: Dict[str, Any]
    consent: Dict[str, bool]
    last_activity: float

class Storage(Protocol):
    def load(self) -> Optional[Dict[str, Any]]: ...
    def save(self, data: Dict[str, Any]) -> None: ...

class MemoryStorage:
    def __init__(self): self._data: Optional[Dict[str, Any]] = None
    def load(self): return self._data
    def save(self, data): self._data = data

class IdentityManager:
    def __init__(self, storage: Storage, session_timeout_s: int = 1800):
        self.storage, self.timeout = storage, session_timeout_s
        self.state = self._load_or_new()

    def _new_state(self) -> SessionState:
        now = time.time()
        return SessionState(
            anonymous_id=str(uuid.uuid4()),
            user_id=None,
            session_id=str(uuid.uuid4()),
            traits={},
            consent={},
            last_activity=now,
        )

    def _load_or_new(self) -> SessionState:
        raw = self.storage.load()
        if not raw: return self._new_state()
        s = SessionState(**raw)
        if time.time() - s.last_activity > self.timeout:
            s.session_id = str(uuid.uuid4())
        return s

    def touch(self):
        self.state.last_activity = time.time()
        self.storage.save(asdict(self.state))

    def set_user(self, user_id: Optional[str], traits: Dict[str, Any] | None = None):
        self.state.user_id = user_id
        if traits: self.state.traits.update(traits)
        self.touch()

    def set_consent(self, consent: Dict[str, bool]):
        self.state.consent.update(consent)
        self.touch()

# ---------- Plugin Engine (≈60 lines) ----------

@dataclass
class Event:
    type: str                 # "track" | "page" | "identify"
    name: Optional[str]
    properties: Dict[str, Any]
    user_id: Optional[str]
    anonymous_id: str
    session_id: str
    timestamp: float
    context: Dict[str, Any]
    traits: Dict[str, Any]
    consent: Dict[str, bool]

class Plugin(Protocol):
    name: str
    def init(self, router: "EventRouter") -> None: ...
    def handle(self, event: Event) -> None: ...

class PluginEngine:
    def __init__(self):
        self._plugins: List[Plugin] = []
        self._filters: List[Callable[[Event], bool]] = []

    def register(self, plugin: Plugin):
        self._plugins.append(plugin)

    def add_filter(self, fn: Callable[[Event], bool]):
        """Return False to drop event."""
        self._filters.append(fn)

    def init_all(self, router: "EventRouter"):
        for p in self._plugins:
            try: p.init(router)
            except Exception as e: print(f"[eventkit] init error {p.name}: {e}")

    def dispatch(self, event: Event):
        for f in self._filters:
            try:
                if not f(event): return
            except Exception as e:
                print(f"[eventkit] filter error: {e}")
        for p in self._plugins:
            try: p.handle(event)
            except Exception as e:
                print(f"[eventkit] plugin error {p.name}: {e}")

class ConsoleLogger:
    name = "console_logger"
    def init(self, router: "EventRouter"): pass
    def handle(self, event: Event):
        print(f"[{event.type}] {event.name} props={event.properties} "
              f"user={event.user_id} session={event.session_id}")

# ---------- Event Router (≈50 lines) ----------

class EventRouter:
    def __init__(
        self,
        storage: Optional[Storage] = None,
        plugins: Optional[List[Plugin]] = None,
        default_consent: Optional[Dict[str, bool]] = None,
        environment: Optional[str] = None,
    ):
        self.storage = storage or MemoryStorage()
        self.identity = IdentityManager(self.storage)
        if default_consent: self.identity.set_consent(default_consent)
        self.engine = PluginEngine()
        for p in (plugins or []): self.engine.register(p)
        self.engine.add_filter(self._consent_filter)
        self.engine.init_all(self)
        self.environment = environment or platform.system().lower()

    # --- public API ---

    def track(self, event_name: str, properties: Dict[str, Any] | None = None):
        self._emit("track", event_name, properties)

    def page(self, name: str, properties: Dict[str, Any] | None = None):
        self._emit("page", name, properties)

    def identify(self, user_id: str, traits: Dict[str, Any] | None = None):
        self.identity.set_user(user_id, traits)
        self._emit("identify", "identify", traits or {})

    def set_consent(self, consent: Dict[str, bool]):
        self.identity.set_consent(consent)

    # --- internals ---

    def _base_context(self) -> Dict[str, Any]:
        return {
            "env": self.environment,
            "hostname": platform.node(),
        }

    def _emit(self, ev_type: str, name: str, properties: Dict[str, Any] | None):
        self.identity.touch()
        s = self.identity.state
        evt = Event(
            type=ev_type,
            name=name,
            properties=properties or {},
            user_id=s.user_id,
            anonymous_id=s.anonymous_id,
            session_id=s.session_id,
            timestamp=time.time(),
            context=self._base_context(),
            traits=dict(s.traits),
            consent=dict(s.consent),
        )
        self.engine.dispatch(evt)

    def _consent_filter(self, event: Event) -> bool:
        # Example: if analytics explicitly false, drop all
        if event.consent.get("analytics") is False:
            return False
        return True

# ---------- Tiny usage example ----------

if __name__ == "__main__":
    router = EventRouter(
        plugins=[ConsoleLogger()],
        default_consent={"analytics": True},
        environment="cli",
    )
    router.page("startup", {"version": "0.1.0"})
    router.identify("user-123", {"role": "admin"})
    router.track("job.enqueued", {"id": "abc", "priority": "high"})

How this fits “anything usual”

• No web assumptions: context is just {env, hostname} — you can extend that in your app.
• Storage is a protocol: swap MemoryStorage for Redis, SQLite, a JSON file, whatever.
• PluginEngine is generic: plugins can log, send HTTP, enqueue to Kafka, call an AI agent, write to a file, etc.
• Router API is boring and universal: track / page / identify, plus set_consent.

If you want next step, we can bolt on:

• a FileStorage (drop-in JSON-on-disk session persistence), or
• a RequestsDestination plugin that POSTs events to any URL, or
• an “AI sidecar” plugin that batches events and feeds them into your local model.