docs(standards): vendor template into MCP bundle; add component tools + start_app prompt
This commit is contained in:
54
standards/mcp/templates/app/pipelines/shared/loader.py
Normal file
54
standards/mcp/templates/app/pipelines/shared/loader.py
Normal file
@@ -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 {})
|
||||
Reference in New Issue
Block a user