PyRapide is an open-source Python library for modeling, executing, and analyzing causal event-driven architectures. Built on the mathematical foundations of RAPIDE 1.0 from Stanford University, PyRapide gives engineers first-class tools to trace causality, detect patterns, and enforce constraints across distributed systems, MCP servers, and agentic AI workflows. Now with drop-in adapters for AutoGen, LangGraph, CrewAI, LlamaIndex, Agno, OpenAI Swarm, MetaGPT, and FlowiseAI.
____ ____ _ __
/ __ \__ __/ __ \____ _____ (_)___/ /__
/ /_/ / / / / /_/ / __ `/ __ \/ / __ / _ \
/ ____/ /_/ / _, _/ /_/ / /_/ / / /_/ / __/
/_/ \__, /_/ |_|\__,_/ .___/_/\__,_/\___/
/____/ /_/
<e1> --causes--> <e2>
| |
v v
<e3> <e4> --causes--> <e5>What Is PyRapide?
Most event-driven systems treat events as a flat stream: timestamped entries in a log, ordered by when they happened. PyRapide treats events as a causal graph: a directed acyclic structure that captures why each event happened, which events caused which other events, and which events are truly independent.
When you have ten microservices, fifty MCP servers, or a fleet of autonomous agents all generating events concurrently, a timestamp-ordered log cannot answer the question that actually matters: what caused this failure?
PyRapide can.
Core Concepts
A complete toolkit for causal event-driven architecture.
Define
Declare system components with typed interfaces, actions, and behavioral contracts.
Connect
Route events between components through pattern-matched connections with three causal semantics.
Constrain
Specify what must happen and what must never happen with declarative behavioral rules.
Execute
Run architectures with an async-native engine that tracks full causal history.
Analyze
Query causal computations with root-cause analysis, impact sets, predictions, and anomaly detection.
Stream
Process events from multiple concurrent sources in real time with sliding-window processing.
Your First Causal Architecture
Define interfaces, wire connections, and let the engine track every causal relationship automatically.
from pyrapide import interface, action, module, when
from pyrapide import architecture, connect, Pattern, Engine
import asyncio
@interface
class Sensor:
@action
async def reading(self, temperature: float) -> None: ...
@interface
class Alerter:
@action
async def alert(self, message: str, severity: str) -> None: ...
@architecture
class MonitoringSystem:
sensor: Sensor
alerter: Alerter
def connections(self):
return [connect(Pattern.match("Sensor.reading"), "alerter")]
# e1 --causes--> e2 --causes--> e3
# Causal history, not just a log.
Built for Real Systems
Causal event-driven architecture applies everywhere complex, concurrent events need to be understood.
Healthcare
Patient safety monitoring with causal tracing across ICU devices and medication systems.
View example →Finance
Transaction fraud detection through causal sequence analysis across payment networks.
View example →Government
Inter-agency incident response with full causal accountability and after-action review.
View example →Education
Learning analytics that distinguish causal interventions from mere correlations.
View example →Manufacturing
Predictive maintenance linking equipment sensor data to product quality outcomes.
View example →Autonomous Agents
Multi-agent orchestration with traceable tool invocations and plan-step accountability.
View example →MCP Server Integration
When dozens of MCP servers fire tool calls simultaneously, timestamp-ordered logs cannot answer what caused a failure. PyRapide's MCPEventAdapter translates MCP protocol events into causally-linked event graphs.
- ✓ Track full causal chains across multiple servers
- ✓ Enforce constraints like "every tool call must produce a result"
- ✓ Find the blast radius of any failure with forward slicing
from pyrapide import MCPEventAdapter, MCPPatterns
from pyrapide import StreamProcessor, must_match, never
processor = StreamProcessor()
processor.add_source("db", MCPEventAdapter("database-server", db_client))
processor.add_source("ai", MCPEventAdapter("ai-server", ai_client))
# Every tool call must produce a result
processor.enforce(must_match(
MCPPatterns.tool_roundtrip(),
name="tool_completion"
))
await processor.run(window_size=10000)
Stanford Roots, Modern Python
PyRapide traces its intellectual lineage to RAPIDE, an architecture description language developed at Stanford University's Computer Systems Laboratory during the 1990s by David C. Luckham and team. Their key insight, that causal relationships rather than mere timestamps are the fundamental structure worth capturing, has proven prescient in the era of MCP servers and autonomous AI agents.
Read the full history →