API Reference

Complete reference for eventkit’s HTTP API, configuration, and protocols.

HTTP API

Health Endpoints


GET /health

Liveness probe - Returns 200 if process is running (no dependencies checked).


{"status": "ok"}


GET /ready

Readiness probe - Returns 200 if dependencies (Firestore, GCS) are healthy, 503 if not.


{"status": "ready"}

Collection Endpoints


POST /collect/{stream}
POST /collect

Flexible event collection - Accept any JSON payload.

Parameters:

stream (path, optional): Stream name for routing (default: “default”)

Request Body:

Single event (object)
Batch of events (array)

Response: 202 Accepted


{"status": "accepted"}

Example:


curl -X POST http://localhost:8000/collect/users \
  -H "Content-Type: application/json" \
  -d '{"type": "identify", "userId": "user_123", "traits": {"email": "user@example.com"}}'

Segment-Compatible Endpoints


POST /v1/identify

Identify users - Create or update user profiles.

Request Body:


{
  "type": "identify",
  "userId": "user_123",
  "traits": {
    "email": "user@example.com",
    "name": "John Doe",
    "plan": "enterprise"
  }
}

Routes to users stream.


POST /v1/track

Track events - Record user actions.

Request Body:


{
  "type": "track",
  "userId": "user_123",
  "event": "Button Clicked",
  "properties": {
    "button_id": "cta-signup",
    "page": "/home"
  }
}

Routes to events stream.


POST /v1/page

Track page views - Record page navigation.

Request Body:


{
  "type": "page",
  "userId": "user_123",
  "name": "Home Page",
  "properties": {
    "url": "/home",
    "referrer": "/landing"
  }
}

Routes to pages stream.

Configuration

See Configuration Reference for all environment variables.

Key Settings:

Category	Variable	Default	Description
Storage	`EVENTKIT_EVENT_STORE`	`"gcs"`	Storage backend (`gcs`, `firestore`)
	`GCP_GCS_BUCKET`	required	GCS bucket name
	`GCP_BIGQUERY_DATASET`	required	BigQuery dataset
Queue	`EVENTKIT_QUEUE_MODE`	`"async"`	Queue mode (`async`, `pubsub`)
	`EVENTKIT_ASYNC_WORKERS`	`4`	Number of async workers
Ring Buffer	`EVENTKIT_RING_BUFFER_DB_PATH`	`"eventkit_ring_buffer.db"`	SQLite database path
	`EVENTKIT_RING_BUFFER_MAX_SIZE`	`100000`	Max events to keep

Protocols

eventkit defines several protocols (interfaces) for extensibility.

EventStore Protocol


class EventStore(Protocol):
    async def store(self, event: TypedEvent) -> None:
        """Store a single event."""
 
    async def store_batch(self, events: list[TypedEvent]) -> None:
        """Store a batch of events (more efficient)."""
 
    def health_check(self) -> bool:
        """Check if storage backend is healthy."""

Implementations:

GCSEventStore - Google Cloud Storage (default)
FirestoreEventStore - Firestore (development/testing)

Custom Implementation:


class MyCustomStore(EventStore):
    async def store_batch(self, events: list[TypedEvent]) -> None:
        # Your implementation
        pass
 
    def health_check(self) -> bool:
        # Your implementation
        return True

EventQueue Protocol


class EventQueue(Protocol):
    async def enqueue(self, event: RawEvent) -> None:
        """Add event to queue."""
 
    async def start(self) -> None:
        """Start queue processing."""
 
    async def stop(self) -> None:
        """Stop queue processing gracefully."""

Implementations:

AsyncQueue - In-process async workers (default)
PubSubQueue - Google Pub/Sub distributed workers

WarehouseLoader Protocol


class WarehouseLoader(Protocol):
    async def start(self) -> None:
        """Start background loading process."""
 
    async def stop(self) -> None:
        """Stop background loading process."""
 
    async def load_files(self, file_paths: list[str]) -> None:
        """Load specific files (for manual triggering)."""

Implementations:

BigQueryLoader - GCS → BigQuery (default)
Future: SnowflakeLoader, RedshiftLoader, ClickHouseLoader

Event Schema

RawEvent

Flexible container for any JSON payload.


class RawEvent(BaseModel):
    payload: dict[str, Any]       # Original JSON
    received_at: datetime          # Ingestion timestamp
    stream: str | None = None      # Routing/isolation

TypedEvent

Base class for validated events.


class TypedEvent(BaseModel):
    event_id: str
    timestamp: datetime
    user_id: str | None
    anonymous_id: str | None
    event_type: str
    properties: dict[str, Any]
    stream: str | None

Subclasses:

IdentifyEvent - User identification
TrackEvent - User actions
PageEvent - Page views

Metrics

Prometheus metrics exposed on port 9090 (configurable).


curl http://localhost:9090/metrics

Key Metrics:

eventkit_events_received_total - Events received
eventkit_events_processed_total - Events processed
eventkit_events_failed_total - Events failed validation
eventkit_queue_depth - Current queue depth
eventkit_storage_bytes_written_total - Bytes written

See Monitoring for complete metrics reference.

Next Steps

Configuration Reference - Complete environment variable reference
Protocols - Implement custom backends
Local Development - Try the API locally