EventMesh

An MCP-native event store. Capture arbitrary events from any system, extract structured signals, enrich with a local LLM, and serve the result to AI agents over Model Context Protocol — 4 read-only tools, zero third-party dependencies, deterministic millisecond reads because the summarising already happened at write time.

What is EventMesh?

EventMesh is a Go service that:

Receives events on a webhook (POST /webhook/events) from any producer.
Persists them in Postgres alongside their raw JSON and a structured facets view (actors, locations, references, timestamps, quantities, status words).
Deduplicates identical events using a SHA-256 fingerprint (UNIQUE constraint).
Enriches every event asynchronously with a local LLM (Ollama) — adding a one-line summary, 3-7 tags, and a severity label.
Exposes the resulting event store to AI agents (Claude Desktop, Claude Code, Cursor, custom agents) via an MCP server with 4 read-only tools — describe_data, query_events, get_entity_state, summarize_window — returning agent-readable data instead of raw JSON.

Two binaries:

Binary	Purpose	Status
`cmd/eventmesh-api`	Webhook server + AI worker	✅ Working
`cmd/eventmesh-mcp`	MCP server (stdio transport) for Claude Desktop / Claude Code	✅ Working

Why it exists

Most event stores hand back raw JSON. Agents waste tokens parsing it and frequently misread domain-specific fields. EventMesh inverts the contract: structured facets are extracted at write time, and an LLM summary is attached out-of-band — so any downstream agent can ask "what happened to shipment X this week?" and get a sub-200ms answer assembled from pre-computed summaries, with no per-query inference cost.

Highlights

Schema-agnostic. Send any JSON; the facet extractor infers actors/locations/references/timestamps without configuration.
Idempotent. Same event sent twice → silent no-op (deduped: true).
Domain-portable. Same model handles logistics events, IoT sensors, user-activity events, etc. — verified end-to-end.
Local-first. Postgres in Docker, Ollama on host. Zero cloud cost. Zero secrets.
12-factor compliant. Every knob lives in an env var with a sane default.
Graceful. SIGINT drains HTTP for 10s, cancels worker context, closes the pool.
Layered. Strict downward dependency direction makes every package unit-testable.

Architecture

High level

PRODUCER
  │
  │  POST /webhook/events  (JSON)
  ▼
┌─────────────────┐
│  api/handlers   │  thin HTTP layer
└────────┬────────┘
         │  IncomingEvent
         ▼
┌─────────────────┐
│  ingest/Service │  validate → extract facets → fingerprint → persist
└────────┬────────┘
         ▼
┌─────────────────┐
│  Postgres       │  events row written (ai_processed_at = NULL)
└────────┬────────┘
         │  ┄┄┄ ~1–15s later ┄┄┄
         ▼
┌─────────────────────────┐
│  worker/Enricher        │  ticks every 1s
│   ├ FetchUnprocessed    │
│   ├ ai.EnrichEvent ─────┼──► Ollama  (qwen2.5:7b)
│   └ MarkAIEnriched      │
└─────────────────────────┘
         │
         ▼
   events row updated:
     ai_summary + ai_tags + ai_severity + ai_processed_at

Dependency direction

  api  ─────────────► ingest ────────► structure
                         │                  │
                         └─────► db ◄───────┘
                                 │
                         worker ─┼─► ai ─► (Ollama HTTP)
                                 │
                              domain  ◄─── everyone

  config  ◄─── main only

domain sits at the bottom and depends only on stdlib. Everyone depends on domain. This is what makes every package independently testable.

Project layout

This repo follows the golang-standards/project-layout convention.

