Skip to Content
eventkit is in active development. APIs may change.
Introduction

eventkit

Event ingestion and processing kit for Python.

Overview

eventkit is a production-ready kit for building event collection pipelines. Clone it, customize it, make it yours.

Philosophy: Provide a solid starting point with battle-tested patterns, then get out of your way. Customize for your specific needs.

Key Features

  • Flexible ingestion - Accept any JSON payload with Segment-compatible API
  • Stream-based routing - Separate processing pipelines by event type for isolation and scalability
  • Adapter pattern - Pluggable validators for multiple event formats and sources
  • Pluggable storage - Write to GCS + BigQuery (default) or implement custom backends
  • Error handling - Built-in dead letter queue for validation failures and retries
  • Type-safe - Full Pydantic v2 validation with strict typing throughout
  • Async-first - Built on FastAPI with async/await for high throughput
  • Cloud-native - Designed for Cloud Run, GKE, and containerized deployments

Quick Start

# Clone and customize
git clone https://github.com/prosdevlab/eventkit.git my-event-pipeline
cd my-event-pipeline
uv sync
 
# Start local dev
docker-compose up -d  # GCS + PubSub emulators
uv run uvicorn eventkit.api.app:app --reload
 
# Send your first event
curl -X POST http://localhost:8000/api/v1/identify \
  -H "Content-Type: application/json" \
  -d '{
    "userId": "user_123",
    "traits": {
      "email": "user@example.com",
      "plan": "enterprise"
    }
  }'

Use Cases

  • Build a lightweight CDP without vendor lock-in
  • Collect product analytics events at scale
  • Centralize event data from multiple sources (web, mobile, server-side)
  • Create custom event processing pipelines with your own business logic
  • Prototype and validate data architectures before committing to enterprise solutions

Next Steps

Last updated on
Advanced