diff --git a/build.py b/build.py index ae9a097..91a83f1 100644 --- a/build.py +++ b/build.py @@ -168,6 +168,151 @@ def build_bundle(folder: Path, dry_run: bool = False) -> None: print(f"[MCP] {name} bundle written to {out_dir.relative_to(ROOT)}/ ({len(nav)} modules)") +# ── Template bundle ────────────────────────────────────────────────────────── + +# Component kinds shipped in the vendored template. ``placeholder`` names the +# seed token an agent renames: "resource" -> item/Item, "family" -> example/ +# Example; "" means ship as-is. +_TEMPLATE_COMPONENTS: list[dict[str, str]] = [ + # Per-resource slice (seed: item/Item). + {"kind": "model", "path": "domain/item/models.py", "placeholder": "resource", + "description": "Business Pydantic model for a resource."}, + {"kind": "ports-db", "path": "domain/item/ports/db.py", "placeholder": "resource", + "description": "Document model + repository factory for a resource."}, + {"kind": "ports-service", "path": "domain/item/ports/service.py", "placeholder": "resource", + "description": "CRUD service wired through the DI library."}, + {"kind": "converters", "path": "domain/item/ports/converters.py", "placeholder": "resource", + "description": "Optional pure converters between domain types."}, + {"kind": "routes", "path": "interfaces/item/routes.py", "placeholder": "resource", + "description": "Thin HTTP handlers bound to operationIds."}, + {"kind": "test-model", "path": "tests/unit/test_item_models.py", "placeholder": "resource", + "description": "Unit test for the resource model."}, + {"kind": "test-routes", "path": "tests/integration/test_item_routes.py", "placeholder": "resource", + "description": "Integration test at the HTTP edge."}, + # Per-family pipeline scaffold (seed: example/Example). + {"kind": "pipeline-abc", "path": "pipelines/example/abc.py", "placeholder": "family", + "description": "Port contracts for a pipeline family."}, + {"kind": "pipeline-impl", "path": "pipelines/example/impl.py", "placeholder": "family", + "description": "Concrete stages, wired root, and family registration."}, + {"kind": "pipeline-yaml", "path": "pipelines/example/pipeline.yaml", "placeholder": "family", + "description": "Graph spec (nodes, edges, roots, schema)."}, + {"kind": "pipeline-resources", "path": "pipelines/example/resources/pipeline.yaml", "placeholder": "family", + "description": "Runtime spec consumed by the shared loader."}, + {"kind": "pipeline-pattern", "path": "pipelines/example/resources/patterns/example.yaml", "placeholder": "family", + "description": "Pattern/rule fragment merged by the loader."}, + # Shared pipeline scaffolding (ship as-is). + {"kind": "pipeline-ports", "path": "pipelines/shared/ports.py", "placeholder": "", + "description": "Shared pipeline ports (FinalizePort, RegistryPort)."}, + {"kind": "pipeline-loader", "path": "pipelines/shared/loader.py", "placeholder": "", + "description": "Shared loader: DI wiring + pattern merge."}, + {"kind": "pipeline-registry", "path": "pipelines/shared/registry.py", "placeholder": "", + "description": "Shared registry: (kind, variant, mode) -> root class."}, + {"kind": "pipeline-simple", "path": "pipelines/item_sync.py", "placeholder": "", + "description": "Smallest pipeline form (plain async steps)."}, + # Second resource slice (seed: job). + {"kind": "job-model", "path": "domain/job/models.py", "placeholder": "", + "description": "Seeded second resource model (job)."}, + {"kind": "job-ports-db", "path": "domain/job/ports/db.py", "placeholder": "", + "description": "Seeded second resource repository (job)."}, + {"kind": "job-ports-service", "path": "domain/job/ports/service.py", "placeholder": "", + "description": "Seeded second resource service (job)."}, + {"kind": "job-routes", "path": "interfaces/job/routes.py", "placeholder": "", + "description": "Seeded second resource routes (job)."}, + # Root / shared files (ship as-is). + {"kind": "config", "path": "config.py", "placeholder": "", + "description": "Settings dataclass + singleton."}, + {"kind": "init", "path": "init.py", "placeholder": "", + "description": "Composition root (init/shutdown, get_svc)."}, + {"kind": "main", "path": "main.py", "placeholder": "", + "description": "Spec-first app bootstrap + lifespan."}, + {"kind": "worker", "path": "worker.py", "placeholder": "", + "description": "Poll loop + job status machine."}, + {"kind": "openapi", "path": "openapi.yaml", "placeholder": "", + "description": "OpenAPI contract skeleton."}, + {"kind": "pyproject", "path": "pyproject.toml", "placeholder": "", + "description": "Packaging + dependency pins."}, + {"kind": "dockerfile", "path": "Dockerfile", "placeholder": "", + "description": "API + worker image with private-index wiring."}, + {"kind": "compose", "path": "docker-compose.yaml", "placeholder": "", + "description": "api + worker + mongo compose file."}, + {"kind": "env-example", "path": ".env.example", "placeholder": "", + "description": "Environment variable template."}, + {"kind": "db-setup", "path": "infra/db/setup.py", "placeholder": "", + "description": "Mongo connection lifecycle."}, + {"kind": "repo", "path": "infra/db/repos/mongo.py", "placeholder": "", + "description": "Repository factory (base/populating/cached)."}, + {"kind": "service-adapter", "path": "infra/adapters/service.py", "placeholder": "", + "description": "Generic CRUD service adaptor."}, + {"kind": "script-seed", "path": "scripts/seed.py", "placeholder": "", + "description": "Operator seed script."}, + {"kind": "script-drop", "path": "scripts/drop.py", "placeholder": "", + "description": "Operator drop script."}, +] + + +def build_components_manifest(template_root: Path) -> dict[str, object]: + """Describe every component in the vendored template. + + Args: + template_root (Path): The vendored ``templates/app`` directory. + + Returns: + dict[str, object]: Manifest with the seed tokens and component entries. + """ + components: list[dict[str, str]] = [] + for entry in _TEMPLATE_COMPONENTS: + rel = entry["path"] + if not (template_root / rel).exists(): + raise SystemExit(f"[ERR] component path missing from template: {rel}") + components.append({ + "kind": entry["kind"], + "path": rel, + "placeholder": entry["placeholder"], + "description": entry["description"], + }) + return { + "project": "standards", + "template": "templates/app", + "seed": {"resource": "item", "family": "example"}, + "components": components, + } + + +def build_templates(folder: Path, dry_run: bool = False) -> None: + """Vendor ``folder/src/templates/app`` into ``/mcp/templates/app``. + + Ships the runnable skeleton in the MCP bundle so an agent can pull exact, + current component code, and emits ``components.json`` describing the kinds. + Excludes ``__pycache__`` / ``*.pyc``. + + Args: + folder (Path): The hub folder owning ``src/templates/app``. + dry_run (bool): Print the plan without executing. + """ + name = folder.name + src = folder / "src" / "templates" / "app" + if not src.is_dir(): + return + out_dir = folder / "mcp" / "templates" / "app" + if dry_run: + print(f"[TPL] {name} {src.relative_to(ROOT)}/** -> {out_dir.relative_to(ROOT)}/**") + return + + if out_dir.exists(): + shutil.rmtree(out_dir) + out_dir.parent.mkdir(parents=True, exist_ok=True) + shutil.copytree( + src, out_dir, + ignore=shutil.ignore_patterns("__pycache__", "*.pyc"), + ) + manifest = build_components_manifest(out_dir) + (folder / "mcp" / "components.json").write_text( + json.dumps(manifest, indent=2, ensure_ascii=False), encoding="utf-8" + ) + print(f"[TPL] {name} template vendored to {out_dir.relative_to(ROOT)}/ " + f"({len(manifest['components'])} components)") + + # ── CLI ────────────────────────────────────────────────────────────────────── @@ -186,6 +331,7 @@ def main() -> None: for folder in folders: build_bundle(folder, dry_run=args.dry_run) + build_templates(folder, dry_run=args.dry_run) build_html(folder, dry_run=args.dry_run) diff --git a/registry.py b/registry.py index 140b320..c95f374 100644 --- a/registry.py +++ b/registry.py @@ -1,20 +1,28 @@ """Standards MCP server. Serves the hand-authored `standards` bundle (the application blueprint plus the -library registry) and adds a discovery tool so an agent can find the right -internal library — and its MCP endpoint — before writing any infrastructure. +library registry and the runnable template) and adds discovery/scaffolding +helpers so an agent can find the right internal library — and pull exact +template code — before writing any infrastructure. Resources: - doc://index, doc://nav, doc://modules/{module} (from the base server) - registry://libraries (the structured catalog) + doc://index, doc://nav, doc://modules/{module} (from the base server) + registry://libraries (the structured catalog) + registry://components (the template component index) Tools: - ping (from the base server) - find_library (capability -> matching libraries + MCP endpoints) + ping (from the base server) + find_library (capability -> matching libraries + MCP endpoints) + list_app_components (the template component kinds) + get_app_component (kind -> exact template file, seed tokens renamed) + +Prompts: + start_app (collect the two axes, then emit the build prompt) """ from __future__ import annotations import json +import re from pathlib import Path from typing import Any @@ -22,35 +30,85 @@ from config import MCPBundle from mcp_server import MCPServer +def _pascal(name: str) -> str: + """Convert a snake/kebab name to PascalCase.""" + return "".join(part.capitalize() for part in re.split(r"[_\-\s]+", name) if part) + + class StandardsMCPServer(MCPServer): - """MCP server for the `standards` bundle with a library-discovery tool. + """MCP server for the `standards` bundle with discovery/scaffolding tools. Attributes: mcp_root (Path): Directory containing the generated bundle (index.json, nav.json, - modules/, and the machine-readable libraries.json). + modules/, libraries.json, components.json, and templates/). """ def __init__(self, bundle: MCPBundle) -> None: super().__init__(bundle) self._register_standards() + # ------------------------------------------------------------------ + # Bundle readers + # ------------------------------------------------------------------ + def _catalog(self) -> dict[str, Any]: """Load the machine-readable library registry.""" path = Path(self.mcp_root) / "libraries.json" if not path.exists(): - return { - "error": "not_found", - "path": str(path), - "libraries": [], - } + return {"error": "not_found", "path": str(path), "libraries": []} return json.loads(path.read_bytes()) + def _components(self) -> dict[str, Any]: + """Load the template component manifest.""" + path = Path(self.mcp_root) / "components.json" + if not path.exists(): + return {"error": "not_found", "path": str(path), "components": []} + return json.loads(path.read_bytes()) + + def _find_component(self, kind: str) -> dict[str, Any] | None: + for entry in self._components().get("components", []): + if entry.get("kind") == kind: + return entry + return None + + def _read_component( + self, + entry: dict[str, Any], + resource: str | None, + family: str | None, + ) -> dict[str, Any]: + """Read a vendored template file, renaming seed tokens when asked.""" + rel = entry["path"] + path = Path(self.mcp_root) / "templates" / "app" / rel + if not path.exists(): + return {"error": "not_found", "path": str(path)} + text = path.read_text(encoding="utf-8") + placeholder = entry.get("placeholder") + if resource and placeholder == "resource": + text = text.replace("Item", _pascal(resource)).replace("item", resource) + if family and placeholder == "family": + text = text.replace("Example", _pascal(family)).replace("example", family) + return { + "kind": entry["kind"], + "path": rel, + "placeholder": placeholder, + "content": text, + } + + # ------------------------------------------------------------------ + # Standards surface + # ------------------------------------------------------------------ + def _register_standards(self) -> None: @self.app.resource("registry://libraries") def libraries() -> dict[str, Any]: return self._catalog() + @self.app.resource("registry://components") + def components() -> dict[str, Any]: + return self._components() + @self.app.tool() def find_library(capability: str) -> list[dict[str, Any]]: """Find internal libraries matching a capability keyword. @@ -85,3 +143,96 @@ class StandardsMCPServer(MCPServer): if needle in haystack: matches.append(lib) return matches + + @self.app.tool() + def list_app_components() -> list[dict[str, Any]]: + """List the reusable component kinds shipped in the app template. + + Returns: + list[dict[str, Any]]: + Entries with ``kind``, ``path``, ``placeholder`` and + ``description``. Fetch one with :func:`get_app_component`. + """ + return self._components().get("components", []) + + @self.app.tool() + def get_app_component( + kind: str, + resource: str | None = None, + family: str | None = None, + ) -> dict[str, Any]: + """Return the exact template file for a component kind. + + Use this to copy current, app-neutral skeleton code instead of + inventing it. See :func:`list_app_components` for valid kinds. + + Args: + kind (str): Component kind, e.g. "routes", "pipeline-impl". + resource (str | None): Rename the resource seed ("item"/"Item") + to this name (snake_case), for per-resource kinds. + family (str | None): Rename the family seed ("example"/"Example") + to this name (snake_case), for per-family kinds. + + Returns: + dict[str, Any]: ``{kind, path, placeholder, content}`` or an + error entry when the kind is unknown. + """ + entry = self._find_component(kind) + if entry is None: + return { + "error": "unknown_kind", + "kind": kind, + "available": [c["kind"] for c in self._components().get("components", [])], + } + return self._read_component(entry, resource, family) + + @self.app.prompt() + def start_app( + app_name: str = "", + resources: str = "", + pipelines: str = "", + auth: str = "no", + ) -> str: + """Entry point to build a new app from the standards blueprint. + + Call with no arguments to get the intake questions; call again with + the answers to get the deterministic build prompt. + + Args: + app_name (str): App name + one-line purpose. + resources (str): Axis 1 — resources, fields, relationships + (or a pasted openapi.yaml). + pipelines (str): Axis 2 — pipeline families + input formats. + auth (str): Whether auth is needed now ("yes"/"no"). + + Returns: + str: The intake questions, or the build prompt when complete. + """ + if not app_name or not resources: + return ( + "# Start a new app — intake\n\n" + "Gather the following, then call `start_app` again with the values:\n\n" + "1. App name + one-line purpose.\n" + "2. Axis 1 — resources: each resource, its fields, and relationships\n" + " (or paste `openapi.yaml`).\n" + "3. Axis 2 — pipelines: multi-step inputs? families + formats (or `none`).\n" + "4. Auth needed now? (yes/no; auth is deferred — `py-jwt` only).\n" + "5. Integrations: Mongo / ingestion formats / email / LLM provider.\n\n" + "Then: `start_app(app_name=..., resources=..., pipelines=..., auth=...)`.\n" + ) + return ( + f"# Build: {app_name}\n\n" + "## Spec\n" + f"- resources: {resources}\n" + f"- pipelines: {pipelines or 'none'}\n" + f"- auth: {auth}\n\n" + "## Method (deterministic)\n" + "1. Read `doc://modules/build_recipe` and follow it step by step.\n" + "2. Read `doc://modules/genericity_matrix` (Tier A/B/C) and " + "`doc://modules/layering_rules`.\n" + "3. Call `find_library(capability)` for db, di, routing, ingestion, " + "and pipeline; never reimplement a Tier A library.\n" + "4. Pull exact files with `list_app_components()` and " + "`get_app_component(kind, resource=..., family=...)`.\n" + "5. Author only Tier C; keep Tier B as shipped.\n" + ) diff --git a/standards/lib/09_infra_adaptors/index.html b/standards/lib/09_infra_adaptors/index.html index 1cd4fbe..e6309eb 100644 --- a/standards/lib/09_infra_adaptors/index.html +++ b/standards/lib/09_infra_adaptors/index.html @@ -691,7 +691,9 @@

Infra Adapters

Summary

All external systems are reached through adapters in infra/. Prefer an -existing internal library over writing a new adapter.

+existing internal library over writing a new adapter; where no library exists, +copy the Tier B scaffold from templates/app. See the Genericity Matrix for the +Tier A / B / C split.

Database (MongoDB)