eventmesh/
├── api/                        # Public API contracts
│   └── openapi.yaml            # OpenAPI 3.1 spec for the HTTP surface
├── cmd/                        # Main binaries (composition roots)
│   ├── eventmesh-api/          # Webhook server + AI worker
│   │   └── main.go
│   └── eventmesh-mcp/          # MCP server (stdio, 4 read-only tools)
├── configs/                    # Sample env files, runtime configs
│   └── .env.example
├── deployments/                # Plug-and-play deployment surface (any cloud)
│   ├── docker-compose.yml      # Local Postgres (+ optional `--profile app`)
│   ├── kubernetes/             # Vanilla manifests (kubectl / kustomize)
│   ├── helm/eventmesh/         # Helm chart
│   ├── cloudrun/               # GCP Cloud Run Knative spec
│   ├── ecs/                    # AWS ECS / Fargate task + service defs
│   ├── terraform/              # Reference Terraform for AWS
│   └── README.md
├── docs/                       # Long-form documentation
│   ├── ARCHITECTURE.md
│   ├── DESIGN_DECISIONS.md
│   ├── TESTING.md
│   ├── HLD.txt
│   └── Project.md
├── internal/                   # Private application code (not importable externally)
│   ├── api/                    # HTTP handlers (thin layer)
│   ├── ai/                     # Ollama client + enrichment prompts
│   ├── config/                 # Env-driven configuration
│   ├── db/                     # Postgres pool + typed queries
│   ├── domain/                 # Canonical types (stdlib-only)
│   ├── ingest/                 # Orchestration layer
│   ├── structure/              # Facet extraction + fingerprint (pure)
│   └── worker/                 # Background AI enrichment loop
├── migrations/                 # SQL schema, applied at first Postgres boot
│   └── 001_init.sql
├── scripts/                    # Dev / ops helper scripts
│   ├── dev.sh
│   └── seed.sh
├── test/                       # Black-box tests, integration tests, fixtures
│   ├── integration/            # DB-bound tests (build tag: integration)
│   ├── unit/{ai,domain,structure}/  # External package_test unit tests
│   └── testdata/               # JSON fixtures used in tests + by `make seed`
├── .github/                    # CI workflows, Dependabot, issue/PR templates, CODEQL
├── Dockerfile                  # Multi-stage build → distroless nonroot image
├── .dockerignore
├── Makefile                    # `make help` lists every target
├── .golangci.yml               # Linter config
├── .editorconfig               # Editor defaults
├── go.mod / go.sum
├── CHANGELOG.md
├── CODEOWNERS
├── CODE_OF_CONDUCT.md
├── CONTRIBUTING.md
├── SECURITY.md
├── LICENSE
└── README.md  ← you are here

Layout rules

cmd/<binary>/main.go is always thin: parse flags/env, build dependencies, run.
internal/... cannot be imported by anything outside this module — enforced by the Go toolchain.
domain imports only stdlib. All other packages may import domain.
Tests live next to source (*_test.go) for unit coverage. test/ holds black-box and integration tests.

Quick start

Prerequisites

Tool	Min version	Install
Go	1.26	`brew install go`
Docker	24+	`brew install --cask docker`
Ollama	latest	`brew install ollama && brew services start ollama`
qwen2.5:7b	—	`ollama pull qwen2.5:7b` (~4.7 GB)

Run

# 1. Start Postgres (migration runs automatically the first time)
make docker-up
# or directly:
docker compose -f deployments/docker-compose.yml up -d

# 2. Run the API + AI worker
make run
# or directly:
go run ./cmd/eventmesh-api

You should see:

level=INFO msg="config loaded" api_port=8080 ollama_model=qwen2.5:7b ai_worker=true
level=INFO msg="db connected"
level=INFO msg=listening addr=:8080
level=INFO msg="ollama reachable; enrichment loop starting" component=enricher tick=1s

Send an event

curl -X POST http://localhost:8080/webhook/events \
  -H 'Content-Type: application/json' \
  -d @test/testdata/maersk_event.json

Response:

{"event_id":"<uuid>","deduped":false}

Send the same event again → {"deduped": true}.

Watch enrichment happen

Within ~5 seconds the worker logs:

level=INFO msg="enriching batch" component=enricher count=1
level=INFO msg="event enriched" event_type=vessel_departure severity=low \
                                summary="MAERSK LABREA departs Bremerhaven with container MMAU1286399."

Verify in Postgres

docker exec eventmesh-pg psql -U eventmesh -d eventmesh -c \
  "SELECT event_type, ai_severity, ai_tags, ai_summary FROM events ORDER BY ingested_at DESC LIMIT 5;"

Configuration

Every knob is environment-driven. Defaults match local development out-of-the-box.

