iknowhy 0.2.0

Causal event capture for AI agents — with native PyRapide integration

iknowhy

Python SDK for iKnowhy.ai — the causal event intelligence platform.

Log events, trace causal chains, and surface constraint violations across your systems.

pip install iknowhy

Quickstart

from iknowhy import CausewayClient

client = CausewayClient(api_key="cwy_...")

# Log a causal chain: user_login → process_spawn (caused by login)
login_id = client.log("user_login", payload={"user": "alice"}, session_id="sess1")
proc_id  = client.log("process_spawn", payload={"pid": 1234}, cause_ids=[login_id])

# Query the chain
chain = client.chain(proc_id)
# [{"id": login_id, "name": "user_login", ...}, {"id": proc_id, ...}]

---

PyRapide Integration

iKnowhy integrates natively with PyRapide, the causal event-driven architecture library for Python by Shane Morris.

PyRapide implements RAPIDE (Rapid Prototyping for Application Domain-specific Integrated Environments), developed at Stanford University by David Luckham. It gives you first-class tools to model, execute, and analyze causal event architectures in Python with async-native execution, Pydantic events, and a composable pattern algebra.

iKnowhy's PyRapideAdapter bridges PyRapide's local causal computation into iKnowhy's persistent event log — giving you durable storage, a web dashboard, and API access to every causal chain your PyRapide architecture produces.

pip install iknowhy pyrapide

Batch mode (after engine.run())

import asyncio
from pyrapide import Engine, architecture, interface, action, module, when, Pattern, connect, get_context
from iknowhy import CausewayClient
from iknowhy.pyrapide import PyRapideAdapter

@interface
class Scanner:
    @action
    async def scan(self, target: str) -> None: ...

@interface
class Classifier:
    @action
    async def classify(self, finding: str) -> None: ...

@module(implements=Scanner)
class ScannerModule:
    async def start(self):
        ctx = get_context(self)
        ctx.generate_event("Scanner.scan", payload={"target": "192.168.1.1"})

@module(implements=Classifier)
class ClassifierModule:
    @when(Pattern.match("Scanner.scan"))
    async def on_scan(self, match):
        ctx = get_context(self)
        event = list(match.events)[0]
        ctx.generate_event("Classifier.classify",
                           payload={"finding": "open port 22"},
                           caused_by=list(match.events))

@architecture
class SecurityArch:
    scanner: Scanner
    classifier: Classifier
    def connections(self):
        return [connect(Pattern.match("Scanner.scan"), "classifier")]

async def main():
    client = CausewayClient(api_key="cwy_...")
    adapter = PyRapideAdapter(client, session_id="security-scan-001")

    arch = SecurityArch()
    engine = Engine()
    engine.bind(arch, "scanner", ScannerModule)
    engine.bind(arch, "classifier", ClassifierModule)

    comp = await engine.run(arch, timeout=5.0)

    # Upload entire computation to iKnowhy — causal links preserved
    event_ids = await adapter.flush_computation(comp)
    print(f"Logged {len(event_ids)} events to iKnowhy")

asyncio.run(main())

Streaming mode (real-time via StreamProcessor)

from pyrapide import StreamProcessor, InMemoryEventSource, Event
from iknowhy import CausewayClient
from iknowhy.pyrapide import PyRapideAdapter

async def main():
    client = CausewayClient(api_key="cwy_...")
    adapter = PyRapideAdapter(client, session_id="live-monitor")

    source = InMemoryEventSource("agent")
    processor = StreamProcessor()
    processor.add_source("agent", source)

    # Attach iKnowhy — events are logged in real time as they flow through PyRapide
    adapter.attach_stream(processor, session_id="live-monitor")

    # Send events; they'll appear in your iKnowhy dashboard immediately
    await source.put(Event(name="tool_call", payload={"tool": "browser"}))
    await source.close()
    await processor.run()

asyncio.run(main())

---

API Reference

CausewayClient

Method	Description
log(name, payload, session_id, cause_ids, tags)	Log an event, returns event_id
chain(event_id)	Full causal ancestry for an event
violations()	Current constraint violations
sessions()	List of sessions
events(session_id, name, limit)	Query events with filters

PyRapideAdapter

Method	Description
flush_computation(comp, session_id, tags)	Upload a completed PyRapide Computation
attach_stream(processor, session_id, tags)	Hook into a live StreamProcessor
id_map()	PyRapide event.id → iKnowhy event_id mapping
clear()	Reset ID map between computations

---

Links

• iKnowhy Dashboard
• API Docs
• PyRapide on PyPI — by Shane Morris
• PyRapide on GitHub