Read the Wiki for the philosophy and rationale; these pages are the diff --git a/standards/lib/search/search_index.json b/standards/lib/search/search_index.json index 7fd0e5e..38c20a2 100644 --- a/standards/lib/search/search_index.json +++ b/standards/lib/search/search_index.json @@ -1 +1 @@ -{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"Standards Reference","text":"

The reference half of the Standards description: the enforceable rules, patterns, and the internal library catalog that agents consult while coding.

Read the Wiki for the philosophy and rationale; these pages are the reference that mirrors the standards MCP bundle.

"},{"location":"02_layering_rules/","title":"Layering Rules","text":""},{"location":"02_layering_rules/#summary","title":"Summary","text":"

The dependency direction is fixed. Violating it is a design defect.

"},{"location":"02_layering_rules/#allowed-imports","title":"Allowed imports","text":"
interfaces/*      ->  domain/*            (services, ports, models)\ninterfaces/*      ->  infra/adapters/*    (only read-only value objects)\ndomain/*          ->  domain/*            (same layer)\ndomain/*          ->  pipelines/*         (call a pipeline via a port)\ndomain/*/ports/*  ->  infra/*             (bind adapters + libraries at the seam)\ninfra/adapters/*  ->  domain/ports/*      (implement the interface)\nmain.py/init.py   ->  everything          (composition root only)\n
"},{"location":"02_layering_rules/#forbidden-imports","title":"Forbidden imports","text":"
domain/*          ->  interfaces/*        (never)\ndomain/*          ->  fastapi/starlette   (never)\ninfra/adapters/*  ->  interfaces/*        (never)\ninterfaces/*      ->  infra/db/*          (state lives in resources, not imports)\n
"},{"location":"02_layering_rules/#why","title":"Why","text":""},{"location":"02_layering_rules/#the-portadapter-contract","title":"The port/adapter contract","text":"

The ports/ package is the seam where the domain meets infrastructure. Each domain/<resource>/ports/ module declares the resource's models/document shape and exposes factory functions that bind a concrete adapter \u2014 so a port module is allowed to import from infra/ and from the persistence libraries. Everything outside ports/ in domain/ stays framework- and driver-free.

domain/<resource>/ports/db.py       -> document model + create_<resource>_repo()\ndomain/<resource>/ports/service.py  -> service + get_<resource>_service()\ninfra/db/repos/<vendor>.py          -> repository adapter factory\ninfra/adapters/service.py           -> generic CRUD service base\n

The composition root calls the port factories; nothing else constructs an adapter.

"},{"location":"02_layering_rules/#enforcement","title":"Enforcement","text":""},{"location":"03_composition_root/","title":"Composition Root","text":""},{"location":"03_composition_root/#summary","title":"Summary","text":"

All wiring happens in exactly one module. It builds a flat resource registry and exposes it to the HTTP layer and to scripts.

"},{"location":"03_composition_root/#the-resource-registry","title":"The resource registry","text":"

init.py returns a plain dict[str, Any]. Keys follow two conventions:

Plus named infrastructure adaptors:

The API stores the dict on app.state.db_resources during lifespan startup. Handlers read from it; nothing imports concrete adapters.

"},{"location":"03_composition_root/#shape","title":"Shape","text":"
_resources: dict = {}\n_initialized = False\n\nasync def init(settings=None) -> dict:\n    global _initialized\n    if _initialized:\n        return _resources\n\n    await setup_database(uri=settings.mongo_uri, db_name=settings.mongo_db)\n\n    # 1. leaf repositories\n    user_repo = create_user_repo()\n\n    # 2. dependent repositories (dependencies passed by keyword)\n    wallet_repo = create_wallet_repo(user_repo=user_repo)\n\n    # 3. services (build via the DI library, injecting the repo)\n    _resources[\"user_repo\"]   = user_repo\n    _resources[\"wallet_repo\"] = wallet_repo\n    _resources[\"user_svc\"]    = get_user_service(user_repo)\n    _resources[\"wallet_svc\"]  = get_wallet_service(wallet_repo)\n\n    # 4. infrastructure adaptors (optional)\n    _resources[\"http_cache_store\"] = HttpCacheStore(...)\n    _resources[\"progress_bus\"]     = ProgressEventBus()\n\n    _initialized = True\n    return _resources\n\nasync def shutdown() -> None:\n    global _initialized\n    await close_stores(_resources)\n    await shutdown_database()\n    _resources.clear()\n    _initialized = False\n
"},{"location":"03_composition_root/#typed-accessors","title":"Typed accessors","text":"

Expose a get_svc(name) helper with one @overload per key so scripts get typing without importing concrete classes:

@overload\ndef get_svc(name: Literal[\"user_svc\"]) -> UserService: ...\n@overload\ndef get_svc(name: Literal[\"wallet_svc\"]) -> WalletService: ...\n\ndef get_svc(name: str):\n    return _resources[name]\n

Also expose get_db() (raw database handle) and pure helpers such as compute_fingerprint(payload) (canonical hash used for deduplication).

"},{"location":"03_composition_root/#rules","title":"Rules","text":""},{"location":"04_spec_first_routing/","title":"Spec-First Routing","text":""},{"location":"04_spec_first_routing/#summary","title":"Summary","text":"

The OpenAPI document is the single source of truth for the HTTP contract. Handlers are bound to it by name; there is no decorator router.

"},{"location":"04_spec_first_routing/#the-contract","title":"The contract","text":""},{"location":"04_spec_first_routing/#app-bootstrap","title":"App bootstrap","text":"
# main.py\nfrom openapi_first.app import OpenAPIFirstApp\n\napp = OpenAPIFirstApp(\n    openapi_path=\"openapi.yaml\",\n    routes_module=interfaces.openapi,   # module whose namespace holds handlers\n    title=\"<App> API\",\n    version=\"0.1.0\",\n    lifespan=lifespan,\n)\n

lifespan calls the composition root:

@asynccontextmanager\nasync def lifespan(app):\n    app.state.db_resources = await init()\n    yield\n    await shutdown()\n
"},{"location":"04_spec_first_routing/#handler-registry","title":"Handler registry","text":"

interfaces/openapi.py is a flat re-export of every handler, so the framework can resolve operationId across the whole app:

from interfaces.item.routes import *\nfrom interfaces.wallet.routes import *\nfrom interfaces.progress.sse import *\n
"},{"location":"04_spec_first_routing/#handler-shape","title":"Handler shape","text":"

Handlers are thin and synchronous with respect to business logic:

async def create_item(request: Request, body: dict):\n    svc = request.app.state.db_resources[\"item_svc\"]\n    return await svc.create(Item.model_validate(body))\n
"},{"location":"04_spec_first_routing/#conventions","title":"Conventions","text":""},{"location":"04_spec_first_routing/#codegen","title":"Codegen","text":"

The openapi_first library can scaffold models and route stubs from the spec, and ships app templates (CRUD app, model app, service). Prefer codegen over hand-writing boilerplate.

"},{"location":"05_crud_template/","title":"CRUD Template","text":""},{"location":"05_crud_template/#summary","title":"Summary","text":"

Every resource follows the same four-file slice. Repetition is intentional: it makes the codebase predictable and generatable.

"},{"location":"05_crud_template/#the-slice","title":"The slice","text":"
domain/<resource>/\n    models.py               # pure Pydantic model (no DB fields)\n    ports/\n        db.py               # document model + repository factory\n        service.py          # service class + factory\n        converters.py       # (optional) transforms\n
"},{"location":"05_crud_template/#modelspy","title":"models.py","text":"

A framework-free Pydantic model describing the business shape.

class Item(BaseModel):\n    name: str\n    quantity: int = 0\n
"},{"location":"05_crud_template/#portsdbpy","title":"ports/db.py","text":"

The persistence model extends the business model and adds database concerns. The repository factory selects the right repository flavour.

class ItemDocument(Item, BaseDocument):   # BaseDocument adds id/created_at/...\n    pass\n\ndef create_item_repo(*, owner_repo=None) -> ItemRepository:\n    return repo_factory(\n        collection=\"items\",\n        document=ItemDocument,\n        populate_rules=[PopulateRule(field=\"owner\", collection=\"users\")],\n        cache=None,               # or a CachedBaseRepository config\n    )\n
"},{"location":"05_crud_template/#portsservicepy","title":"ports/service.py","text":"

The service is built through the DI library, injecting the repository. The generic CRUD base provides create/read/update/patch/delete/search/....

class ItemService(CrudsServiceAdaptor[Item, ItemDocument]):\n    model = Item\n    schema = ItemDocument\n\ndef get_item_service(repo) -> ItemService:\n    return build(ItemService, instances={\"repo\": repo})\n
"},{"location":"05_crud_template/#naming-conventions","title":"Naming conventions","text":"Concept Convention collection plural snake case (items) document <Resource>Document repository factory create_<resource>_repo service <Resource>Service service factory get_<resource>_service resource key <resource>_svc operationId list_items, create_item, ..."},{"location":"05_crud_template/#optional-files","title":"Optional files","text":""},{"location":"06_pipeline_scaffold/","title":"Pipeline Scaffold","text":""},{"location":"06_pipeline_scaffold/#summary","title":"Summary","text":"

Data-driven processing is expressed as a composition graph. The graph is built by a DI library; execution is driven by a DAG engine. Stages are pure and individually testable.

"},{"location":"06_pipeline_scaffold/#when-to-use-a-pipeline","title":"When to use a pipeline","text":"

Use a pipeline when a resource is produced by a multi-step transformation of an input (a document, a file, a message) rather than by a simple CRUD write.

"},{"location":"06_pipeline_scaffold/#structure-of-a-family","title":"Structure of a family","text":"

A family is one input shape (e.g. a file format or a message kind). Each family has the same skeleton:

pipelines/<family>/\n    abc.py            # port contracts (stage interfaces + root interface)\n    impl.py           # concrete stages + root implementation\n    pipeline.yaml     # graph spec: nodes, edges, roots, schema\n    resources/\n        pipeline.yaml # runtime spec consumed by the loader\n        patterns/*.yaml  # rule fragments (pattern registry)\n    banks/<variant>.py   # optional variants that re-pin a few slots\n
"},{"location":"06_pipeline_scaffold/#shared-scaffolding","title":"Shared scaffolding","text":"
pipelines/shared/\n    ports.py     # RegistryPort, FinalizePort, etc.\n    loader.py    # build(root_cls) -> wires the graph + engine\n    registry.py  # resolve_root(...) -> root class for a (kind, variant)\n
"},{"location":"06_pipeline_scaffold/#the-loader","title":"The loader","text":"
def build(root_cls, *, instances=None, config=None):\n    # merge resources/patterns/*.yaml into a pattern registry\n    # bind registry/config into each graph node by id\n    # build the schema (state) class and the engine\n    root = root_cls()\n    root._engine = Engine(graph, state_cls, payload_cls)\n    return root\n
"},{"location":"06_pipeline_scaffold/#the-registry","title":"The registry","text":"

A single lookup replaces hand-rolled if/else selection:

_FAMILIES = {\n    \"csv\":  {\"acme\": {\"heuristic\": AcmeCsvPipeline}},\n    \"json\": {\"globex\": {\"heuristic\": GlobexJsonPipeline}},\n}\n\ndef resolve_root(kind, variant, mode=\"heuristic\"):\n    return _FAMILIES[kind][variant][mode]\n

Adding a new variant means adding one class and one registry entry.

"},{"location":"06_pipeline_scaffold/#driving-a-pipeline","title":"Driving a pipeline","text":"

Stages are driven explicitly so intermediate state can be persisted:

pipeline = build(resolve_root(kind, variant))\ncontent = pipeline.ingest.materialize(raw_bytes)\nrecords = pipeline.reader.read(content)\nitems   = pipeline.normalize.build(records)\n
"},{"location":"06_pipeline_scaffold/#testing","title":"Testing","text":"

Because stages are pure, test each stage in isolation and test the assembled root against fixture inputs whose expected outputs are stored as golden files. A parity test replays fixtures headlessly.

"},{"location":"07_worker_and_progress/","title":"Worker & Progress","text":""},{"location":"07_worker_and_progress/#summary","title":"Summary","text":"

Long-running work runs in a background worker. Jobs are durable documents that move through an explicit status machine. Progress is streamed to clients.

"},{"location":"07_worker_and_progress/#the-status-machine","title":"The status machine","text":"
pending -> processing -> completed\n                     \\-> failed    (retried up to N times)\n                     \\-> paused    (waiting for human input)\n
"},{"location":"07_worker_and_progress/#the-poll-loop","title":"The poll loop","text":"
async def poll(resources):\n    backoff = MIN_POLL\n    while True:\n        jobs = await job_svc.search(filter={\"status\": \"pending\"}, limit=100)\n        backoff = MIN_POLL if jobs else min(backoff * 2, MAX_POLL)\n        for job in jobs:\n            if (job.retry_count or 0) >= MAX_RETRIES:\n                continue\n            await job_svc.update_status(job.id, \"processing\")\n            try:\n                await process(job, resources)\n            except PauseForResolution:\n                await job_svc.update_status(job.id, \"paused\")\n            except Exception as e:\n                await job_svc.update_status(\n                    job.id, \"failed\", error=str(e),\n                    retry_count=(job.retry_count or 0) + 1,\n                )\n        await asyncio.sleep(backoff)\n

Tunables live in environment variables: WORKER_POLL_MIN, WORKER_POLL_MAX, WORKER_MAX_RETRIES.

"},{"location":"07_worker_and_progress/#progress-events","title":"Progress events","text":"

A worker publishes structured progress events to a bus (a port). The HTTP edge exposes a streaming endpoint that replays events for a given job id.

class ProgressEvent(BaseModel):\n    job_id: str\n    step: str\n    status: str            # started | progress | completed | skipped | paused\n    message: dict | None = None\n

The bus is an in-memory adapter by default; because it is a port, it can be swapped for a durable broker. Emit from the orchestrator at each stage:

await self.bus.publish(ProgressEvent(job_id=jid, step=\"normalize\",\n                                     status=\"completed\", message={\"count\": n}))\n
"},{"location":"07_worker_and_progress/#streaming-endpoint","title":"Streaming endpoint","text":"
async def stream_job_events(request, id):\n    bus = request.app.state.db_resources.get(\"progress_bus\")\n    if bus is None:\n        return {\"events\": []}\n\n    async def gen():\n        sent = 0\n        while True:\n            events = bus.get_events(id)\n            for e in events[sent:]:\n                yield f\"data: {e.model_dump_json(exclude_none=True)}\\n\\n\"\n                sent += 1\n            await asyncio.sleep(0.5)\n\n    return StreamingResponse(gen(), media_type=\"text/event-stream\")\n

Known limitation. The default in-memory bus is per-process. If the worker and the API run as separate processes, events published by the worker are not visible to the API. For production, back the bus with a shared broker. The port boundary makes this a one-adapter change.

"},{"location":"08_human_in_loop/","title":"Human-in-the-Loop","text":""},{"location":"08_human_in_loop/#summary","title":"Summary","text":"

Some inputs are ambiguous and must be resolved by a human before processing can continue. The blueprint models this as a durable pause: an exception is raised, a question is recorded, the job pauses, and later resumes.

"},{"location":"08_human_in_loop/#the-mechanism","title":"The mechanism","text":"
  1. During processing, a stage raises a domain exception carrying the context needed to ask the question:
raise PauseForResolution(\n    job_id=jid,\n    reference=\"<stable key for this decision>\",\n    options=[...],\n    context={...},\n)\n
  1. The orchestrator catches it, records a question document, and re-raises so the worker marks the job paused.
  2. A client lists pending questions and answers one:
GET  /jobs/{id}/questions            -> list pending questions\nPOST /questions/{qid}/resolve  {choice}  -> record the choice\n
  1. When the last pending question is resolved, the resolve endpoint flips the parent job back to pending.
  2. The worker picks it up again, rebuilds a resolution map from previously answered questions, and seeds it into the pipeline so the same decision is not asked twice.
"},{"location":"08_human_in_loop/#the-question-document","title":"The question document","text":"
class Question(BaseDocument):\n    job_id: str\n    reference: str            # stable key: stage|field|value\n    status: str               # pending | resolved\n    options: list[dict]\n    context: dict\n    chosen: dict | None = None\n    resolved_at: str | None = None\n
"},{"location":"08_human_in_loop/#resume-map","title":"Resume map","text":"

The worker turns answered questions into a lookup keyed by reference, then injects it into the pipeline so that stage can skip re-raising:

resolved = {q.reference: q.chosen for q in await svc.search(\n    filter={\"job_id\": jid, \"status\": \"resolved\"})}\n\npipeline = build(root, instances={\"pause\": handler}, config={\"resolved\": resolved})\n
"},{"location":"08_human_in_loop/#resolve-endpoint","title":"Resolve endpoint","text":"
async def resolve_question(request, id, body):\n    svc = request.app.state.db_resources[\"question_svc\"]\n    q = await svc.read(id)\n    await svc.update(id, {\"chosen\": body[\"chosen\"], \"status\": \"resolved\",\n                          \"resolved_at\": now()})\n\n    pending = await svc.search(filter={\"job_id\": q.job_id, \"status\": \"pending\"},\n                               limit=1)\n    if not pending:\n        jobs = request.app.state.db_resources[\"job_svc\"]\n        job = await jobs.read(q.job_id)\n        if job and job.status == \"paused\":\n            await jobs.update_status(q.job_id, \"pending\")\n    return q\n
"},{"location":"08_human_in_loop/#cli-parity","title":"CLI parity","text":"

The same flow is available through scripts/CLI: list, get, create, questions, resolve, retry.

"},{"location":"09_infra_adaptors/","title":"Infra Adapters","text":""},{"location":"09_infra_adaptors/#summary","title":"Summary","text":"

All external systems are reached through adapters in infra/. Prefer an existing internal library over writing a new adapter.

"},{"location":"09_infra_adaptors/#database-mongodb","title":"Database (MongoDB)","text":" 
repo_factory(collection, document, engine=None,\n             populate_rules=None, cache=None)\n
PopulateRule(field=\"owner\", collection=\"users\",\n             nested_rules=[], max_depth=2, projection=None)\n

Library: the Mongo operations library (see registry). Do not write raw Motor calls in the domain.

"},{"location":"09_infra_adaptors/#http-response-cache","title":"HTTP response cache","text":"

A generic GET cache sits behind a decorator.

@cache_response(ttl=60)\nasync def list_items(request): ...\n
"},{"location":"09_infra_adaptors/#event-progress-bus","title":"Event / progress bus","text":"

A port with an in-memory default adapter (publish, get_events). Swap for a durable broker in production (see Worker & Progress).

"},{"location":"09_infra_adaptors/#external-clients-ingestion","title":"External clients (ingestion)","text":"

External fetchers (files, mail, HTTP) are adapters that wrap ingestion libraries. Construct them in the composition root and inject into orchestrators.

Libraries: the ingestion library (formats) and the mail library (email).

"},{"location":"09_infra_adaptors/#llm-clients","title":"LLM clients","text":"

When a pipeline calls a model provider, route through a pool adaptor that handles provider selection, retries, and client lifecycle:

pool = setup_async_pool(provider=settings.llm_provider)\nresult = await pool.complete(model=\"...\", messages=[...])\n

Providers are selected by LLM_PROVIDER and configured by environment.

"},{"location":"09_infra_adaptors/#key-value-store","title":"Key-value store","text":"

For report/snapshot storage, provide a pluggable store port with a Redis adapter and an in-memory fallback.

"},{"location":"09_infra_adaptors/#rule","title":"Rule","text":"

If a capability already exists as a library (DB, DI, ingestion, mail, auth, docs), use it. Writing a private copy of a library is a defect.

"},{"location":"12_genericity_matrix/","title":"Genericity Matrix","text":""},{"location":"12_genericity_matrix/#summary","title":"Summary","text":"

This page states precisely what is reusable as-is, what is a template to adapt, and what is app-specific and must be authored. Use it to scope a new build.

"},{"location":"12_genericity_matrix/#reusable-as-is-do-not-rewrite","title":"Reusable as-is (do not rewrite)","text":"Area Artifact Layering & dependency rule the package layout and import rules Composition root the init.py shape, idempotency, get_svc overloading Web edge OpenAPIFirstApp bootstrap, handler-name = operationId, thin handlers CRUD slice the four-file pattern (model / ports-db / ports-service / converters) Generic CRUD service the base service adaptor Repository factory base / populating / cached selection Worker the poll loop, backoff, retry, status machine Progress bus the bus port + in-memory adapter + streaming endpoint Human-in-the-loop the durable pause + question document + resolve endpoint + resume map Infra adaptors HTTP cache decorator, KV store port, LLM pool scaffold Testing integration harness with fakes, fixture-parity pattern Docs GSDFC + generated stubs + MCP bundle pipeline"},{"location":"12_genericity_matrix/#template-to-adapt-copy-then-change-names","title":"Template to adapt (copy, then change names)","text":"Artifact Change pyproject.toml project name, dependencies Dockerfile / docker-compose.yaml service name, ports, services .env.example connection strings, provider keys openapi.yaml every path/schema/operationId scripts/ seed and operator tasks"},{"location":"12_genericity_matrix/#app-specific-must-be-authored","title":"App-specific (must be authored)","text":"Artifact Why Resource models the business shape Repository populate/cache rules the relationships Service business logic domain behaviour Pipeline families (if any) the transformation graph Pattern/rule resources data-driven policy Fixtures & golden outputs test data"},{"location":"12_genericity_matrix/#the-two-axes","title":"The two axes","text":"

Everything app-specific reduces to two axes:

  1. Resource model \u2014 the set of resources, their fields, relationships, and the OpenAPI contract.
  2. Pipeline family \u2014 only if inputs require multi-step transformation.

If you can specify those two axes, the rest of the app follows mechanically.

"},{"location":"13_library_registry/","title":"Library Registry","text":""},{"location":"13_library_registry/#summary","title":"Summary","text":"

The Aetos ecosystem provides internal libraries for the shared concerns of every application. Use them. Writing a private copy of any capability below is a defect.

Each library also exposes its own documentation as an MCP server. An AI agent can read a library's full API by connecting to its MCP and reading doc://index, doc://nav, and doc://modules/{module}.

"},{"location":"13_library_registry/#catalog","title":"Catalog","text":"Library Import Purpose Use instead of MCP DAG Pipe dagpipe Deterministic state-transformation DAG engine hand-rolled orchestration/topological ordering :8006 Hexa hexa Declarative DI / composition-root tooling manual wiring, hand-written factories (see note) Mongo Ops mongo_ops Async MongoDB repositories, cache, populate, transactions raw Motor CRUD and connection lifecycle :8007 Omniread omniread Format-agnostic fetch + parse (HTML, PDF, CSV, XLSX) custom scrapers/parsers :8003 OpenAPI First openapi_first Spec-first FastAPI bootstrap, client, codegen decorator routers and hand-written clients :8004 Mail Intake mail_intake Provider-agnostic read-only email ingestion Gmail OAuth + MIME parsing :8002 Py JWT jwtlib JWT auth primitives (register/login/introspect) password hashing + token code :8005 Doc Forge docforge Docstring compiler \u2192 MkDocs + MCP docs custom doc generators :8001 Auth Server (service) Central identity service (register/login/me/introspect) building auth endpoints yourself \u2014"},{"location":"13_library_registry/#how-to-choose","title":"How to choose","text":"You need to\u2026 Use persist and query documents Mongo Ops compose dependencies Hexa run a multi-step transformation DAG Pipe read/parse an arbitrary input format Omniread expose a contract-first HTTP API OpenAPI First ingest email Mail Intake authenticate users / verify tokens Py JWT (+ Auth Server) generate documentation Doc Forge"},{"location":"13_library_registry/#discovery","title":"Discovery","text":"

A single standards MCP exposes find_library(capability), which returns the matching library, its import name, its \"use instead of\" guidance, and its MCP endpoint. Query it before writing any infrastructure.

"},{"location":"13_library_registry/#notes","title":"Notes","text":""}]} \ No newline at end of file +{"config":{"lang":["en"],"separator":"[\\s\\-]+","pipeline":["stopWordFilter"]},"docs":[{"location":"","title":"Standards Reference","text":"

The reference half of the Standards description: the enforceable rules, patterns, and the internal library catalog that agents consult while coding.

Read the Wiki for the philosophy and rationale; these pages are the reference that mirrors the standards MCP bundle.

"},{"location":"02_layering_rules/","title":"Layering Rules","text":""},{"location":"02_layering_rules/#summary","title":"Summary","text":"

The dependency direction is fixed. Violating it is a design defect.

"},{"location":"02_layering_rules/#allowed-imports","title":"Allowed imports","text":"
interfaces/*      ->  domain/*            (services, ports, models)\ninterfaces/*      ->  infra/adapters/*    (only read-only value objects)\ndomain/*          ->  domain/*            (same layer)\ndomain/*          ->  pipelines/*         (call a pipeline via a port)\ndomain/*/ports/*  ->  infra/*             (bind adapters + libraries at the seam)\ninfra/adapters/*  ->  domain/ports/*      (implement the interface)\nmain.py/init.py   ->  everything          (composition root only)\n
"},{"location":"02_layering_rules/#forbidden-imports","title":"Forbidden imports","text":"
domain/*          ->  interfaces/*        (never)\ndomain/*          ->  fastapi/starlette   (never)\ninfra/adapters/*  ->  interfaces/*        (never)\ninterfaces/*      ->  infra/db/*          (state lives in resources, not imports)\n
"},{"location":"02_layering_rules/#why","title":"Why","text":""},{"location":"02_layering_rules/#the-portadapter-contract","title":"The port/adapter contract","text":"

The ports/ package is the seam where the domain meets infrastructure. Each domain/<resource>/ports/ module declares the resource's models/document shape and exposes factory functions that bind a concrete adapter \u2014 so a port module is allowed to import from infra/ and from the persistence libraries. Everything outside ports/ in domain/ stays framework- and driver-free.

domain/<resource>/ports/db.py       -> document model + create_<resource>_repo()\ndomain/<resource>/ports/service.py  -> service + get_<resource>_service()\ninfra/db/repos/<vendor>.py          -> repository adapter factory\ninfra/adapters/service.py           -> generic CRUD service base\n

The composition root calls the port factories; nothing else constructs an adapter.

"},{"location":"02_layering_rules/#enforcement","title":"Enforcement","text":""},{"location":"03_composition_root/","title":"Composition Root","text":""},{"location":"03_composition_root/#summary","title":"Summary","text":"

All wiring happens in exactly one module. It builds a flat resource registry and exposes it to the HTTP layer and to scripts.

"},{"location":"03_composition_root/#the-resource-registry","title":"The resource registry","text":"

init.py returns a plain dict[str, Any]. Keys follow two conventions:

Plus named infrastructure adaptors:

The API stores the dict on app.state.db_resources during lifespan startup. Handlers read from it; nothing imports concrete adapters.

"},{"location":"03_composition_root/#shape","title":"Shape","text":"
_resources: dict = {}\n_initialized = False\n\nasync def init(settings=None) -> dict:\n    global _initialized\n    if _initialized:\n        return _resources\n\n    await setup_database(uri=settings.mongo_uri, db_name=settings.mongo_db)\n\n    # 1. leaf repositories\n    user_repo = create_user_repo()\n\n    # 2. dependent repositories (dependencies passed by keyword)\n    wallet_repo = create_wallet_repo(user_repo=user_repo)\n\n    # 3. services (build via the DI library, injecting the repo)\n    _resources[\"user_repo\"]   = user_repo\n    _resources[\"wallet_repo\"] = wallet_repo\n    _resources[\"user_svc\"]    = get_user_service(user_repo)\n    _resources[\"wallet_svc\"]  = get_wallet_service(wallet_repo)\n\n    # 4. infrastructure adaptors (optional)\n    _resources[\"http_cache_store\"] = HttpCacheStore(...)\n    _resources[\"progress_bus\"]     = ProgressEventBus()\n\n    _initialized = True\n    return _resources\n\nasync def shutdown() -> None:\n    global _initialized\n    await close_stores(_resources)\n    await shutdown_database()\n    _resources.clear()\n    _initialized = False\n
"},{"location":"03_composition_root/#typed-accessors","title":"Typed accessors","text":"

Expose a get_svc(name) helper with one @overload per key so scripts get typing without importing concrete classes:

@overload\ndef get_svc(name: Literal[\"user_svc\"]) -> UserService: ...\n@overload\ndef get_svc(name: Literal[\"wallet_svc\"]) -> WalletService: ...\n\ndef get_svc(name: str):\n    return _resources[name]\n

Also expose get_db() (raw database handle) and pure helpers such as compute_fingerprint(payload) (canonical hash used for deduplication).

"},{"location":"03_composition_root/#rules","title":"Rules","text":""},{"location":"04_spec_first_routing/","title":"Spec-First Routing","text":""},{"location":"04_spec_first_routing/#summary","title":"Summary","text":"

The OpenAPI document is the single source of truth for the HTTP contract. Handlers are bound to it by name; there is no decorator router.

"},{"location":"04_spec_first_routing/#the-contract","title":"The contract","text":""},{"location":"04_spec_first_routing/#app-bootstrap","title":"App bootstrap","text":"
# main.py\nfrom openapi_first.app import OpenAPIFirstApp\n\napp = OpenAPIFirstApp(\n    openapi_path=\"openapi.yaml\",\n    routes_module=interfaces.openapi,   # module whose namespace holds handlers\n    title=\"<App> API\",\n    version=\"0.1.0\",\n    lifespan=lifespan,\n)\n

lifespan calls the composition root:

@asynccontextmanager\nasync def lifespan(app):\n    app.state.db_resources = await init()\n    yield\n    await shutdown()\n
"},{"location":"04_spec_first_routing/#handler-registry","title":"Handler registry","text":"

interfaces/openapi.py is a flat re-export of every handler, so the framework can resolve operationId across the whole app:

from interfaces.item.routes import *\nfrom interfaces.wallet.routes import *\nfrom interfaces.progress.sse import *\n
"},{"location":"04_spec_first_routing/#handler-shape","title":"Handler shape","text":"

Handlers are thin and synchronous with respect to business logic:

async def create_item(request: Request, body: dict):\n    svc = request.app.state.db_resources[\"item_svc\"]\n    return await svc.create(Item.model_validate(body))\n
"},{"location":"04_spec_first_routing/#conventions","title":"Conventions","text":""},{"location":"04_spec_first_routing/#codegen","title":"Codegen","text":"

The openapi_first library can scaffold models and route stubs from the spec, and ships app templates (CRUD app, model app, service). Prefer codegen over hand-writing boilerplate.

"},{"location":"05_crud_template/","title":"CRUD Template","text":""},{"location":"05_crud_template/#summary","title":"Summary","text":"

Every resource follows the same four-file slice. Repetition is intentional: it makes the codebase predictable and generatable.

"},{"location":"05_crud_template/#the-slice","title":"The slice","text":"
domain/<resource>/\n    models.py               # pure Pydantic model (no DB fields)\n    ports/\n        db.py               # document model + repository factory\n        service.py          # service class + factory\n        converters.py       # (optional) transforms\n
"},{"location":"05_crud_template/#modelspy","title":"models.py","text":"

A framework-free Pydantic model describing the business shape.

class Item(BaseModel):\n    name: str\n    quantity: int = 0\n
"},{"location":"05_crud_template/#portsdbpy","title":"ports/db.py","text":"

The persistence model extends the business model and adds database concerns. The repository factory selects the right repository flavour.

class ItemDocument(Item, BaseDocument):   # BaseDocument adds id/created_at/...\n    pass\n\ndef create_item_repo(*, owner_repo=None) -> ItemRepository:\n    return repo_factory(\n        collection=\"items\",\n        document=ItemDocument,\n        populate_rules=[PopulateRule(field=\"owner\", collection=\"users\")],\n        cache=None,               # or a CachedBaseRepository config\n    )\n
"},{"location":"05_crud_template/#portsservicepy","title":"ports/service.py","text":"

The service is built through the DI library, injecting the repository. The generic CRUD base provides create/read/update/patch/delete/search/....

class ItemService(CrudsServiceAdaptor[Item, ItemDocument]):\n    model = Item\n    schema = ItemDocument\n\ndef get_item_service(repo) -> ItemService:\n    return build(ItemService, instances={\"repo\": repo})\n
"},{"location":"05_crud_template/#naming-conventions","title":"Naming conventions","text":"Concept Convention collection plural snake case (items) document <Resource>Document repository factory create_<resource>_repo service <Resource>Service service factory get_<resource>_service resource key <resource>_svc operationId list_items, create_item, ..."},{"location":"05_crud_template/#optional-files","title":"Optional files","text":""},{"location":"06_pipeline_scaffold/","title":"Pipeline Scaffold","text":""},{"location":"06_pipeline_scaffold/#summary","title":"Summary","text":"

Data-driven processing is expressed as a composition graph. The graph is built by a DI library; execution is driven by a DAG engine. Stages are pure and individually testable.

"},{"location":"06_pipeline_scaffold/#when-to-use-a-pipeline","title":"When to use a pipeline","text":"

Use a pipeline when a resource is produced by a multi-step transformation of an input (a document, a file, a message) rather than by a simple CRUD write.

"},{"location":"06_pipeline_scaffold/#structure-of-a-family","title":"Structure of a family","text":"

A family is one input shape (e.g. a file format or a message kind). Each family has the same skeleton:

pipelines/<family>/\n    abc.py            # port contracts (stage interfaces + root interface)\n    impl.py           # concrete stages + root implementation\n    pipeline.yaml     # graph spec: nodes, edges, roots, schema\n    resources/\n        pipeline.yaml # runtime spec consumed by the loader\n        patterns/*.yaml  # rule fragments (pattern registry)\n    banks/<variant>.py   # optional variants that re-pin a few slots\n
"},{"location":"06_pipeline_scaffold/#shared-scaffolding","title":"Shared scaffolding","text":"
pipelines/shared/\n    ports.py     # RegistryPort, FinalizePort, etc.\n    loader.py    # build(root_cls) -> wires the graph + engine\n    registry.py  # resolve_root(...) -> root class for a (kind, variant)\n
"},{"location":"06_pipeline_scaffold/#the-loader","title":"The loader","text":"
def build(root_cls, *, instances=None, config=None):\n    # merge resources/patterns/*.yaml into a pattern registry\n    # bind registry/config into each graph node by id\n    # build the schema (state) class and the engine\n    root = root_cls()\n    root._engine = Engine(graph, state_cls, payload_cls)\n    return root\n
"},{"location":"06_pipeline_scaffold/#the-registry","title":"The registry","text":"

A single lookup replaces hand-rolled if/else selection:

_FAMILIES = {\n    \"csv\":  {\"acme\": {\"heuristic\": AcmeCsvPipeline}},\n    \"json\": {\"globex\": {\"heuristic\": GlobexJsonPipeline}},\n}\n\ndef resolve_root(kind, variant, mode=\"heuristic\"):\n    return _FAMILIES[kind][variant][mode]\n

Adding a new variant means adding one class and one registry entry.

"},{"location":"06_pipeline_scaffold/#driving-a-pipeline","title":"Driving a pipeline","text":"

Stages are driven explicitly so intermediate state can be persisted:

pipeline = build(resolve_root(kind, variant))\ncontent = pipeline.ingest.materialize(raw_bytes)\nrecords = pipeline.reader.read(content)\nitems   = pipeline.normalize.build(records)\n
"},{"location":"06_pipeline_scaffold/#testing","title":"Testing","text":"

Because stages are pure, test each stage in isolation and test the assembled root against fixture inputs whose expected outputs are stored as golden files. A parity test replays fixtures headlessly.

"},{"location":"07_worker_and_progress/","title":"Worker & Progress","text":""},{"location":"07_worker_and_progress/#summary","title":"Summary","text":"

Long-running work runs in a background worker. Jobs are durable documents that move through an explicit status machine. Progress is streamed to clients.

"},{"location":"07_worker_and_progress/#the-status-machine","title":"The status machine","text":"
pending -> processing -> completed\n                     \\-> failed    (retried up to N times)\n                     \\-> paused    (waiting for human input)\n
"},{"location":"07_worker_and_progress/#the-poll-loop","title":"The poll loop","text":"
async def poll(resources):\n    backoff = MIN_POLL\n    while True:\n        jobs = await job_svc.search(filter={\"status\": \"pending\"}, limit=100)\n        backoff = MIN_POLL if jobs else min(backoff * 2, MAX_POLL)\n        for job in jobs:\n            if (job.retry_count or 0) >= MAX_RETRIES:\n                continue\n            await job_svc.update_status(job.id, \"processing\")\n            try:\n                await process(job, resources)\n            except PauseForResolution:\n                await job_svc.update_status(job.id, \"paused\")\n            except Exception as e:\n                await job_svc.update_status(\n                    job.id, \"failed\", error=str(e),\n                    retry_count=(job.retry_count or 0) + 1,\n                )\n        await asyncio.sleep(backoff)\n

Tunables live in environment variables: WORKER_POLL_MIN, WORKER_POLL_MAX, WORKER_MAX_RETRIES.

"},{"location":"07_worker_and_progress/#progress-events","title":"Progress events","text":"

A worker publishes structured progress events to a bus (a port). The HTTP edge exposes a streaming endpoint that replays events for a given job id.

class ProgressEvent(BaseModel):\n    job_id: str\n    step: str\n    status: str            # started | progress | completed | skipped | paused\n    message: dict | None = None\n

The bus is an in-memory adapter by default; because it is a port, it can be swapped for a durable broker. Emit from the orchestrator at each stage:

await self.bus.publish(ProgressEvent(job_id=jid, step=\"normalize\",\n                                     status=\"completed\", message={\"count\": n}))\n
"},{"location":"07_worker_and_progress/#streaming-endpoint","title":"Streaming endpoint","text":"
async def stream_job_events(request, id):\n    bus = request.app.state.db_resources.get(\"progress_bus\")\n    if bus is None:\n        return {\"events\": []}\n\n    async def gen():\n        sent = 0\n        while True:\n            events = bus.get_events(id)\n            for e in events[sent:]:\n                yield f\"data: {e.model_dump_json(exclude_none=True)}\\n\\n\"\n                sent += 1\n            await asyncio.sleep(0.5)\n\n    return StreamingResponse(gen(), media_type=\"text/event-stream\")\n

Known limitation. The default in-memory bus is per-process. If the worker and the API run as separate processes, events published by the worker are not visible to the API. For production, back the bus with a shared broker. The port boundary makes this a one-adapter change.

"},{"location":"08_human_in_loop/","title":"Human-in-the-Loop","text":""},{"location":"08_human_in_loop/#summary","title":"Summary","text":"

Some inputs are ambiguous and must be resolved by a human before processing can continue. The blueprint models this as a durable pause: an exception is raised, a question is recorded, the job pauses, and later resumes.

"},{"location":"08_human_in_loop/#the-mechanism","title":"The mechanism","text":"
  1. During processing, a stage raises a domain exception carrying the context needed to ask the question:
raise PauseForResolution(\n    job_id=jid,\n    reference=\"<stable key for this decision>\",\n    options=[...],\n    context={...},\n)\n
  1. The orchestrator catches it, records a question document, and re-raises so the worker marks the job paused.
  2. A client lists pending questions and answers one:
GET  /jobs/{id}/questions            -> list pending questions\nPOST /questions/{qid}/resolve  {choice}  -> record the choice\n
  1. When the last pending question is resolved, the resolve endpoint flips the parent job back to pending.
  2. The worker picks it up again, rebuilds a resolution map from previously answered questions, and seeds it into the pipeline so the same decision is not asked twice.
"},{"location":"08_human_in_loop/#the-question-document","title":"The question document","text":"
class Question(BaseDocument):\n    job_id: str\n    reference: str            # stable key: stage|field|value\n    status: str               # pending | resolved\n    options: list[dict]\n    context: dict\n    chosen: dict | None = None\n    resolved_at: str | None = None\n
"},{"location":"08_human_in_loop/#resume-map","title":"Resume map","text":"

The worker turns answered questions into a lookup keyed by reference, then injects it into the pipeline so that stage can skip re-raising:

resolved = {q.reference: q.chosen for q in await svc.search(\n    filter={\"job_id\": jid, \"status\": \"resolved\"})}\n\npipeline = build(root, instances={\"pause\": handler}, config={\"resolved\": resolved})\n
"},{"location":"08_human_in_loop/#resolve-endpoint","title":"Resolve endpoint","text":"
async def resolve_question(request, id, body):\n    svc = request.app.state.db_resources[\"question_svc\"]\n    q = await svc.read(id)\n    await svc.update(id, {\"chosen\": body[\"chosen\"], \"status\": \"resolved\",\n                          \"resolved_at\": now()})\n\n    pending = await svc.search(filter={\"job_id\": q.job_id, \"status\": \"pending\"},\n                               limit=1)\n    if not pending:\n        jobs = request.app.state.db_resources[\"job_svc\"]\n        job = await jobs.read(q.job_id)\n        if job and job.status == \"paused\":\n            await jobs.update_status(q.job_id, \"pending\")\n    return q\n
"},{"location":"08_human_in_loop/#cli-parity","title":"CLI parity","text":"

The same flow is available through scripts/CLI: list, get, create, questions, resolve, retry.

"},{"location":"09_infra_adaptors/","title":"Infra Adapters","text":""},{"location":"09_infra_adaptors/#summary","title":"Summary","text":"

All external systems are reached through adapters in infra/. Prefer an existing internal library over writing a new adapter; where no library exists, copy the Tier B scaffold from templates/app. See the Genericity Matrix for the Tier A / B / C split.

"},{"location":"09_infra_adaptors/#database-mongodb","title":"Database (MongoDB)","text":" 
repo_factory(collection, document, engine=None,\n             populate_rules=None, cache=None)\n
PopulateRule(field=\"owner\", collection=\"users\",\n             nested_rules=[], max_depth=2, projection=None)\n

Tier A \u2014 Library (Mongo Ops). Do not write raw Motor calls in the domain; consult the registry and the library's MCP before touching the driver.

"},{"location":"09_infra_adaptors/#http-response-cache","title":"HTTP response cache","text":"

Tier B \u2014 template. Copy from templates/app; not a standalone library.

A generic GET cache sits behind a decorator.

@cache_response(ttl=60)\nasync def list_items(request): ...\n
"},{"location":"09_infra_adaptors/#event-progress-bus","title":"Event / progress bus","text":"

Tier B \u2014 template. Copy from templates/app.

A port with an in-memory default adapter (publish, get_events). Swap for a durable broker in production (see Worker & Progress).

"},{"location":"09_infra_adaptors/#external-clients-ingestion","title":"External clients (ingestion)","text":"

External fetchers (files, mail, HTTP) are adapters that wrap ingestion libraries. Construct them in the composition root and inject into orchestrators.

Tier A \u2014 Libraries: Omniread (formats) and Mail Intake (email).

"},{"location":"09_infra_adaptors/#llm-clients","title":"LLM clients","text":"

Tier B \u2014 template. Copy the pool scaffold from templates/app.

When a pipeline calls a model provider, route through a pool adaptor that handles provider selection, retries, and client lifecycle:

pool = setup_async_pool(provider=settings.llm_provider)\nresult = await pool.complete(model=\"...\", messages=[...])\n

Providers are selected by LLM_PROVIDER and configured by environment.

"},{"location":"09_infra_adaptors/#key-value-store","title":"Key-value store","text":"

Tier B \u2014 template. Copy from templates/app.

For report/snapshot storage, provide a pluggable store port with a Redis adapter and an in-memory fallback.

"},{"location":"09_infra_adaptors/#rule","title":"Rule","text":""},{"location":"12_genericity_matrix/","title":"Genericity Matrix","text":""},{"location":"12_genericity_matrix/#summary","title":"Summary","text":"

This page states precisely what is library-backed, what is template-provided, and what is app-specific and must be authored. Use it to scope a new build and to know, for any given capability, whether to call a library, copy template code, or write it yourself.

"},{"location":"12_genericity_matrix/#tier-a-library-backed-use-never-reimplement","title":"Tier A \u2014 Library-backed (use; never reimplement)","text":"

These capabilities already exist as internal libraries. Reach them through the library registry MCP (find_library(capability)) and read the library's own MCP for its API. Writing a private copy of any of these is a defect.

Capability Library MCP Persistence: repositories, cache, populate, transactions Mongo Ops (mongo_ops) :8007 Dependency injection / composition Hexa (hexa) :8008 Deterministic multi-step transformation engine DAG Pipe (dagpipe) :8006 Spec-first HTTP edge, client, codegen OpenAPI First (openapi_first) :8004 Format-agnostic fetch + parse (HTML, PDF, CSV, XLSX) Omniread (omniread) :8003 Read-only email ingestion Mail Intake (mail_intake) :8002 JWT auth primitives (register, login, introspect) Py JWT (jwtlib) :8005 Documentation build (MkDocs + MCP) Doc Forge (docforge) :8001"},{"location":"12_genericity_matrix/#tier-b-template-provided-copy-from-templatesapp-then-adapt","title":"Tier B \u2014 Template-provided (copy from templates/app, then adapt)","text":"

These are reusable internal components shipped inside the template \u2014 they are not independent libraries. Copy them as-is, then rename/populate only what is app-specific. They give every app the same wiring shape as the reference app.

Component Template location Action Layering & dependency rule the package layout + import rules keep Composition root shape init.py (get_svc/get_db, idempotency) keep Spec-first web edge interfaces/openapi.py, thin handlers bound to operationId keep CRUD slice pattern domain/<resource>/{models.py, ports/db.py, ports/service.py, converters.py} rename + fill Generic CRUD service infra/adapters/service.py (CrudsServiceAdaptor) keep Repository factory infra/db/repos/mongo.py (_repo_factory) keep Worker poll + status loop worker.py keep Progress bus + streaming bus port + in-memory adapter + SSE endpoint keep Human-in-the-loop durable pause + question document + resolve endpoint + resume map keep HTTP response cache cache_response decorator + generation invalidation keep Key-value store KV store port + Redis adapter + in-memory fallback keep LLM client pool provider-select + retry pool scaffold keep Pipeline scaffold pipelines/shared/{ports,loader,registry}.py + pipelines/<family>/ rename + fill Operator scripts scripts/ (init() / get_svc() / shutdown()) rename Testing harness integration fakes + fixture-parity pattern keep Project scaffolding pyproject.toml, Dockerfile, docker-compose.yaml, .env.example, openapi.yaml rename + fill"},{"location":"12_genericity_matrix/#tier-c-app-specific-must-be-authored","title":"Tier C \u2014 App-specific (must be authored)","text":"Artifact Why Resource models the business shape Repository populate/cache rules the relationships Service business logic domain behaviour Pipeline families the transformation graph Pattern/rule resources data-driven policy Fixtures & golden outputs test data"},{"location":"12_genericity_matrix/#the-two-axes","title":"The two axes","text":"

Everything app-specific reduces to two axes:

  1. Resource model \u2014 the set of resources, their fields, relationships, and the OpenAPI contract.
  2. Pipeline family \u2014 only if inputs require multi-step transformation.

If you can specify those two axes, the rest of the app follows mechanically: Tier A is imported, Tier B is copied, and Tier C is the only thing you author.

"},{"location":"13_library_registry/","title":"Library Registry","text":""},{"location":"13_library_registry/#summary","title":"Summary","text":"

The Aetos ecosystem provides internal libraries for the shared concerns of every application. Use them. Writing a private copy of any capability below is a defect.

Each library also exposes its own documentation as an MCP server. An AI agent can read a library's full API by connecting to its MCP and reading doc://index, doc://nav, and doc://modules/{module}.

"},{"location":"13_library_registry/#catalog","title":"Catalog","text":"Library Import Purpose Use instead of MCP DAG Pipe dagpipe Deterministic state-transformation DAG engine hand-rolled orchestration/topological ordering :8006 Hexa hexa Declarative DI / composition-root tooling manual wiring, hand-written factories :8008 Mongo Ops mongo_ops Async MongoDB repositories, cache, populate, transactions raw Motor CRUD and connection lifecycle :8007 Omniread omniread Format-agnostic fetch + parse (HTML, PDF, CSV, XLSX) custom scrapers/parsers :8003 OpenAPI First openapi_first Spec-first FastAPI bootstrap, client, codegen decorator routers and hand-written clients :8004 Mail Intake mail_intake Provider-agnostic read-only email ingestion Gmail OAuth + MIME parsing :8002 Py JWT jwtlib JWT auth primitives (register/login/introspect) password hashing + token code :8005 Doc Forge docforge Docstring compiler \u2192 MkDocs + MCP docs custom doc generators :8001 Auth Server (service) Central identity service (register/login/me/introspect) building auth endpoints yourself (deferred)"},{"location":"13_library_registry/#how-to-choose","title":"How to choose","text":"You need to\u2026 Use persist and query documents Mongo Ops compose dependencies Hexa run a multi-step transformation DAG Pipe read/parse an arbitrary input format Omniread expose a contract-first HTTP API OpenAPI First ingest email Mail Intake authenticate users / verify tokens Py JWT (+ Auth Server) generate documentation Doc Forge"},{"location":"13_library_registry/#discovery","title":"Discovery","text":"

A single standards MCP exposes find_library(capability), which returns the matching library, its import name, its \"use instead of\" guidance, and its MCP endpoint. Query it before writing any infrastructure.

"},{"location":"13_library_registry/#notes","title":"Notes","text":""}]} \ No newline at end of file diff --git a/standards/mcp/components.json b/standards/mcp/components.json new file mode 100644 index 0000000..b144741 --- /dev/null +++ b/standards/mcp/components.json @@ -0,0 +1,214 @@ +{ + "project": "standards", + "template": "templates/app", + "seed": { + "resource": "item", + "family": "example" + }, + "components": [ + { + "kind": "model", + "path": "domain/item/models.py", + "placeholder": "resource", + "description": "Business Pydantic model for a resource." + }, + { + "kind": "ports-db", + "path": "domain/item/ports/db.py", + "placeholder": "resource", + "description": "Document model + repository factory for a resource." + }, + { + "kind": "ports-service", + "path": "domain/item/ports/service.py", + "placeholder": "resource", + "description": "CRUD service wired through the DI library." + }, + { + "kind": "converters", + "path": "domain/item/ports/converters.py", + "placeholder": "resource", + "description": "Optional pure converters between domain types." + }, + { + "kind": "routes", + "path": "interfaces/item/routes.py", + "placeholder": "resource", + "description": "Thin HTTP handlers bound to operationIds." + }, + { + "kind": "test-model", + "path": "tests/unit/test_item_models.py", + "placeholder": "resource", + "description": "Unit test for the resource model." + }, + { + "kind": "test-routes", + "path": "tests/integration/test_item_routes.py", + "placeholder": "resource", + "description": "Integration test at the HTTP edge." + }, + { + "kind": "pipeline-abc", + "path": "pipelines/example/abc.py", + "placeholder": "family", + "description": "Port contracts for a pipeline family." + }, + { + "kind": "pipeline-impl", + "path": "pipelines/example/impl.py", + "placeholder": "family", + "description": "Concrete stages, wired root, and family registration." + }, + { + "kind": "pipeline-yaml", + "path": "pipelines/example/pipeline.yaml", + "placeholder": "family", + "description": "Graph spec (nodes, edges, roots, schema)." + }, + { + "kind": "pipeline-resources", + "path": "pipelines/example/resources/pipeline.yaml", + "placeholder": "family", + "description": "Runtime spec consumed by the shared loader." + }, + { + "kind": "pipeline-pattern", + "path": "pipelines/example/resources/patterns/example.yaml", + "placeholder": "family", + "description": "Pattern/rule fragment merged by the loader." + }, + { + "kind": "pipeline-ports", + "path": "pipelines/shared/ports.py", + "placeholder": "", + "description": "Shared pipeline ports (FinalizePort, RegistryPort)." + }, + { + "kind": "pipeline-loader", + "path": "pipelines/shared/loader.py", + "placeholder": "", + "description": "Shared loader: DI wiring + pattern merge." + }, + { + "kind": "pipeline-registry", + "path": "pipelines/shared/registry.py", + "placeholder": "", + "description": "Shared registry: (kind, variant, mode) -> root class." + }, + { + "kind": "pipeline-simple", + "path": "pipelines/item_sync.py", + "placeholder": "", + "description": "Smallest pipeline form (plain async steps)." + }, + { + "kind": "job-model", + "path": "domain/job/models.py", + "placeholder": "", + "description": "Seeded second resource model (job)." + }, + { + "kind": "job-ports-db", + "path": "domain/job/ports/db.py", + "placeholder": "", + "description": "Seeded second resource repository (job)." + }, + { + "kind": "job-ports-service", + "path": "domain/job/ports/service.py", + "placeholder": "", + "description": "Seeded second resource service (job)." + }, + { + "kind": "job-routes", + "path": "interfaces/job/routes.py", + "placeholder": "", + "description": "Seeded second resource routes (job)." + }, + { + "kind": "config", + "path": "config.py", + "placeholder": "", + "description": "Settings dataclass + singleton." + }, + { + "kind": "init", + "path": "init.py", + "placeholder": "", + "description": "Composition root (init/shutdown, get_svc)." + }, + { + "kind": "main", + "path": "main.py", + "placeholder": "", + "description": "Spec-first app bootstrap + lifespan." + }, + { + "kind": "worker", + "path": "worker.py", + "placeholder": "", + "description": "Poll loop + job status machine." + }, + { + "kind": "openapi", + "path": "openapi.yaml", + "placeholder": "", + "description": "OpenAPI contract skeleton." + }, + { + "kind": "pyproject", + "path": "pyproject.toml", + "placeholder": "", + "description": "Packaging + dependency pins." + }, + { + "kind": "dockerfile", + "path": "Dockerfile", + "placeholder": "", + "description": "API + worker image with private-index wiring." + }, + { + "kind": "compose", + "path": "docker-compose.yaml", + "placeholder": "", + "description": "api + worker + mongo compose file." + }, + { + "kind": "env-example", + "path": ".env.example", + "placeholder": "", + "description": "Environment variable template." + }, + { + "kind": "db-setup", + "path": "infra/db/setup.py", + "placeholder": "", + "description": "Mongo connection lifecycle." + }, + { + "kind": "repo", + "path": "infra/db/repos/mongo.py", + "placeholder": "", + "description": "Repository factory (base/populating/cached)." + }, + { + "kind": "service-adapter", + "path": "infra/adapters/service.py", + "placeholder": "", + "description": "Generic CRUD service adaptor." + }, + { + "kind": "script-seed", + "path": "scripts/seed.py", + "placeholder": "", + "description": "Operator seed script." + }, + { + "kind": "script-drop", + "path": "scripts/drop.py", + "placeholder": "", + "description": "Operator drop script." + } + ] +} \ No newline at end of file diff --git a/standards/mcp/modules/genericity_matrix.json b/standards/mcp/modules/genericity_matrix.json index 18253f3..688e482 100644 --- a/standards/mcp/modules/genericity_matrix.json +++ b/standards/mcp/modules/genericity_matrix.json @@ -2,7 +2,7 @@ "module": "genericity_matrix", "content": { "path": "genericity_matrix", - "docstring": "# Genericity Matrix\n\n## Summary\n\nThis page states precisely what is **reusable as-is**, what is a **template to\nadapt**, and what is **app-specific and must be authored**. Use it to scope a\nnew build.\n\n---\n\n## Reusable as-is (do not rewrite)\n\n| Area | Artifact |\n|---|---|\n| Layering & dependency rule | the package layout and import rules |\n| Composition root | the `init.py` shape, idempotency, `get_svc` overloading |\n| Web edge | `OpenAPIFirstApp` bootstrap, handler-name = `operationId`, thin handlers |\n| CRUD slice | the four-file pattern (model / ports-db / ports-service / converters) |\n| Generic CRUD service | the base service adaptor |\n| Repository factory | base / populating / cached selection |\n| Worker | the poll loop, backoff, retry, status machine |\n| Progress bus | the bus port + in-memory adapter + streaming endpoint |\n| Human-in-the-loop | the durable pause + question document + resolve endpoint + resume map |\n| Infra adaptors | HTTP cache decorator, KV store port, LLM pool scaffold |\n| Testing | integration harness with fakes, fixture-parity pattern |\n| Docs | GSDFC + generated stubs + MCP bundle pipeline |\n\n## Template to adapt (copy, then change names)\n\n| Artifact | Change |\n|---|---|\n| `pyproject.toml` | project name, dependencies |\n| `Dockerfile` / `docker-compose.yaml` | service name, ports, services |\n| `.env.example` | connection strings, provider keys |\n| `openapi.yaml` | every path/schema/operationId |\n| `scripts/` | seed and operator tasks |\n\n## App-specific (must be authored)\n\n| Artifact | Why |\n|---|---|\n| Resource models | the business shape |\n| Repository populate/cache rules | the relationships |\n| Service business logic | domain behaviour |\n| Pipeline families (if any) | the transformation graph |\n| Pattern/rule resources | data-driven policy |\n| Fixtures & golden outputs | test data |\n\n## The two axes\n\nEverything app-specific reduces to **two axes**:\n\n1. **Resource model** — the set of resources, their fields, relationships, and\n the OpenAPI contract.\n2. **Pipeline family** — only if inputs require multi-step transformation.\n\nIf you can specify those two axes, the rest of the app follows mechanically.\n", + "docstring": "# Genericity Matrix\n\n## Summary\n\nThis page states precisely what is **library-backed**, what is\n**template-provided**, and what is **app-specific and must be authored**. Use it\nto scope a new build and to know, for any given capability, whether to call a\nlibrary, copy template code, or write it yourself.\n\n---\n\n## Tier A — Library-backed (use; never reimplement)\n\nThese capabilities already exist as internal libraries. Reach them through the\nlibrary registry MCP (`find_library(capability)`) and read the library's own MCP\nfor its API. Writing a private copy of any of these is a defect.\n\n| Capability | Library | MCP |\n|---|---|---|\n| Persistence: repositories, cache, populate, transactions | Mongo Ops (`mongo_ops`) | `:8007` |\n| Dependency injection / composition | Hexa (`hexa`) | `:8008` |\n| Deterministic multi-step transformation engine | DAG Pipe (`dagpipe`) | `:8006` |\n| Spec-first HTTP edge, client, codegen | OpenAPI First (`openapi_first`) | `:8004` |\n| Format-agnostic fetch + parse (HTML, PDF, CSV, XLSX) | Omniread (`omniread`) | `:8003` |\n| Read-only email ingestion | Mail Intake (`mail_intake`) | `:8002` |\n| JWT auth primitives (register, login, introspect) | Py JWT (`jwtlib`) | `:8005` |\n| Documentation build (MkDocs + MCP) | Doc Forge (`docforge`) | `:8001` |\n\n## Tier B — Template-provided (copy from `templates/app`, then adapt)\n\nThese are reusable internal components shipped **inside** the template — they are\nnot independent libraries. Copy them as-is, then rename/populate only what is\napp-specific. They give every app the same wiring shape as the reference app.\n\n| Component | Template location | Action |\n|---|---|---|\n| Layering & dependency rule | the package layout + import rules | keep |\n| Composition root shape | `init.py` (`get_svc`/`get_db`, idempotency) | keep |\n| Spec-first web edge | `interfaces/openapi.py`, thin handlers bound to `operationId` | keep |\n| CRUD slice pattern | `domain//{models.py, ports/db.py, ports/service.py, converters.py}` | rename + fill |\n| Generic CRUD service | `infra/adapters/service.py` (`CrudsServiceAdaptor`) | keep |\n| Repository factory | `infra/db/repos/mongo.py` (`_repo_factory`) | keep |\n| Worker poll + status loop | `worker.py` | keep |\n| Progress bus + streaming | bus port + in-memory adapter + SSE endpoint | keep |\n| Human-in-the-loop | durable pause + question document + resolve endpoint + resume map | keep |\n| HTTP response cache | `cache_response` decorator + generation invalidation | keep |\n| Key-value store | KV store port + Redis adapter + in-memory fallback | keep |\n| LLM client pool | provider-select + retry pool scaffold | keep |\n| Pipeline scaffold | `pipelines/shared/{ports,loader,registry}.py` + `pipelines//` | rename + fill |\n| Operator scripts | `scripts/` (`init()` / `get_svc()` / `shutdown()`) | rename |\n| Testing harness | integration fakes + fixture-parity pattern | keep |\n| Project scaffolding | `pyproject.toml`, `Dockerfile`, `docker-compose.yaml`, `.env.example`, `openapi.yaml` | rename + fill |\n\n## Tier C — App-specific (must be authored)\n\n| Artifact | Why |\n|---|---|\n| Resource models | the business shape |\n| Repository populate/cache rules | the relationships |\n| Service business logic | domain behaviour |\n| Pipeline families | the transformation graph |\n| Pattern/rule resources | data-driven policy |\n| Fixtures & golden outputs | test data |\n\n---\n\n## The two axes\n\nEverything app-specific reduces to **two axes**:\n\n1. **Resource model** — the set of resources, their fields, relationships, and\n the OpenAPI contract.\n2. **Pipeline family** — only if inputs require multi-step transformation.\n\nIf you can specify those two axes, the rest of the app follows mechanically:\nTier A is imported, Tier B is copied, and Tier C is the only thing you author.\n", "objects": {} } } \ No newline at end of file diff --git a/standards/mcp/modules/index.json b/standards/mcp/modules/index.json index 6ea50b1..625805d 100644 --- a/standards/mcp/modules/index.json +++ b/standards/mcp/modules/index.json @@ -2,7 +2,7 @@ "module": "index", "content": { "path": "index", - "docstring": "# Standards Reference\n\nThe reference half of the Standards description: the enforceable rules,\npatterns, and the internal library catalog that agents consult while coding.\n\n- **Layering Rules** — the strict request → domain → repository layering and\n the rules that enforce it.\n- **Composition Root** — a single composition point wiring the whole app.\n- **Spec-First Routing** — the OpenAPI contract as the routing source of truth.\n- **CRUD Template** — the standardized generic resource endpoint scaffold.\n- **Pipeline Scaffold** — the deterministic pipeline skeleton for multi-stage\n jobs.\n- **Worker & Progress** — queued work, progress reporting, and visibility.\n- **Human-in-the-Loop** — guardrails and approvals inside automations.\n- **Infra Adapters** — how infrastructure connects without leaking into domain.\n- **Genericity Matrix** — what is generic vs app-specific.\n- **Library Registry** — the catalog of internal libraries behind all of it.\n\nRead the **Wiki** for the philosophy and rationale; these pages are the\nreference that mirrors the `standards` MCP bundle.\n", + "docstring": "# Standards Reference\n\nThe reference half of the Standards description: the enforceable rules,\npatterns, and the internal library catalog that agents consult while coding.\n\n- **Layering Rules** — the strict request → domain → repository layering and\n the rules that enforce it.\n- **Composition Root** — a single composition point wiring the whole app.\n- **Spec-First Routing** — the OpenAPI contract as the routing source of truth.\n- **CRUD Template** — the standardized generic resource endpoint scaffold.\n- **Pipeline Scaffold** — the deterministic pipeline skeleton for multi-stage\n jobs.\n- **Worker & Progress** — queued work, progress reporting, and visibility.\n- **Human-in-the-Loop** — guardrails and approvals inside automations.\n- **Infra Adapters** — how infrastructure connects without leaking into domain.\n- **Genericity Matrix** — what is library-backed, template-provided, or authored.\n- **Library Registry** — the catalog of internal libraries behind all of it.\n\nRead the **Wiki** for the philosophy and rationale; these pages are the\nreference that mirrors the `standards` MCP bundle.\n", "objects": {} } } \ No newline at end of file diff --git a/standards/mcp/modules/infra_adaptors.json b/standards/mcp/modules/infra_adaptors.json index bb00edb..fb3833a 100644 --- a/standards/mcp/modules/infra_adaptors.json +++ b/standards/mcp/modules/infra_adaptors.json @@ -2,7 +2,7 @@ "module": "infra_adaptors", "content": { "path": "infra_adaptors", - "docstring": "# Infra Adapters\n\n## Summary\n\nAll external systems are reached through adapters in `infra/`. Prefer an\nexisting internal library over writing a new adapter.\n\n---\n\n## Database (MongoDB)\n\n- A single connection manager owns the client lifecycle.\n `setup_database(uri, db_name)` / `shutdown_database()`.\n- A repository **factory** selects the repository flavour:\n\n ```python\n repo_factory(collection, document, engine=None,\n populate_rules=None, cache=None)\n ```\n\n - base repository — plain CRUD;\n - populating repository — resolves referenced documents (foreign keys);\n - cached repository — an LRU/TTL cache in front of reads.\n\n- Referenced documents are described declaratively:\n\n ```python\n PopulateRule(field=\"owner\", collection=\"users\",\n nested_rules=[], max_depth=2, projection=None)\n ```\n\n- Transactions use a session context manager provided by the library.\n\n**Library:** the Mongo operations library (see registry). Do not write raw\nMotor calls in the domain.\n\n## HTTP response cache\n\nA generic GET cache sits behind a decorator.\n\n```python\n@cache_response(ttl=60)\nasync def list_items(request): ...\n```\n\n- Keyed on path + query + user.\n- No-op when no store is wired.\n- `invalidate(*resources)` bumps a generation counter for write-triggered\n invalidation.\n\n## Event / progress bus\n\nA port with an in-memory default adapter (`publish`, `get_events`). Swap for a\ndurable broker in production (see Worker & Progress).\n\n## External clients (ingestion)\n\nExternal fetchers (files, mail, HTTP) are adapters that wrap ingestion\nlibraries. Construct them in the composition root and inject into\norchestrators.\n\n**Libraries:** the ingestion library (formats) and the mail library (email).\n\n## LLM clients\n\nWhen a pipeline calls a model provider, route through a pool adaptor that\nhandles provider selection, retries, and client lifecycle:\n\n```python\npool = setup_async_pool(provider=settings.llm_provider)\nresult = await pool.complete(model=\"...\", messages=[...])\n```\n\nProviders are selected by `LLM_PROVIDER` and configured by environment.\n\n## Key-value store\n\nFor report/snapshot storage, provide a pluggable store port with a Redis\nadapter and an in-memory fallback.\n\n## Rule\n\nIf a capability already exists as a library (DB, DI, ingestion, mail, auth,\ndocs), **use it**. Writing a private copy of a library is a defect.\n", + "docstring": "# Infra Adapters\n\n## Summary\n\nAll external systems are reached through adapters in `infra/`. Prefer an\nexisting internal library over writing a new adapter; where no library exists,\ncopy the Tier B scaffold from `templates/app`. See the Genericity Matrix for the\nTier A / B / C split.\n\n---\n\n## Database (MongoDB)\n\n- A single connection manager owns the client lifecycle.\n `setup_database(uri, db_name)` / `shutdown_database()`.\n- A repository **factory** selects the repository flavour:\n\n ```python\n repo_factory(collection, document, engine=None,\n populate_rules=None, cache=None)\n ```\n\n - base repository — plain CRUD;\n - populating repository — resolves referenced documents (foreign keys);\n - cached repository — an LRU/TTL cache in front of reads.\n\n- Referenced documents are described declaratively:\n\n ```python\n PopulateRule(field=\"owner\", collection=\"users\",\n nested_rules=[], max_depth=2, projection=None)\n ```\n\n- Transactions use a session context manager provided by the library.\n\n**Tier A — Library (Mongo Ops).** Do not write raw Motor calls in the domain;\nconsult the registry and the library's MCP before touching the driver.\n\n## HTTP response cache\n\n**Tier B — template.** Copy from `templates/app`; not a standalone library.\n\nA generic GET cache sits behind a decorator.\n\n```python\n@cache_response(ttl=60)\nasync def list_items(request): ...\n```\n\n- Keyed on path + query + user.\n- No-op when no store is wired.\n- `invalidate(*resources)` bumps a generation counter for write-triggered\n invalidation.\n\n## Event / progress bus\n\n**Tier B — template.** Copy from `templates/app`.\n\nA port with an in-memory default adapter (`publish`, `get_events`). Swap for a\ndurable broker in production (see Worker & Progress).\n\n## External clients (ingestion)\n\nExternal fetchers (files, mail, HTTP) are adapters that wrap ingestion\nlibraries. Construct them in the composition root and inject into\norchestrators.\n\n**Tier A — Libraries:** Omniread (formats) and Mail Intake (email).\n\n## LLM clients\n\n**Tier B — template.** Copy the pool scaffold from `templates/app`.\n\nWhen a pipeline calls a model provider, route through a pool adaptor that\nhandles provider selection, retries, and client lifecycle:\n\n```python\npool = setup_async_pool(provider=settings.llm_provider)\nresult = await pool.complete(model=\"...\", messages=[...])\n```\n\nProviders are selected by `LLM_PROVIDER` and configured by environment.\n\n## Key-value store\n\n**Tier B — template.** Copy from `templates/app`.\n\nFor report/snapshot storage, provide a pluggable store port with a Redis\nadapter and an in-memory fallback.\n\n## Rule\n\n- **Tier A (libraries: DB, DI, ingestion, mail, auth, docs, pipelines)** — use\n the library; writing a private copy is a defect. Query `find_library` first.\n- **Tier B (cache, progress bus, LLM pool, KV store, composition shape, worker,\n tests)** — these are template code, not libraries: copy them from\n `templates/app` rather than searching the registry.\n- **Tier C (models, business logic, pipeline families, rules)** — author them\n yourself.\n", "objects": {} } } \ No newline at end of file diff --git a/standards/mcp/modules/library_registry.json b/standards/mcp/modules/library_registry.json index bee7171..91684d2 100644 --- a/standards/mcp/modules/library_registry.json +++ b/standards/mcp/modules/library_registry.json @@ -2,7 +2,7 @@ "module": "library_registry", "content": { "path": "library_registry", - "docstring": "# Library Registry\n\n## Summary\n\nThe Aetos ecosystem provides internal libraries for the shared concerns of\nevery application. **Use them.** Writing a private copy of any capability below\nis a defect.\n\nEach library also exposes its own documentation as an MCP server. An AI agent\ncan read a library's full API by connecting to its MCP and reading\n`doc://index`, `doc://nav`, and `doc://modules/{module}`.\n\n---\n\n## Catalog\n\n| Library | Import | Purpose | Use instead of | MCP |\n|---|---|---|---|---|\n| DAG Pipe | `dagpipe` | Deterministic state-transformation DAG engine | hand-rolled orchestration/topological ordering | `:8006` |\n| Hexa | `hexa` | Declarative DI / composition-root tooling | manual wiring, hand-written factories | *(see note)* |\n| Mongo Ops | `mongo_ops` | Async MongoDB repositories, cache, populate, transactions | raw Motor CRUD and connection lifecycle | `:8007` |\n| Omniread | `omniread` | Format-agnostic fetch + parse (HTML, PDF, CSV, XLSX) | custom scrapers/parsers | `:8003` |\n| OpenAPI First | `openapi_first` | Spec-first FastAPI bootstrap, client, codegen | decorator routers and hand-written clients | `:8004` |\n| Mail Intake | `mail_intake` | Provider-agnostic read-only email ingestion | Gmail OAuth + MIME parsing | `:8002` |\n| Py JWT | `jwtlib` | JWT auth primitives (register/login/introspect) | password hashing + token code | `:8005` |\n| Doc Forge | `docforge` | Docstring compiler → MkDocs + MCP docs | custom doc generators | `:8001` |\n| Auth Server | *(service)* | Central identity service (register/login/me/introspect) | building auth endpoints yourself | — |\n\n## How to choose\n\n| You need to… | Use |\n|---|---|\n| persist and query documents | Mongo Ops |\n| compose dependencies | Hexa |\n| run a multi-step transformation | DAG Pipe |\n| read/parse an arbitrary input format | Omniread |\n| expose a contract-first HTTP API | OpenAPI First |\n| ingest email | Mail Intake |\n| authenticate users / verify tokens | Py JWT (+ Auth Server) |\n| generate documentation | Doc Forge |\n\n## Discovery\n\nA single **standards MCP** exposes `find_library(capability)`, which returns the\nmatching library, its import name, its \"use instead of\" guidance, and its MCP\nendpoint. Query it before writing any infrastructure.\n\n## Notes\n\n- **Hexa** ships an MCP bundle but is not yet assigned a dedicated port in the\n hub config; it is reachable via the standards registry entry and will be\n wired to a port.\n- Library MCP servers expose **documentation resources** (not executable tools):\n they tell an agent the available API surface so it does not invent code.\n", + "docstring": "# Library Registry\n\n## Summary\n\nThe Aetos ecosystem provides internal libraries for the shared concerns of\nevery application. **Use them.** Writing a private copy of any capability below\nis a defect.\n\nEach library also exposes its own documentation as an MCP server. An AI agent\ncan read a library's full API by connecting to its MCP and reading\n`doc://index`, `doc://nav`, and `doc://modules/{module}`.\n\n---\n\n## Catalog\n\n| Library | Import | Purpose | Use instead of | MCP |\n|---|---|---|---|---|\n| DAG Pipe | `dagpipe` | Deterministic state-transformation DAG engine | hand-rolled orchestration/topological ordering | `:8006` |\n| Hexa | `hexa` | Declarative DI / composition-root tooling | manual wiring, hand-written factories | `:8008` |\n| Mongo Ops | `mongo_ops` | Async MongoDB repositories, cache, populate, transactions | raw Motor CRUD and connection lifecycle | `:8007` |\n| Omniread | `omniread` | Format-agnostic fetch + parse (HTML, PDF, CSV, XLSX) | custom scrapers/parsers | `:8003` |\n| OpenAPI First | `openapi_first` | Spec-first FastAPI bootstrap, client, codegen | decorator routers and hand-written clients | `:8004` |\n| Mail Intake | `mail_intake` | Provider-agnostic read-only email ingestion | Gmail OAuth + MIME parsing | `:8002` |\n| Py JWT | `jwtlib` | JWT auth primitives (register/login/introspect) | password hashing + token code | `:8005` |\n| Doc Forge | `docforge` | Docstring compiler → MkDocs + MCP docs | custom doc generators | `:8001` |\n| Auth Server | *(service)* | Central identity service (register/login/me/introspect) | building auth endpoints yourself | *(deferred)* |\n\n## How to choose\n\n| You need to… | Use |\n|---|---|\n| persist and query documents | Mongo Ops |\n| compose dependencies | Hexa |\n| run a multi-step transformation | DAG Pipe |\n| read/parse an arbitrary input format | Omniread |\n| expose a contract-first HTTP API | OpenAPI First |\n| ingest email | Mail Intake |\n| authenticate users / verify tokens | Py JWT (+ Auth Server) |\n| generate documentation | Doc Forge |\n\n## Discovery\n\nA single **standards MCP** exposes `find_library(capability)`, which returns the\nmatching library, its import name, its \"use instead of\" guidance, and its MCP\nendpoint. Query it before writing any infrastructure.\n\n## Notes\n\n- **Hexa** is a full library with its own MCP server on `:8008`: read\n `doc://index`, `doc://nav`, and `doc://modules/{module}` (e.g.\n `doc://modules/hexa.container`) to discover the DI surface — `build(cls, *,\n instances, config)` and the `PortNode` model.\n- **Auth Server** is deferred: it has no `pyproject.toml`/importable package and\n no MCP bundle. Do not reach for it yet; `py-jwt` covers token primitives.\n- Library MCP servers expose **documentation resources** (not executable tools):\n they tell an agent the available API surface so it does not invent code.\n", "objects": {} } } \ No newline at end of file diff --git a/standards/mcp/templates/app/.dockerignore b/standards/mcp/templates/app/.dockerignore new file mode 100644 index 0000000..7143c6c --- /dev/null +++ b/standards/mcp/templates/app/.dockerignore @@ -0,0 +1,8 @@ +.venv/ +__pycache__/ +*.pyc +.env +.git/ +docs/ +dist/ +build/ \ No newline at end of file diff --git a/standards/mcp/templates/app/.env.example b/standards/mcp/templates/app/.env.example new file mode 100644 index 0000000..3c7d11b --- /dev/null +++ b/standards/mcp/templates/app/.env.example @@ -0,0 +1,5 @@ +DEBUG=false +MONGO_URI=mongodb://localhost:27017 +MONGO_DB=app_template +WORKER_POLL_SECONDS=30 +WORKER_MAX_RETRIES=3 \ No newline at end of file diff --git a/standards/mcp/templates/app/Dockerfile b/standards/mcp/templates/app/Dockerfile new file mode 100644 index 0000000..b39d077 --- /dev/null +++ b/standards/mcp/templates/app/Dockerfile @@ -0,0 +1,20 @@ +FROM python:3.11-slim + +WORKDIR /app + +# Private index for the internal Aetos libraries — pass credentials as build args. +ARG PIP_REGISTRY_USERNAME +ARG PIP_REGISTRY_PASSWORD +ENV PIP_EXTRA_INDEX_URL=https://${PIP_REGISTRY_USERNAME}:${PIP_REGISTRY_PASSWORD}@pip.aetoskia.com/simple + +COPY pyproject.toml ./ +COPY . . +RUN pip install --no-cache-dir --upgrade pip \ + && pip install --no-cache-dir --extra-index-url "${PIP_EXTRA_INDEX_URL}" . + +EXPOSE 8000 + +# Two processes, per the two axes: +# - main.py : the spec-first FastAPI application +# - worker.py : the pipeline consumer polling pending jobs +CMD ["sh", "-c", "uvicorn main:app --host 0.0.0.0 --port 8000 & exec python worker.py"] \ No newline at end of file diff --git a/standards/mcp/templates/app/config.py b/standards/mcp/templates/app/config.py new file mode 100644 index 0000000..a163e38 --- /dev/null +++ b/standards/mcp/templates/app/config.py @@ -0,0 +1,36 @@ +"""# Summary + +App configuration loaded from environment — a Settings dataclass with defaults +plus a singleton settings instance populated from OS environment variables at +import time. +""" +import os +from dataclasses import dataclass + + +@dataclass +class Settings: + """Runtime configuration settings for the application. + + Attributes: + debug (bool): Enable debug behavior. Defaults to False. + mongo_uri (str): MongoDB connection URI. Defaults to mongodb://localhost:27017. + mongo_db (str): MongoDB database name. Defaults to app_template. + worker_poll_seconds (int): Worker poll interval in seconds. Defaults to 30. + worker_max_retries (int): Maximum job retries before skipping. Defaults to 3. + """ + + debug: bool = False + mongo_uri: str = "mongodb://localhost:27017" + mongo_db: str = "app_template" + worker_poll_seconds: int = 30 + worker_max_retries: int = 3 + + +settings = Settings( + debug=os.environ.get("DEBUG", "").lower() in ("1", "true", "yes"), + mongo_uri=os.environ.get("MONGO_URI", "mongodb://localhost:27017"), + mongo_db=os.environ.get("MONGO_DB", "app_template"), + worker_poll_seconds=int(os.environ.get("WORKER_POLL_SECONDS", "30")), + worker_max_retries=int(os.environ.get("WORKER_MAX_RETRIES", "3")), +) \ No newline at end of file diff --git a/standards/mcp/templates/app/docker-compose.yaml b/standards/mcp/templates/app/docker-compose.yaml new file mode 100644 index 0000000..d011d83 --- /dev/null +++ b/standards/mcp/templates/app/docker-compose.yaml @@ -0,0 +1,21 @@ +services: + api: + build: . + command: uvicorn main:app --host 0.0.0.0 --port 8000 + env_file: .env + ports: + - "8000:8000" + depends_on: + - mongo + + worker: + build: . + command: python worker.py + env_file: .env + depends_on: + - mongo + + mongo: + image: mongo:7 + ports: + - "27017:27017" \ No newline at end of file diff --git a/standards/mcp/templates/app/domain/__init__.py b/standards/mcp/templates/app/domain/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/domain/item/__init__.py b/standards/mcp/templates/app/domain/item/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/domain/item/models.py b/standards/mcp/templates/app/domain/item/models.py new file mode 100644 index 0000000..63c3875 --- /dev/null +++ b/standards/mcp/templates/app/domain/item/models.py @@ -0,0 +1,19 @@ +"""# Summary + +Item domain model — the neutral resource used throughout the template. +""" +from pydantic import BaseModel, Field + + +class Item(BaseModel): + """A single countable resource. + + Attributes: + name (str): Item name. + quantity (int): Item quantity. Defaults to 1. + """ + + name: str + quantity: int = Field(default=1) + + model_config = {"extra": "ignore"} \ No newline at end of file diff --git a/standards/mcp/templates/app/domain/item/ports/__init__.py b/standards/mcp/templates/app/domain/item/ports/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/domain/item/ports/converters.py b/standards/mcp/templates/app/domain/item/ports/converters.py new file mode 100644 index 0000000..f386df4 --- /dev/null +++ b/standards/mcp/templates/app/domain/item/ports/converters.py @@ -0,0 +1,18 @@ +"""# Summary + +Optional converters for the item slice — pure functions mapping between domain +types. Add a converter here when a service must reshape data; keep it pure. +""" +from domain.item.models import Item + + +def to_item(data: dict) -> Item: + """Build an :class:`Item` from a raw mapping. + + Args: + data (dict): Field mapping matching the item model. + + Returns: + Item: The constructed item model. + """ + return Item(**data) diff --git a/standards/mcp/templates/app/domain/item/ports/db.py b/standards/mcp/templates/app/domain/item/ports/db.py new file mode 100644 index 0000000..d1dbbb5 --- /dev/null +++ b/standards/mcp/templates/app/domain/item/ports/db.py @@ -0,0 +1,20 @@ +"""# Summary + +Mongo persistence for items: document schema and repository factory. +""" +from domain.item.models import Item +from infra.db.repos.mongo import _repo_factory as mongo_repo_factory +from mongo_ops import BaseDocument, BaseRepository + + +class ItemDocument(Item, BaseDocument): + """Mongo document form of :class:`Item`.""" + + +def create_item_repo() -> BaseRepository: + """Create the Mongo-backed item repository. + + Returns: + BaseRepository: Repository wired to the ``items`` collection. + """ + return mongo_repo_factory("items", ItemDocument) \ No newline at end of file diff --git a/standards/mcp/templates/app/domain/item/ports/service.py b/standards/mcp/templates/app/domain/item/ports/service.py new file mode 100644 index 0000000..59daba7 --- /dev/null +++ b/standards/mcp/templates/app/domain/item/ports/service.py @@ -0,0 +1,33 @@ +"""# Summary + +CRUD service wiring for the item aggregate via the hexa container. +""" +from hexa import build + +from domain.item.models import Item +from domain.item.ports.db import ItemDocument +from infra.adapters.service import CrudsServiceAdaptor + + +class ItemCrudsService(CrudsServiceAdaptor[Item, ItemDocument]): + """Item CRUD service resolving :class:`ItemDocument` from a repo. + + Notes: + Instances are created by :func:`get_item_service`. The ``model`` and + ``schema`` class attributes bind the domain model and its document type. + """ + + model = Item + schema = ItemDocument + + +def get_item_service(repo) -> ItemCrudsService: + """Build an item CRUD service wired to ``repo``. + + Args: + repo (Any): Mongo-backed item repository from :func:`create_item_repo`. + + Returns: + ItemCrudsService: Ready-to-use item CRUD service. + """ + return build(ItemCrudsService, instances={"repo": repo}) \ No newline at end of file diff --git a/standards/mcp/templates/app/domain/job/__init__.py b/standards/mcp/templates/app/domain/job/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/domain/job/models.py b/standards/mcp/templates/app/domain/job/models.py new file mode 100644 index 0000000..17ec51e --- /dev/null +++ b/standards/mcp/templates/app/domain/job/models.py @@ -0,0 +1,26 @@ +"""# Summary + +Job domain model — a pipeline work item consumed by the background worker. +""" +from typing import Literal +from pydantic import BaseModel, Field + + +class Job(BaseModel): + """A unit of background work. + + Attributes: + name (str): Job name. + status (str): Lifecycle status. Defaults to "pending". + retry_count (int): Times the job failed. Defaults to 0. + error_message (str | None): Last failure message. Defaults to None. + payload (dict | None): Arbitrary job payload. Defaults to None. + """ + + name: str + status: Literal["pending", "processing", "done", "failed"] = "pending" + retry_count: int = Field(default=0) + error_message: str | None = None + payload: dict | None = None + + model_config = {"extra": "ignore"} \ No newline at end of file diff --git a/standards/mcp/templates/app/domain/job/ports/__init__.py b/standards/mcp/templates/app/domain/job/ports/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/domain/job/ports/db.py b/standards/mcp/templates/app/domain/job/ports/db.py new file mode 100644 index 0000000..547e2b5 --- /dev/null +++ b/standards/mcp/templates/app/domain/job/ports/db.py @@ -0,0 +1,20 @@ +"""# Summary + +Mongo persistence for jobs: document schema and repository factory. +""" +from domain.job.models import Job +from infra.db.repos.mongo import _repo_factory as mongo_repo_factory +from mongo_ops import BaseDocument, BaseRepository + + +class JobDocument(Job, BaseDocument): + """Mongo document form of :class:`Job`.""" + + +def create_job_repo() -> BaseRepository: + """Create the Mongo-backed job repository. + + Returns: + BaseRepository: Repository wired to the ``jobs`` collection. + """ + return mongo_repo_factory("jobs", JobDocument) \ No newline at end of file diff --git a/standards/mcp/templates/app/domain/job/ports/service.py b/standards/mcp/templates/app/domain/job/ports/service.py new file mode 100644 index 0000000..9f6b714 --- /dev/null +++ b/standards/mcp/templates/app/domain/job/ports/service.py @@ -0,0 +1,33 @@ +"""# Summary + +CRUD service wiring for the job aggregate via the hexa container. +""" +from hexa import build + +from domain.job.models import Job +from domain.job.ports.db import JobDocument +from infra.adapters.service import CrudsServiceAdaptor + + +class JobCrudsService(CrudsServiceAdaptor[Job, JobDocument]): + """Job CRUD service resolving :class:`JobDocument` from a repo. + + Notes: + Instances are created by :func:`get_job_service`. The ``model`` and + ``schema`` class attributes bind the domain model and its document type. + """ + + model = Job + schema = JobDocument + + +def get_job_service(repo) -> JobCrudsService: + """Build a job CRUD service wired to ``repo``. + + Args: + repo (Any): Mongo-backed job repository from :func:`create_job_repo`. + + Returns: + JobCrudsService: Ready-to-use job CRUD service. + """ + return build(JobCrudsService, instances={"repo": repo}) \ No newline at end of file diff --git a/standards/mcp/templates/app/infra/__init__.py b/standards/mcp/templates/app/infra/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/infra/adapters/__init__.py b/standards/mcp/templates/app/infra/adapters/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/infra/adapters/service.py b/standards/mcp/templates/app/infra/adapters/service.py new file mode 100644 index 0000000..dfa46d8 --- /dev/null +++ b/standards/mcp/templates/app/infra/adapters/service.py @@ -0,0 +1,128 @@ +"""# Summary + +Generic CRUD service adaptor over a Mongo repository. + +Bridge pydantic models and mongo-ops documents, exposing create, read, update, +patch, delete, and search operations for every domain that needs them. +""" +from typing import Generic, Type, TypeVar + +from infra.db.repos import mongo + +from pydantic import BaseModel +from mongo_ops import BaseDocument +from mongo_ops.models import PyObjectId + +Model = TypeVar("Model", bound=BaseModel) +Schema = TypeVar("Schema", bound=BaseDocument) + + +class CrudsServiceAdaptor(Generic[Model, Schema]): + """Adapt a generic Mongo repository to a model/schema CRUD service. + + Attributes: + repo (mongo.BaseRepository): Repository backing all CRUD operations. + model (Type[Model]): Pydantic model type handled by this service. + schema (Type[Schema]): Document schema type persisted by the repository. + """ + + repo: mongo.BaseRepository + model: Type[Model] + schema: Type[Schema] + + def __init__(self, *, repo=None): + """Initialize the adaptor with a backing repository. + + Args: + repo (mongo.BaseRepository): Repository used for all operations. + """ + self.repo = repo + + @staticmethod + def model_to_dict(model_obj: Model) -> dict: + """Serialize a model into a sparse dictionary. + + Args: + model_obj (Model): Model instance to serialize. + + Returns: + dict: Fields that are set and not ``None``. + """ + return model_obj.model_dump(exclude_unset=True, exclude_none=True) + + async def create(self, data: Model | dict) -> Schema: + """Create a document from a model or raw mapping. + + Args: + data (Model | dict): Model instance or field mapping to persist. + + Returns: + Schema: Persisted document. + """ + if isinstance(data, dict): + data = self.schema(**data) + return await self.repo.create(data) + + async def read(self, id: PyObjectId) -> Schema | None: + """Read a document by primary key. + + Args: + id (PyObjectId): Document identifier. + + Returns: + Schema | None: Matching document, or ``None`` when absent. + """ + return await self.repo.get_by_id(id) + + async def update(self, id: PyObjectId, data: Model | dict) -> Schema | None: + """Replace a document by primary key. + + Args: + id (PyObjectId): Document identifier. + data (Model | dict): Replacement model or field mapping. + + Returns: + Schema | None: Updated document, or ``None`` when absent. + """ + return await self.repo.update( + id, + data if isinstance(data, dict) else self.model_to_dict(data), + ) + + async def patch(self, id: PyObjectId, data: dict) -> Schema | None: + """Partially update a document by primary key. + + Args: + id (PyObjectId): Document identifier. + data (dict): Fields to patch. + + Returns: + Schema | None: Updated document, or ``None`` when absent. + """ + return await self.repo.patch(id, data) + + async def delete(self, id: PyObjectId) -> bool: + """Delete a document by primary key. + + Args: + id (PyObjectId): Document identifier. + + Returns: + bool: ``True`` when a document was deleted. + """ + return await self.repo.delete(id) + + async def search(self, filter: dict | None = None, *args, **kwargs) -> list[Schema]: + """List documents matching an optional filter. + + Args: + filter (dict | None): Query filter passed to the repository. + + Returns: + list[Schema]: Matching documents. + + Notes: + Extra positional and keyword arguments are forwarded to the + repository. + """ + return await self.repo.get_many(filter=filter, *args, **kwargs) \ No newline at end of file diff --git a/standards/mcp/templates/app/infra/db/__init__.py b/standards/mcp/templates/app/infra/db/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/infra/db/repos/__init__.py b/standards/mcp/templates/app/infra/db/repos/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/infra/db/repos/mongo.py b/standards/mcp/templates/app/infra/db/repos/mongo.py new file mode 100644 index 0000000..a68f8f3 --- /dev/null +++ b/standards/mcp/templates/app/infra/db/repos/mongo.py @@ -0,0 +1,42 @@ +"""# Summary + +Generic MongoDB CRUD repo — single implementation for all domains. + +Select a concrete mongo-ops repository (base, cached, or populating) from the +options supplied to the factory. +""" +from typing import List, Optional, Type + +from mongo_ops import BaseRepository, CachedBaseRepository, PopulatingRepository +from mongo_ops.cache import CacheBackend, CacheConfig +from mongo_ops.models import BaseDocument +from mongo_ops.populate import PopulationEngine, PopulateRule + + +def _repo_factory( + collection: str, + db_model: Type[BaseDocument], + cache_backend: Optional[CacheBackend] = None, + cache_config: Optional[CacheConfig] = None, + engine: Optional[PopulationEngine] = None, + populate_rules: Optional[List[PopulateRule]] = None, +) -> BaseRepository: + """Build the appropriate mongo-ops repository. + + Args: + collection (str): Collection name to operate on. + db_model (Type[BaseDocument]): Document model bound to the collection. + cache_backend (Optional[CacheBackend]): Cache backend enabling cached reads. + cache_config (Optional[CacheConfig]): Cache configuration for the backend. + engine (Optional[PopulationEngine]): Population engine for reference expansion. + populate_rules (Optional[List[PopulateRule]]): Rules applied by the engine. + + Returns: + BaseRepository: Repository instance matching the supplied options. + """ + if engine and populate_rules is not None: + return PopulatingRepository(collection, db_model, engine, populate_rules) + elif cache_backend is not None: + return CachedBaseRepository(collection, db_model, cache_backend, cache_config) + else: + return BaseRepository(collection, db_model) \ No newline at end of file diff --git a/standards/mcp/templates/app/infra/db/setup.py b/standards/mcp/templates/app/infra/db/setup.py new file mode 100644 index 0000000..0cca126 --- /dev/null +++ b/standards/mcp/templates/app/infra/db/setup.py @@ -0,0 +1,23 @@ +"""# Summary + +Database setup — connection lifecycle only. No domain imports. + +Connect and disconnect the shared Mongo connection manager at application +startup and shutdown. +""" +from mongo_ops import MongoConnectionManager + + +async def setup_database(uri: str, db_name: str): + """Connect the shared Mongo connection manager. + + Args: + uri (str): Mongo connection URI. + db_name (str): Target database name. + """ + await MongoConnectionManager.connect(uri=uri, db_name=db_name) + + +async def shutdown_database(): + """Disconnect the shared Mongo connection manager.""" + await MongoConnectionManager.disconnect() \ No newline at end of file diff --git a/standards/mcp/templates/app/init.py b/standards/mcp/templates/app/init.py new file mode 100644 index 0000000..68a6fac --- /dev/null +++ b/standards/mcp/templates/app/init.py @@ -0,0 +1,51 @@ +"""# Summary + +Infrastructure initialization — connects MongoDB and wires all repos and services. +Shared bootstrap for both the FastAPI app and the worker, guarded against +repeated initialization via a module-level flag. +""" +from typing import Any, Dict + +from config import settings +from domain.item.ports.db import create_item_repo +from domain.item.ports.service import get_item_service +from domain.job.ports.db import create_job_repo +from domain.job.ports.service import get_job_service +from infra.db.setup import setup_database, shutdown_database + +_resources: Dict[str, Any] = {} +_initialized: bool = False + + +async def init() -> Dict[str, Any]: + """Connect MongoDB, wire all repos and services, and return the resources dict. + + Idempotent: repeated calls short-circuit and return the cached resources. + + Returns: + Dict[str, Any]: Named services keyed by resource name. + """ + global _resources, _initialized + if _initialized: + return _resources + + await setup_database(uri=settings.mongo_uri, db_name=settings.mongo_db) + + item_repo = create_item_repo() + job_repo = create_job_repo() + + _resources["item_repo"] = item_repo + _resources["job_repo"] = job_repo + _resources["item_svc"] = get_item_service(item_repo) + _resources["job_svc"] = get_job_service(job_repo) + + _initialized = True + return _resources + + +async def shutdown() -> None: + """Disconnect MongoDB and reset the resources cache.""" + global _initialized + await shutdown_database() + _resources.clear() + _initialized = False \ No newline at end of file diff --git a/standards/mcp/templates/app/interfaces/__init__.py b/standards/mcp/templates/app/interfaces/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/interfaces/item/__init__.py b/standards/mcp/templates/app/interfaces/item/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/interfaces/item/routes.py b/standards/mcp/templates/app/interfaces/item/routes.py new file mode 100644 index 0000000..995f782 --- /dev/null +++ b/standards/mcp/templates/app/interfaces/item/routes.py @@ -0,0 +1,42 @@ +"""# Summary + +Item route handlers — names must match openapi.yaml operationIds. +""" +from fastapi import Request +from fastapi.responses import JSONResponse + +from domain.item.models import Item + + +def svc(request: Request): + """Resolve the item CRUD service from app state.""" + return request.app.state.db_resources["item_svc"] + + +async def create_item(request: Request, body: dict): + """Create an item from the request body.""" + data = body.model_dump() if hasattr(body, "model_dump") else dict(body) + doc = await svc(request).create(Item(**data)) + return JSONResponse(status_code=201, content=doc.model_dump(mode="json", by_alias=True)) + + +async def get_item(request: Request, id: str): + """Fetch one item by id.""" + doc = await svc(request).read(id) + if doc is None: + return JSONResponse(status_code=404, content={"detail": "item not found"}) + return doc.model_dump(mode="json", by_alias=True) + + +async def list_items(request: Request): + """List all items.""" + docs = await svc(request).search({}) + return [d.model_dump(mode="json", by_alias=True) for d in docs] + + +async def delete_item(request: Request, id: str): + """Delete one item by id.""" + ok = await svc(request).delete(id) + if not ok: + return JSONResponse(status_code=404, content={"detail": "item not found"}) + return {"deleted": True} \ No newline at end of file diff --git a/standards/mcp/templates/app/interfaces/job/__init__.py b/standards/mcp/templates/app/interfaces/job/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/interfaces/job/routes.py b/standards/mcp/templates/app/interfaces/job/routes.py new file mode 100644 index 0000000..3d4f80f --- /dev/null +++ b/standards/mcp/templates/app/interfaces/job/routes.py @@ -0,0 +1,34 @@ +"""# Summary + +Job route handlers — names must match openapi.yaml operationIds. +""" +from fastapi import Request +from fastapi.responses import JSONResponse + +from domain.job.models import Job + + +def svc(request: Request): + """Resolve the job CRUD service from app state.""" + return request.app.state.db_resources["job_svc"] + + +async def create_job(request: Request, body: dict): + """Create a job from the request body.""" + data = body.model_dump() if hasattr(body, "model_dump") else dict(body) + doc = await svc(request).create(Job(**data)) + return JSONResponse(status_code=201, content=doc.model_dump(mode="json", by_alias=True)) + + +async def get_job(request: Request, id: str): + """Fetch one job by id.""" + doc = await svc(request).read(id) + if doc is None: + return JSONResponse(status_code=404, content={"detail": "job not found"}) + return doc.model_dump(mode="json", by_alias=True) + + +async def list_jobs(request: Request): + """List all jobs.""" + docs = await svc(request).search({}) + return [d.model_dump(mode="json", by_alias=True) for d in docs] \ No newline at end of file diff --git a/standards/mcp/templates/app/interfaces/openapi.py b/standards/mcp/templates/app/interfaces/openapi.py new file mode 100644 index 0000000..8c43aa7 --- /dev/null +++ b/standards/mcp/templates/app/interfaces/openapi.py @@ -0,0 +1,8 @@ +"""# Summary + +All handler functions — names must match openapi.yaml operationId values. The +OpenAPI-first router resolves each operation to a handler by name, so this module +re-exports every route handler from the subpackages into one flat namespace. +""" +from interfaces.item.routes import * +from interfaces.job.routes import * \ No newline at end of file diff --git a/standards/mcp/templates/app/main.py b/standards/mcp/templates/app/main.py new file mode 100644 index 0000000..0a451e3 --- /dev/null +++ b/standards/mcp/templates/app/main.py @@ -0,0 +1,54 @@ +"""# Summary + +FastAPI application bootstrap — uses OpenAPIFirstApp for spec-driven routing. +Handlers are looked up by operationId in interfaces.openapi, so no router tables +are declared here; the lifespan initializes and shuts down app resources. +""" +import logging +from contextlib import asynccontextmanager + +from dotenv import load_dotenv +from openapi_first.app import OpenAPIFirstApp +from starlette.middleware.cors import CORSMiddleware + +from init import init, shutdown + +load_dotenv() + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", +) + +import interfaces.openapi as openapi_routes + +logger = logging.getLogger(__name__) + + +@asynccontextmanager +async def lifespan(app: OpenAPIFirstApp): + """Initialize resource wiring on startup and shut it down on exit. + + Args: + app (OpenAPIFirstApp): The ASGI application exposing app.state. + + Yields: + None: The application runs while the context is open. + """ + resources = await init() + app.state.db_resources = resources + + yield + + await shutdown() + + +app = OpenAPIFirstApp( + openapi_path="openapi.yaml", + routes_module=openapi_routes, + title="Item API", + version="0.1.0", + lifespan=lifespan, +) + +app.add_middleware(CORSMiddleware, allow_origins=["*"], allow_methods=["*"], allow_headers=["*"]) \ No newline at end of file diff --git a/standards/mcp/templates/app/openapi.yaml b/standards/mcp/templates/app/openapi.yaml new file mode 100644 index 0000000..652471a --- /dev/null +++ b/standards/mcp/templates/app/openapi.yaml @@ -0,0 +1,161 @@ +openapi: 3.0.3 +info: + title: Item API + version: 0.1.0 +paths: + /items: + get: + operationId: list_items + summary: List items + responses: + "200": + description: OK + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/Item" + post: + operationId: create_item + summary: Create an item + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/ItemInput" + responses: + "201": + description: Created + content: + application/json: + schema: + $ref: "#/components/schemas/Item" + /items/{id}: + get: + operationId: get_item + summary: Get an item + parameters: + - name: id + in: path + required: true + schema: + type: string + responses: + "200": + description: OK + content: + application/json: + schema: + $ref: "#/components/schemas/Item" + "404": + description: Not found + delete: + operationId: delete_item + summary: Delete an item + parameters: + - name: id + in: path + required: true + schema: + type: string + responses: + "200": + description: Deleted + "404": + description: Not found + /jobs: + get: + operationId: list_jobs + summary: List jobs + responses: + "200": + description: OK + content: + application/json: + schema: + type: array + items: + $ref: "#/components/schemas/Job" + post: + operationId: create_job + summary: Create a job + requestBody: + required: true + content: + application/json: + schema: + $ref: "#/components/schemas/JobInput" + responses: + "201": + description: Created + content: + application/json: + schema: + $ref: "#/components/schemas/Job" + /jobs/{id}: + get: + operationId: get_job + summary: Get a job + parameters: + - name: id + in: path + required: true + schema: + type: string + responses: + "200": + description: OK + content: + application/json: + schema: + $ref: "#/components/schemas/Job" + "404": + description: Not found +components: + schemas: + Item: + type: object + required: [name, quantity] + properties: + id: + type: string + name: + type: string + quantity: + type: integer + ItemInput: + type: object + required: [name] + properties: + name: + type: string + quantity: + type: integer + default: 1 + Job: + type: object + required: [name, status] + properties: + id: + type: string + name: + type: string + status: + type: string + enum: [pending, processing, done, failed] + retry_count: + type: integer + error_message: + type: string + payload: + type: object + JobInput: + type: object + required: [name] + properties: + name: + type: string + payload: + type: object \ No newline at end of file diff --git a/standards/mcp/templates/app/pipelines/__init__.py b/standards/mcp/templates/app/pipelines/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/pipelines/example/__init__.py b/standards/mcp/templates/app/pipelines/example/__init__.py new file mode 100644 index 0000000..196d174 --- /dev/null +++ b/standards/mcp/templates/app/pipelines/example/__init__.py @@ -0,0 +1,7 @@ +"""# Summary + +Example pipeline family — a reusable skeleton, not an app-specific pipeline. + +Replace ``Example*`` names and the three stages with your family's stages. The +shape (abc → impl → pipeline.yaml → resources/) stays the same. +""" diff --git a/standards/mcp/templates/app/pipelines/example/abc.py b/standards/mcp/templates/app/pipelines/example/abc.py new file mode 100644 index 0000000..ebbc570 --- /dev/null +++ b/standards/mcp/templates/app/pipelines/example/abc.py @@ -0,0 +1,53 @@ +"""# Summary + +Example pipeline family — port contracts (stage interfaces + root interface). + +Ports are declared as a DI tree: the root's annotated slots *are* the dependency +decisions. The concrete subclass in ``impl.py`` pins the stage types. +""" +from abc import ABC, abstractmethod +from typing import Any + +from pipelines.shared.ports import FinalizePort + + +class IngestPort(ABC): + """Obtain the raw input as text/bytes.""" + + @abstractmethod + def materialize(self, source: Any) -> str: + """Return the raw input materialized as text.""" + + +class ReadPort(ABC): + """Parse materialized input into records.""" + + @abstractmethod + def read(self, content: str) -> list[dict[str, Any]]: + """Return parsed records.""" + + +class NormalizePort(ABC): + """Normalize raw records into the target shape.""" + + @abstractmethod + def build(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: + """Return normalized records.""" + + +class ExamplePipelinePort(ABC): + """Root of the example family. + + Notes: + The slot annotations below are contracts. A concrete subclass re-annotates + them with stage implementations; the DI container instantiates the tree. + """ + + ingest: IngestPort + reader: ReadPort + normalize: NormalizePort + finalize: FinalizePort + + @abstractmethod + async def run(self, source: Any) -> list[dict[str, Any]]: + """Run the full stage chain and return the normalized records.""" diff --git a/standards/mcp/templates/app/pipelines/example/impl.py b/standards/mcp/templates/app/pipelines/example/impl.py new file mode 100644 index 0000000..d78c30e --- /dev/null +++ b/standards/mcp/templates/app/pipelines/example/impl.py @@ -0,0 +1,71 @@ +"""# Summary + +Example pipeline family — concrete stages, the wired root, and its registration. + +The concrete subclass re-annotates each slot with a stage implementation; the DI +container builds the whole tree. Register the family so the shared registry can +resolve it by ``(kind, variant, mode)``. +""" +from pathlib import Path +from typing import Any + +from pipelines.example.abc import ( + ExamplePipelinePort, + IngestPort, + NormalizePort, + ReadPort, +) +from pipelines.shared.loader import build_pipeline +from pipelines.shared.ports import FinalizePort +from pipelines.shared.registry import register + + +class TextIngest(IngestPort): + """Materialize any source as text.""" + + def materialize(self, source: Any) -> str: + return source if isinstance(source, str) else str(source) + + +class LineReader(ReadPort): + """Read one record per non-empty line.""" + + def read(self, content: str) -> list[dict[str, Any]]: + return [{"line": line} for line in content.splitlines() if line.strip()] + + +class PassthroughNormalize(NormalizePort): + """Return records unchanged (replace with real normalization).""" + + def build(self, records: list[dict[str, Any]]) -> list[dict[str, Any]]: + return records + + +class NoopFinalize(FinalizePort): + """Drop records (replace with a service-backed finalize stage).""" + + async def finalize(self, records: list[dict[str, Any]]) -> list[Any]: + return records + + +class ExamplePipeline(ExamplePipelinePort): + """Concrete example family — the dependency decisions live here.""" + + ingest: TextIngest + reader: LineReader + normalize: PassthroughNormalize + finalize: NoopFinalize + + async def run(self, source: Any) -> list[dict[str, Any]]: + content = self.ingest.materialize(source) + records = self.reader.read(content) + return self.normalize.build(records) + + +def get_example_pipeline() -> ExamplePipeline: + """Wire and return the example pipeline through the shared loader.""" + resources = Path(__file__).parent / "resources" + return build_pipeline(ExamplePipeline, resources_dir=resources) + + +register("example", "default", ExamplePipeline) diff --git a/standards/mcp/templates/app/pipelines/example/pipeline.yaml b/standards/mcp/templates/app/pipelines/example/pipeline.yaml new file mode 100644 index 0000000..0ebfe2d --- /dev/null +++ b/standards/mcp/templates/app/pipelines/example/pipeline.yaml @@ -0,0 +1,18 @@ +# Graph spec for the example family: nodes, edges, and the root. +# Mirrors the port tree in abc.py. Replace with your family's graph. +schema: ExampleRecord +nodes: + ingest: + port: IngestPort + reader: + port: ReadPort + normalize: + port: NormalizePort + finalize: + port: FinalizePort +edges: + - ingest -> reader + - reader -> normalize + - normalize -> finalize +roots: + default: ExamplePipeline diff --git a/standards/mcp/templates/app/pipelines/example/resources/patterns/example.yaml b/standards/mcp/templates/app/pipelines/example/resources/patterns/example.yaml new file mode 100644 index 0000000..0074eff --- /dev/null +++ b/standards/mcp/templates/app/pipelines/example/resources/patterns/example.yaml @@ -0,0 +1,4 @@ +# Pattern / rule fragments for the example family. +# The loader merges every patterns/*.yaml into one registry by key. +example: + enabled: true diff --git a/standards/mcp/templates/app/pipelines/example/resources/pipeline.yaml b/standards/mcp/templates/app/pipelines/example/resources/pipeline.yaml new file mode 100644 index 0000000..8a670be --- /dev/null +++ b/standards/mcp/templates/app/pipelines/example/resources/pipeline.yaml @@ -0,0 +1,5 @@ +# Runtime spec consumed by the shared loader. +kind: example +variant: default +mode: default +patterns: [] diff --git a/standards/mcp/templates/app/pipelines/item_sync.py b/standards/mcp/templates/app/pipelines/item_sync.py new file mode 100644 index 0000000..ff71c0c --- /dev/null +++ b/standards/mcp/templates/app/pipelines/item_sync.py @@ -0,0 +1,32 @@ +"""# Summary + +Item-sync pipeline — the demo pipeline axis of the blueprint. + +Runs on the worker, consumes a pending job, and materializes its payload as an +item. Steps are plain async functions here; once the chain branches or needs +deterministic replay, wrap the steps in a dagpipe DAG (see standards +"06_pipeline_scaffold"). +""" +import logging + +logger = logging.getLogger(__name__) + + +async def run_item_sync(job, resources: dict) -> None: + """Materialize a pending job payload as an item. + + Step 1: extract the payload from the job. + Step 2: create the item through the item service. + + Args: + job: The pending job document. + resources (dict): Wired services and adaptors. + """ + payload = job.payload or {} + item_svc = resources["item_svc"] + + await item_svc.create({ + "name": payload.get("name", job.name), + "quantity": payload.get("quantity", 1), + }) + logger.info("Created item from job %s", str(job.id)) \ No newline at end of file diff --git a/standards/mcp/templates/app/pipelines/shared/__init__.py b/standards/mcp/templates/app/pipelines/shared/__init__.py new file mode 100644 index 0000000..8aef39c --- /dev/null +++ b/standards/mcp/templates/app/pipelines/shared/__init__.py @@ -0,0 +1,5 @@ +"""# Summary + +Shared pipeline scaffolding — ports, loader, and registry reused by every +pipeline family. See standards "06_pipeline_scaffold". +""" diff --git a/standards/mcp/templates/app/pipelines/shared/loader.py b/standards/mcp/templates/app/pipelines/shared/loader.py new file mode 100644 index 0000000..10a3c2b --- /dev/null +++ b/standards/mcp/templates/app/pipelines/shared/loader.py @@ -0,0 +1,54 @@ +"""# Summary + +Pipeline loader — wire a family root through the DI library and attach the +merged pattern registry from ``resources/patterns/*.yaml``. + +All families share this loader so wiring is never hand-rolled. +""" +from pathlib import Path +from typing import Any + +import yaml + +from hexa import build + + +def load_patterns(resources_dir: str | Path) -> dict[str, Any]: + """Merge every pattern fragment under ``resources/patterns`` into one map. + + Args: + resources_dir (str | Path): Family resources directory. + + Returns: + dict[str, Any]: Merged pattern/rule fragments keyed by fragment name. + """ + registry: dict[str, Any] = {} + patterns = Path(resources_dir) / "patterns" + if patterns.is_dir(): + for path in sorted(patterns.glob("*.yaml")): + data = yaml.safe_load(path.read_text(encoding="utf-8")) or {} + registry.update(data) + return registry + + +def build_pipeline( + root_cls, + *, + resources_dir: str | Path = "resources", + instances: dict[str, Any] | None = None, + config: dict[str, Any] | None = None, +): + """Wire ``root_cls`` and inject the merged pattern registry. + + Args: + root_cls: The pipeline root class (a port tree). + resources_dir (str | Path): Directory holding ``patterns/*.yaml``. + instances (dict[str, Any] | None): Runtime objects injected by slot name. + config (dict[str, Any] | None): Config values propagated by field name. + + Returns: + The fully wired pipeline root. + """ + merged_instances = dict(instances or {}) + merged_instances.setdefault("registry", load_patterns(resources_dir)) + return build(root_cls, instances=merged_instances, config=config or {}) diff --git a/standards/mcp/templates/app/pipelines/shared/ports.py b/standards/mcp/templates/app/pipelines/shared/ports.py new file mode 100644 index 0000000..ec15d8e --- /dev/null +++ b/standards/mcp/templates/app/pipelines/shared/ports.py @@ -0,0 +1,25 @@ +"""# Summary + +Shared pipeline ports — the contracts every pipeline family reuses. + +A port is an abstract stage contract. Concrete families pin their implementations +on a root's annotated slots; the DI library instantiates the tree. +""" +from abc import ABC, abstractmethod +from typing import Any + + +class FinalizePort(ABC): + """Terminal stage — turn normalized records into persisted resources.""" + + @abstractmethod + async def finalize(self, records: list[dict[str, Any]]) -> list[Any]: + """Persist records and return the created resources.""" + + +class RegistryPort(ABC): + """Resolve a pattern / rule fragment by key.""" + + @abstractmethod + def resolve(self, key: str) -> dict[str, Any]: + """Return the fragment registered under ``key``.""" diff --git a/standards/mcp/templates/app/pipelines/shared/registry.py b/standards/mcp/templates/app/pipelines/shared/registry.py new file mode 100644 index 0000000..d5de929 --- /dev/null +++ b/standards/mcp/templates/app/pipelines/shared/registry.py @@ -0,0 +1,29 @@ +"""# Summary + +Family registry — resolve a ``(kind, variant, mode)`` triple to a root class. + +A single lookup replaces hand-rolled ``if/else`` family selection. +""" +from typing import Dict, Type + +# Populated by ``register`` as families import. +_FAMILIES: Dict[str, Dict[str, Dict[str, Type]]] = {} + + +def register(kind: str, variant: str, root_cls: Type, mode: str = "default") -> Type: + """Register ``root_cls`` for a ``(kind, variant, mode)`` triple. + + Returns: + Type: The registered class (so it can be used as a decorator). + """ + _FAMILIES.setdefault(kind, {}).setdefault(variant, {})[mode] = root_cls + return root_cls + + +def resolve_root(kind: str, variant: str, mode: str = "default") -> Type: + """Return the root class for a ``(kind, variant, mode)`` triple. + + Raises: + KeyError: When no family matches the triple. + """ + return _FAMILIES[kind][variant][mode] diff --git a/standards/mcp/templates/app/pyproject.toml b/standards/mcp/templates/app/pyproject.toml new file mode 100644 index 0000000..48288f1 --- /dev/null +++ b/standards/mcp/templates/app/pyproject.toml @@ -0,0 +1,44 @@ +[build-system] +requires = ["setuptools>=68"] +build-backend = "setuptools.build_meta" + +[project] +name = "app-template" +version = "0.1.0" +requires-python = ">=3.11" +dependencies = [ + "fastapi", + "uvicorn[standard]", + "pydantic>=2", + "motor", + "python-dotenv", + "pyyaml", + "openapi-first", + "mongo-ops", + "hexa", + "dagpipe", +] + +[project.optional-dependencies] +dev = [ + "pytest", + "pytest-asyncio", + "httpx", + "ruff", + "doc-forge", +] + +[tool.setuptools.packages.find] +where = ["."] +include = ["domain*", "infra*", "interfaces*", "pipelines*", "scripts*"] + +[tool.pytest.ini_options] +asyncio_mode = "auto" +testpaths = ["tests"] + +[tool.ruff] +line-length = 100 +target-version = "py311" + +[tool.ruff.lint] +select = ["E", "F", "I", "UP"] \ No newline at end of file diff --git a/standards/mcp/templates/app/scripts/__init__.py b/standards/mcp/templates/app/scripts/__init__.py new file mode 100644 index 0000000..43e0740 --- /dev/null +++ b/standards/mcp/templates/app/scripts/__init__.py @@ -0,0 +1,7 @@ +"""# Summary + +Operator scripts — run from the project root as ``python -m scripts.``. + +They use the composition root (:func:`init.init` / :func:`init.shutdown`) so they +share the exact wiring the API and worker use. +""" diff --git a/standards/mcp/templates/app/scripts/drop.py b/standards/mcp/templates/app/scripts/drop.py new file mode 100644 index 0000000..042064a --- /dev/null +++ b/standards/mcp/templates/app/scripts/drop.py @@ -0,0 +1,32 @@ +"""# Summary + +Operator script — drop the collections owned by each service for a clean slate. + +Run from the project root: + + python -m scripts.drop +""" +import asyncio + +from init import init, shutdown + + +async def main() -> None: + """Delete every service collection, best-effort, then shut down.""" + resources = await init() + try: + dropped = 0 + for name, svc in resources.items(): + if not name.endswith("_svc"): + continue + delete_many = getattr(svc.repo, "delete_many", None) + if delete_many is not None: + await delete_many({}) + dropped += 1 + print(f"Dropped {dropped} service collection(s).") + finally: + await shutdown() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/standards/mcp/templates/app/scripts/seed.py b/standards/mcp/templates/app/scripts/seed.py new file mode 100644 index 0000000..2a2ad0f --- /dev/null +++ b/standards/mcp/templates/app/scripts/seed.py @@ -0,0 +1,26 @@ +"""# Summary + +Operator script — seed demo data through the composition root. + +Run from the project root: + + python -m scripts.seed +""" +import asyncio + +from init import init, shutdown + + +async def main() -> None: + """Seed one demo item and shut down cleanly.""" + resources = await init() + try: + item_svc = resources["item_svc"] + await item_svc.create({"name": "seed", "quantity": 1}) + print("Seeded one item.") + finally: + await shutdown() + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/standards/mcp/templates/app/tests/__init__.py b/standards/mcp/templates/app/tests/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/tests/integration/__init__.py b/standards/mcp/templates/app/tests/integration/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/tests/integration/test_item_routes.py b/standards/mcp/templates/app/tests/integration/test_item_routes.py new file mode 100644 index 0000000..a7ecc0e --- /dev/null +++ b/standards/mcp/templates/app/tests/integration/test_item_routes.py @@ -0,0 +1,21 @@ +import importlib + +import interfaces.openapi as routes_module + +from openapi_first.app import OpenAPIFirstApp + + +def test_openapi_app_builds(): + """The spec-first app must build and plant each route handler.""" + module = importlib.import_module(routes_module.__name__) + for operation in ("create_item", "get_item", "list_items", "delete_item", + "create_job", "get_job", "list_jobs"): + assert hasattr(module, operation) + + app = OpenAPIFirstApp( + openapi_path="openapi.yaml", + routes_module=module, + title="Item API", + version="0.1.0", + ) + assert app is not None \ No newline at end of file diff --git a/standards/mcp/templates/app/tests/unit/__init__.py b/standards/mcp/templates/app/tests/unit/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/standards/mcp/templates/app/tests/unit/test_item_models.py b/standards/mcp/templates/app/tests/unit/test_item_models.py new file mode 100644 index 0000000..10b58ab --- /dev/null +++ b/standards/mcp/templates/app/tests/unit/test_item_models.py @@ -0,0 +1,22 @@ +from domain.item.models import Item +from domain.job.models import Job + + +def test_item_defaults(): + item = Item(name="widget") + assert item.quantity == 1 + + +def test_job_defaults(): + job = Job(name="sync") + assert job.status == "pending" + assert job.retry_count == 0 + assert job.error_message is None + + +def test_job_rejects_unknown_status(): + try: + Job(name="sync", status="exploded") + except Exception: + return + raise AssertionError("status should be rejected") \ No newline at end of file diff --git a/standards/mcp/templates/app/tests/unit/test_item_sync.py b/standards/mcp/templates/app/tests/unit/test_item_sync.py new file mode 100644 index 0000000..6bc2d99 --- /dev/null +++ b/standards/mcp/templates/app/tests/unit/test_item_sync.py @@ -0,0 +1,28 @@ +from types import SimpleNamespace + +from pipelines.item_sync import run_item_sync + + +class FakeItemSvc: + def __init__(self): + self.created = [] + + async def create(self, data: dict): + self.created.append(dict(data)) + + +async def test_run_item_sync_creates_item(): + svc = FakeItemSvc() + job = SimpleNamespace(id="j1", name="widget", payload={"quantity": 4}) + await run_item_sync(job, {"item_svc": svc}) + + assert len(svc.created) == 1 + assert svc.created[0] == {"name": "widget", "quantity": 4} + + +async def test_run_item_sync_uses_payload_name(): + svc = FakeItemSvc() + job = SimpleNamespace(id="j1", name="ignored", payload={"name": "gadget"}) + await run_item_sync(job, {"item_svc": svc}) + + assert svc.created[0]["name"] == "gadget" \ No newline at end of file diff --git a/standards/mcp/templates/app/worker.py b/standards/mcp/templates/app/worker.py new file mode 100644 index 0000000..c7fda13 --- /dev/null +++ b/standards/mcp/templates/app/worker.py @@ -0,0 +1,67 @@ +"""# Summary + +Background worker — polls for pending jobs and runs the item-sync pipeline. +The poll loop is the pipeline axis of the blueprint: resources are created by +the API axis, workers consume them on a fixed cadence. + +Notes: +- This process is separate from the API process. Any in-memory event bus is per + process — if the pipeline must surface live progress to the API, swap the bus + adaptor for a broker-backed one (see standards 07_worker_and_progress). +""" +import asyncio +import logging + +from config import settings +from pipelines.item_sync import run_item_sync + +logging.basicConfig( + level=logging.INFO, + format="%(asctime)s [%(name)s] %(levelname)s: %(message)s", +) + +logger = logging.getLogger(__name__) + + +async def poll_jobs(resources: dict): + """Poll for pending jobs and run the item-sync pipeline on each. + + Args: + resources (dict): Wired services from :func:`init.init`. + """ + job_svc = resources["job_svc"] + logger.info("WORKER STARTED — poll=%ss max_retries=%d", + settings.worker_poll_seconds, settings.worker_max_retries) + + while True: + pending = await job_svc.search(filter={"status": "pending"}, limit=10) + logger.info("Found %d pending job(s)", len(pending) if pending else 0) + + for job in pending: + job_id = str(job.id) + if (job.retry_count or 0) >= settings.worker_max_retries: + logger.warning("SKIPPING job %s — retries exhausted", job_id) + continue + + await job_svc.patch(job_id, {"status": "processing"}) + try: + await run_item_sync(job, resources) + await job_svc.patch(job_id, {"status": "done", "error_message": ""}) + except Exception as exc: # noqa: BLE001 + logger.exception("job %s FAILED: %s", job_id, exc) + await job_svc.patch(job_id, {"status": "failed", "error_message": str(exc)}) + + await asyncio.sleep(settings.worker_poll_seconds) + + +if __name__ == "__main__": + from init import init, shutdown + + async def _run(): + resources = await init() + try: + await poll_jobs(resources) + finally: + await shutdown() + + asyncio.run(_run()) \ No newline at end of file