Env var	Default	Description
`EVENTMESH_DB_URL`	`postgres://eventmesh:eventmesh_dev@localhost:5432/eventmesh?sslmode=disable`	pgx-compatible Postgres URL
`EVENTMESH_API_PORT`	`8080`	Port the webhook server binds to
`EVENTMESH_OLLAMA_URL`	`http://localhost:11434`	Base URL of the Ollama HTTP server
`EVENTMESH_OLLAMA_MODEL`	`qwen2.5:7b`	Model name registered with Ollama
`EVENTMESH_AI_WORKER`	`true`	Toggle the background AI enrichment loop

Copy configs/.env.example to .env and edit if you need to override anything.

HTTP API

`GET /health`

Liveness probe. Always 200 if the process is up.

{"status":"ok","checked_at":"2026-05-24T16:00:00Z"}

`POST /webhook/events`

Ingest a single event. Synchronous DB write; AI enrichment happens async.

Request body

{
  "entity_id":   "MMAU1286399",
  "event_type":  "vessel_departure",
  "occurred_at": "2026-05-08T14:00:00Z",
  "payload":     { "...": "any json object" }
}

Constraints

entity_id, event_type, occurred_at, payload are all required.
Body is capped at 1 MB (http.MaxBytesReader).
Unknown top-level fields are rejected (DisallowUnknownFields) — keeps client typos loud.

Responses

Status	Body	Meaning
`202 Accepted`	`{"event_id": "<uuid>", "deduped": false}`	Stored successfully
`202 Accepted`	`{"event_id": "<uuid>", "deduped": true}`	Same fingerprint already exists — no-op
`400 Bad Request`	`{"error": "..."}`	Validation failure (missing field, bad JSON, body too big)
`500 Internal Server Error`	`{"error": "ingest failed"}`	Unexpected error; check server logs

MCP server

eventmesh-mcp speaks Model Context Protocol over a stdio transport. All four tools are read-only — an agent triaging an incident has no path to mutate the store, which is a property worth having structurally rather than by convention.

The protocol layer is hand-rolled JSON-RPC 2.0 (internal/mcp) rather than an SDK, which is what keeps this module at exactly one dependency and the release image ~20MB.

Tools

Tool	Purpose	Required args
`describe_data`	List every event type with description, count and time range. Call this first when you don't know the domain.	—
`query_events`	Filter events by entity, type, time window, tags or severity. Returns summaries and facets; raw payload is opt-in.	—
`get_entity_state`	Current merged state of one entity, plus event count and first/last seen.	`entity_id`
`summarize_window`	Precomputed AI summaries across a time range, oldest-first, with a severity tally.	`since`

summarize_window performs no LLM call — summaries were computed once by the enrichment worker at write time. That is what makes it a millisecond read rather than a multi-second inference, and why an agent can afford to call it in a loop.

All timestamps are RFC3339 UTC, normalised on read so identical data serialises identically regardless of where the server runs.

Run it

make run-mcp                     # or: go run ./cmd/eventmesh-mcp

The server logs to stderr and speaks protocol on stdout. Never print to stdout from this binary — a single stray byte corrupts the JSON-RPC stream and the client drops the session.

Connect Claude Desktop

Add to claude_desktop_config.json:

{
  "mcpServers": {
    "eventmesh": {
      "command": "/absolute/path/to/bin/eventmesh-mcp",
      "env": {
        "EVENTMESH_DB_URL": "postgres://eventmesh:eventmesh_dev@localhost:5432/eventmesh?sslmode=disable"
      }
    }
  }
}

Connect Claude Code

claude mcp add eventmesh /absolute/path/to/bin/eventmesh-mcp \
  -e EVENTMESH_DB_URL="postgres://eventmesh:eventmesh_dev@localhost:5432/eventmesh?sslmode=disable"

Drive it by hand

Any MCP client is just newline-delimited JSON-RPC, so a pipe works for debugging:

printf '%s\n' \
 '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18"}}' \
 '{"jsonrpc":"2.0","id":2,"method":"tools/list"}' \
 '{"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"describe_data","arguments":{}}}' \
 | ./bin/eventmesh-mcp

Database schema

Three tables. Seven indexes. Optimized for read-heavy MCP queries with a small write-time index cost.

`events`

Raw event log. One row per webhook call (modulo dedup).

