A small, deliberately readable slice of a KitchenOS-style platform: Python
3.14, AWS Lambda + CDK, an OpenAPI-spec-driven API Gateway, event-driven
services over EventBridge/SQS, DynamoDB, aws-lambda-powertools, Pydantic v2, an
AI seam, and pytest/moto — runnable end to end on a laptop with no AWS account.
Three event-driven services around one EventBridge bus:
| Service | API | Reacts to | Emits |
|---|---|---|---|
| cooking-sessions | start / get / advance sessions | device.telemetry.reported |
cooking.session.started, ...completed |
| device-control | register / get / telemetry | cooking.session.started |
device.registered, device.command.sent, device.telemetry.reported |
| recipes | create / get / search | — | — |
The headline flow:
POST /sessions ─▶ cooking-sessions stores it, emits cooking.session.started
│
▼ (EventBridge rule ─▶ SQS ─▶ Lambda)
device-control sets the device to COOKING, emits device.command.sent
│
POST /devices/{id}/telemetry ─▶ emits device.telemetry.reported
│
▼
cooking-sessions stamps the temperature onto the active session
src/kitchenos/
core/ config, observability (powertools), event envelope, bus, storage
services/
cooking_sessions/ models.py · service.py (pure logic) · handler.py (Lambda)
device_control/ models.py · service.py · handler.py
recipes/ models.py · ai.py (AI seam) · service.py · handler.py
openapi/ per-service OpenAPI specs (the API source of truth)
infra/ AWS CDK app; rest_api_gateway.py = shared SpecRestApi construct
scripts/ local_harness.py (run with no AWS, via moto)
tests/ pytest + moto unit tests, plus a cross-service event-flow test
Design note: each service splits into a transport-agnostic service.py
(business logic) and a thin handler.py (powertools / Lambda glue). The local
harness and the Lambdas both call the same service.py, so behaviour is
identical however it's invoked. The bus has two transports — real EventBridge
in AWS, and an in-process one for local/tests — selected by
KITCHENOS_BUS_TRANSPORT.
The OpenAPI files under openapi/<service>/openapi.yaml are the source of
truth for the HTTP surface. The shared CDK construct KitchenOSRestApiGateway
(infra/rest_api_gateway.py) builds an aws_apigateway.SpecRestApi straight from
a spec — routes and integrations come from the file, and each operation's
x-amazon-apigateway-integration links to a Lambda by logical id (e.g.
CookingSessionsApiLambda), which the construct swaps for the real invoke ARN at
synth time. The local harness loads the same specs and prints their routes on
startup, so the contract is visible without AWS.
cooking-session is wired this way; device-control and recipes are fronted by
a plain HTTP API.
In production these specs live in a separate
platform-specsrepo (authored with$refs, Redocly-bundled), and also drive runtime request/response validation and the generation of Bruno collections + docs. Seeopenapi/README.md.
Requires Python 3.14 and uv.
uv sync --all-extras # creates .venv and installs everything
uv run poe test # pytest (moto-mocked AWS)
uv run poe lint # ruff
uv run poe run-local # HTTP server on :8080, backed by moto(Prefer not to prefix with uv run? source .venv/bin/activate once, then use
poe test, poe run-local, etc. You can also skip Poe entirely and call
uv run pytest / uv run python -m scripts.local_harness directly.)
Then, in another shell:
curl -s localhost:8080/recipes -X POST \
-d '{"title":"Roast Chicken","summary":"sunday roast","tags":["chicken"]}'
curl -s "localhost:8080/recipes/search?q=chicken"
curl -s localhost:8080/devices -X POST -d '{"device_id":"oven-1","model":"SmartOven"}'
curl -s localhost:8080/sessions -X POST -d '{"recipe_id":"r1","device_id":"oven-1"}'
curl -s localhost:8080/devices/oven-1 # state is now "cooking"
curl -s localhost:8080/devices/oven-1/telemetry -X POST -d '{"temperature_c":190}'Needs Node + the AWS CDK CLI (recent enough to expose the Python 3.14 Lambda
runtime), Docker (for Lambda bundling), and AWS credentials. The cooking-session
API is created as a SpecRestApi from openapi/cooking-session/openapi.yaml.
uv sync --extra infra
uv run poe synth # render CloudFormation
uv run poe deploy # deploy to a sandbox accountThis ships a single CDK stack.
pyproject.toml defines Poe tasks (install, lint, format, test,
run-local, synth, deploy); install runs uv sync --all-extras. Lint and
format are ruff.