Column	Type	Filled by	Purpose
`event_id`	`UUID PK`	DB `gen_random_uuid()`	Identity
`fingerprint`	`TEXT UNIQUE`	`structure.Fingerprint`	Dedup key (SHA-256)
`entity_id`	`TEXT`	request	What the event is about
`event_type`	`TEXT`	request	Category
`occurred_at`	`TIMESTAMPTZ`	request	When the real-world thing happened
`ingested_at`	`TIMESTAMPTZ`	DB `NOW()`	When we received it
`payload`	`JSONB`	request	Original event, untouched
`facets`	`JSONB`	`structure.Extract`	Structured signals
`searchable_text`	`TEXT`	`ingest.searchableText`	Flattened text (future FTS)
`ai_summary`	`TEXT NULL`	worker	One-sentence plain English
`ai_tags`	`TEXT[] NULL`	worker	3–7 short lowercase tags
`ai_severity`	`TEXT NULL`	worker	`low \| medium \| high \| critical`
`ai_processed_at`	`TIMESTAMPTZ NULL`	worker	NULL = worker hasn't seen this row

Indexes

idx_events_entity_time     (entity_id, occurred_at DESC)
idx_events_type_time       (event_type, occurred_at DESC)
idx_events_facets_gin      USING GIN (facets jsonb_path_ops)
idx_events_tags_gin        USING GIN (ai_tags)
idx_events_unprocessed     (ingested_at) WHERE ai_processed_at IS NULL  -- partial

`entities`

One row per entity_id. UPSERTed on every event. Powers "what is the current state of X?" in a single 5ms read.

`event_types`

Auto-built catalog. One row per distinct event_type ever seen. ai_description, field_types, and sample_events columns are reserved for the future schema profiler — a feature where an agent calls describe_data via MCP and learns the domain without reading raw events.

See migrations/001_init.sql for full DDL and column-level commentary.

Event lifecycle

Step	What happens	Latency
1	HTTP request hits `/webhook/events`	~5 ms
2	`api/handlers.go` decodes JSON, validates body size	~1 ms
3	`ingest.Service.Ingest` validates required fields	<1 ms
4	`structure.Extract` produces facets	~1 ms
5	`structure.Fingerprint` returns SHA-256 hex	~50 µs
6	`db.InsertEvent` (`ON CONFLICT DO NOTHING`)	~5 ms
7	`db.UpsertEntity`	~3 ms
8	`db.TouchEventType`	~2 ms
9	HTTP returns `202`	total: ~20 ms
⋯	AI columns are still NULL
10	Worker polls (`SELECT … WHERE ai_processed_at IS NULL FOR UPDATE SKIP LOCKED LIMIT 10`)	0–1 s
11	`ai.EnrichEvent` → Ollama with `format: "json"`	3–15 s on CPU
12	`db.MarkAIEnriched` updates the row	~3 ms
✓	Row is fully enriched; any agent query sees AI metadata

Package walkthrough

`internal/domain`

Canonical Go types — IncomingEvent, Event, Facets, Enrichment, UnprocessedEvent. Stdlib-only. Imported by every other package.

`internal/config`

Load() reads env vars, applies defaults, returns a populated Config. 12-factor compliant.

`internal/structure`

Pure functions, zero I/O.

Extract(payload any) Facets — walks decoded JSON and produces predictable facet keys regardless of input shape. Rules: timestamps → references → locations → actors → status words → quantities.
Fingerprint(entity, type, time, payload) string — deterministic SHA-256 hex with 0x1f separators to prevent collision ambiguity.

`internal/db`

Postgres connection pool and typed queries.

Open(ctx, url) *Pool — configures: 10 max conns, 2 min, 1-hour lifetime, 30-min idle, 5-second startup Ping.
InsertEvent (with ErrDuplicate), UpsertEntity, TouchEventType, FetchUnprocessed, MarkAIEnriched, MarkAIFailed.

`internal/ai`

Ollama HTTP client + enrichment prompt. No business logic.

Client.GenerateJSON(ctx, prompt, dst) — uses format: "json" for strict structured output.
Client.EnrichEvent(ctx, e) (*Enrichment, error) — coerces severity into the closed set, dedupes/lowercases tags.

`internal/worker`

One goroutine. Polls Postgres every tick, runs each unprocessed event through ai.EnrichEvent with a 45-second per-event timeout, marks success or failure. Errors are logged, never propagated — one bad event must not stop the loop.

`internal/ingest`

Validates → unmarshals → extracts facets → fingerprints → writes 3 tables. Returns ErrValidation so HTTP handlers can map to 400. Treats ErrDuplicate from db as success-with-deduped=true.

`internal/api`

HTTP layer. Zero business logic. http.MaxBytesReader, DisallowUnknownFields, structured JSON errors.

`cmd/eventmesh-api/main.go`

The composition root — the only file that imports all packages. Sets up signal-aware context, opens DB pool, starts HTTP server and AI worker as goroutines, blocks on signal, runs graceful shutdown.

Development

Layout principles

Composition root only in cmd/. Library packages never construct themselves at the top — main.go wires everything.
Internal beats public. Default location is internal/. Promote to a top-level package only when external consumers exist.
domain imports only stdlib. Anything common goes there.
One type, one truth. Don't define near-identical structs in two packages. Add to domain instead.
Pure functions where possible. structure is the model: zero I/O, fully unit-testable.

Working in the repo

make help        # list every target
make run         # start Postgres + API + worker
make test        # all unit tests
make test-int    # integration tests (needs DB up)
make lint        # golangci-lint
make fmt         # gofmt + goimports
make tidy        # go mod tidy
make build       # produce binaries in bin/
make clean       # rm bin/, stop containers

Testing

The repo has three tiers of tests:

Unit tests (default `go test`)

Live alongside source as *_test.go files. Cover pure logic (facet extraction, fingerprinting, prompt building, config loading, HTTP handler validation). No external dependencies.

go test ./...                   # everything
go test ./internal/structure/   # one package
go test -race -cover ./...      # with race detector + coverage

Black-box tests (`test/unit/`)

Use the package_test external package pattern to verify only the public surface area. Useful for catching unintentional API breaks.

go test ./test/unit/...

Integration tests (`test/integration/`)

Hit a real Postgres. Gated behind the integration build tag.

docker compose up -d
go test -tags=integration ./test/integration/...

Coverage targets

Package	Coverage goal
`internal/structure`	90%+
`internal/ingest`	85%+
`internal/ai`	80%+ (prompt building only; HTTP behind a fake)
`internal/api`	80%+
`internal/config`	100%

Make targets

make help         # show this list
make run          # docker compose up -d && go run ./cmd/eventmesh-api
make build        # build binaries into ./bin
make test         # unit tests
make test-int     # integration tests (needs docker compose up)
make lint         # golangci-lint
make fmt          # gofmt -w + goimports
make tidy         # go mod tidy
make seed         # POST sample events to local server
make db-shell     # psql into the running container
make db-reset     # destroy and recreate Postgres data
make clean        # remove bin/ and stop containers

Deployment

The deployments/ folder is plug-and-play across the major cloud targets. Each subdirectory holds a self-contained manifest set; pick the one that matches your platform.

Target	Path	Quick command
Local Docker Compose	`deployments/docker-compose.yml`	`make docker-up`
Any Kubernetes (EKS/GKE/AKS/DOKS)	`deployments/kubernetes/`	`kubectl apply -k deployments/kubernetes/`
Kubernetes (templated)	`deployments/helm/eventmesh/`	`helm install eventmesh deployments/helm/eventmesh`
GCP Cloud Run	`deployments/cloudrun/service.yaml`	`gcloud run services replace deployments/cloudrun/service.yaml`
AWS ECS / Fargate	`deployments/ecs/`	`aws ecs register-task-definition --cli-input-json file://deployments/ecs/task-definition.json`
AWS reference IaC	`deployments/terraform/`	`cd deployments/terraform && terraform apply`
AWS App Runner / Azure Container Apps	Use `Dockerfile` directly, point to your registry	—

The Dockerfile at the project root produces a static, ~20 MB distroless nonroot image (gcr.io/distroless/static-debian12:nonroot) — no shell, no package manager, no root. The CI pipeline at .github/workflows/ci.yml publishes it to GHCR on every main push.

See deployments/README.md for full per-target instructions.

Codebase graph

This repo is indexed with graphify — a knowledge graph of files, packages, symbols, and their cross-references. The full interactive graph viewer is hosted at:

🌐 https://eventmesh-vercel.vercel.app

Current snapshot: 534 nodes, 585 edges, 53 communities.

Use it to:

Navigate the package hierarchy and see god-nodes (high-fan-in symbols) at a glance.
Trace cross-file relationships before opening a single source file.
Onboard new contributors with a visual map of the codebase.

The graph artifacts live under graphify-out/:

File	Purpose
`graph.html`	Interactive force-directed graph (deployed at the URL above)
`GRAPH_REPORT.md`	Human-readable architecture report
`graph.json`	Raw graph (nodes + edges) for tooling
`manifest.json`	Build metadata for the current snapshot

Regenerate locally after code changes (no LLM/API needed):

graphify update .

Roadmap

Milestone	Status
Webhook ingest with facet extraction + idempotency	✅
Background AI enrichment via Ollama	✅
MCP server with 4 tools (`describe_data`, `query_events`, `get_entity_state`, `summarize_window`)	✅
Unit + integration test suite	🚧
HMAC signature on webhook	⬜
Multi-tenancy (`tenant_id` + row-level security)	⬜
Schema profiler (auto-fills `event_types.ai_description`)	⬜
pgvector + embedding column (for `find_similar` MCP tool)	⬜
LISTEN/NOTIFY instead of polling	⬜
Batched LLM calls	⬜
Pluggable LLM backend (OpenAI / Anthropic / vLLM)	⬜

See docs/DESIGN_DECISIONS.md for the reasoning behind these choices.

Contributing

Pull requests welcome. Please read CONTRIBUTING.md first.

Run make fmt lint test before opening a PR.
One concern per commit; descriptive subject lines.
New behavior needs a unit test. New SQL needs a migration file.

For security issues, see SECURITY.md.

License

MIT — do whatever you want, just keep the copyright notice.

Name		Name	Last commit message	Last commit date
.claude		.claude
.github		.github
api		api
cmd		cmd
configs		configs
deployments		deployments
docs		docs
internal		internal
migrations		migrations
scripts		scripts
test		test
.dockerignore		.dockerignore
.editorconfig		.editorconfig
.gitignore		.gitignore
.golangci.yml		.golangci.yml
CHANGELOG.md		CHANGELOG.md
CLAUDE.md		CLAUDE.md
CODEOWNERS		CODEOWNERS
CODE_OF_CONDUCT.md		CODE_OF_CONDUCT.md
CONTRIBUTING.md		CONTRIBUTING.md
Dockerfile		Dockerfile
LICENSE		LICENSE
Makefile		Makefile
README.md		README.md
SECURITY.md		SECURITY.md
go.mod		go.mod
go.sum		go.sum

Latest commit

History

Folders and files

Repository files navigation

EventMesh

Table of contents

What is EventMesh?

Why it exists

Highlights

Architecture

High level

Dependency direction

Project layout

Layout rules

Quick start

Prerequisites

Run

Send an event

Watch enrichment happen

Verify in Postgres

Configuration

HTTP API

GET /health

POST /webhook/events

MCP server

Tools

Run it

Connect Claude Desktop

Connect Claude Code

Drive it by hand

Database schema

events

entities

event_types

Event lifecycle

Package walkthrough

internal/domain

internal/config

internal/structure

internal/db

internal/ai

internal/worker

internal/ingest

internal/api

cmd/eventmesh-api/main.go

Development

Layout principles

Working in the repo

Testing

Unit tests (default go test)

Black-box tests (test/unit/)

Integration tests (test/integration/)

Coverage targets

Make targets

Deployment

Codebase graph

Roadmap

Contributing

License

About

Topics

Resources

Code of conduct

Contributing

Security policy

Stars

Watchers

Forks

Releases

Packages

Used by

Contributors

Languages

`GET /health`

`POST /webhook/events`

`events`

`entities`

`event_types`

`internal/domain`

`internal/config`

`internal/structure`

`internal/db`

`internal/ai`

`internal/worker`

`internal/ingest`

`internal/api`

`cmd/eventmesh-api/main.go`

Unit tests (default `go test`)

Black-box tests (`test/unit/`)

Integration tests (`test/integration/`)