Skip to content

API documentation

Automatically generated API reference of the Mellea Skills Compiler.

mellea_skills_compiler

Mellea Skills Compiler — Agent specification certification pipeline.

Transforms agent skill specifications into policy-governed, certified pipelines with evidence packages documenting risks, compliance posture, and runtime governance.

Modules:

certification

Modules:

  • classification
  • data
  • ingest

    Mellea Skills Compiler — Ingest an OpenClaw SKILL.md and certify it.

  • nexus_policy

    AI Atlas Nexus policy generation for Mellea Skills Compiler.

  • pipeline

    Mellea Skills Compiler — Full Pipeline: SKILL.md → decomposed pipeline → Guardian → certification.

  • report

    Generate a certification report with audit trail evidence showing which requirements are

Functions:

  • skill_to_use_case

    Compose a use-case description from parsed skill + sensitivity.

skill_to_use_case

skill_to_use_case(parsed: dict, sensitivity: dict) -> str

Compose a use-case description from parsed skill + sensitivity.

classification

Functions:

classify_governance_requirements
classify_governance_requirements(manifest: PolicyManifest, nexus) -> ComplianceSummary

Classify each governance requirement as AUTOMATED, PARTIAL, or MANUAL based on pipeline controls from pipeline_control_mappings.yaml.

Parameters:

  • manifest (PolicyManifest) –

    description

  • nexus (_type_) –

    description

Returns:

classify_skill_sensitivity
classify_skill_sensitivity(tools: list[str], body: str) -> dict

Classify skill sensitivity from tool list and body keywords.

── Tool sensitivity classification ───────────────────────────────
Simple lookup: tool name → sensitivity tier + operation types.
Grounded in POSIX permission model (R/W/X) and Google OAuth
3-tier sensitivity (non-sensitive / sensitive / restricted).
PROVENANCE: claude-suggested — validate tool classifications

Returns: tier, operations, and body-detected capabilities.

data

Functions:

Attributes:

PACKAGEDIR module-attribute
PACKAGEDIR = absolute()
get_data_path
get_data_path()

ingest

Mellea Skills Compiler — Ingest an OpenClaw SKILL.md and certify it.

Pipeline
  1. Parse SKILL.md (YAML frontmatter + markdown body)
  2. Classify tool sensitivity tier (lookup table, no regex)
  3. Compose use-case description for Nexus
  4. Nexus risk identification → policy manifest
  5. Compliance classification
  6. Certification report

Functions:

  • ingest_one

    Risk Analysis and Policy Generation Pipeline for Mellea skill

Attributes:

log module-attribute
ingest_one
ingest_one(spec_path: Path, dry_run: bool = False, model: Optional[str] = None, inference_engine: InferenceEngineType = OLLAMA)

Risk Analysis and Policy Generation Pipeline for Mellea skill

Parameters:

  • skill_name (str) –

    Mellea Skill spec path.

  • dry_run (bool, default: False ) –

    Preview without making LLM calls. Defaults to False.

  • model (Optional[str], default: None ) –

    Model to use for Risk and Action Identification. The inference_engine param must support the model. If set to None, the default model for the inference engine will be used.

  • inference_engine (InferenceEngineType, default: OLLAMA ) –

    Service to use for LLM inference. Defaults to InferenceEngineType.OLLAMA.

nexus_policy

AI Atlas Nexus policy generation for Mellea Skills Compiler.

Functions:

Attributes:

LOGGER module-attribute
LOGGER = configure_logger()
generate_policy_manifest
generate_policy_manifest(use_case: str, nexus, model: Optional[str] = None, inference_engine: InferenceEngineType = OLLAMA, governance_taxonomies: Optional[list[str]] = None) -> PolicyManifest

Identify applicable risks and governance actions to produce a multi-taxonomy policy manifest.

Parameters:

  • use_case (str) –

    Natural language description of the agent's purpose.

  • nexus (AIAtlasNexus) –

    AI Atlas Nexus instance.

  • model (Optional[str], default: None ) –

    Model to use for Risk and Action Identification. The inference_engine param must support the model. If set to None, the default model for the inference engine will be used.

  • inference_engine (InferenceEngineType, default: OLLAMA ) –

    Service to use for LLM inference. Defaults to InferenceEngineType.OLLAMA.

  • governance_taxonomies (Optional[list[str]], default: None ) –

    List of taxonomy IDs for governance actions.

Returns: PolicyManifest with Guardian risks and governance actions.

generate_policy_markdown
generate_policy_markdown(manifest: PolicyManifest) -> str

Generate a human-readable .md policy document from a manifest.

Includes
  • Guardian runtime risk checks (what the hooks monitor)
  • NIST AI RMF governance actions (organisational requirements)
  • Credo UCF controls (specific mitigation measures)
  • Guardrail configuration table
  • Audit trail specification
load_policy_manifest
load_policy_manifest(audit_dir: Path) -> PolicyManifest

Search for policy_manifest.json in standard locations.

Checks the skill root first (portable), then the audit directory.

pipeline

Mellea Skills Compiler — Full Pipeline: SKILL.md → decomposed pipeline → Guardian → certification.

End-to-end demonstration
  1. Ingest a SKILL.md → parse, classify sensitivity, compose use-case
  2. Identify risks via AI Atlas Nexus → policy manifest
  3. Configure Guardian hooks from manifest (pre + post generation)
  4. Run test fixtures through the DECOMPOSED pipeline → Guardian intercepts every m.instruct() call inside the pipeline → Audit trail captures every generation + Guardian verdict
  5. Compliance classification
  6. Certification report with runtime evidence

Functions:

Attributes:

LOGGER module-attribute
LOGGER = configure_logger()
console module-attribute
console = Console(log_time=True)
full_pipeline
full_pipeline(pipeline_dir: Path, fixture_id: Optional[str] = None, enforce: bool = False, model: Optional[str] = None, guardian_model: Optional[str] = None, inference_engine: InferenceEngineType = OLLAMA)

Full Certification Pipeline for Mellea skill

Parameters:

  • pipeline_dir (Path) –

    Compiled Mellea skill pipeline directory.

  • fixture_id (Optional[str], default: None ) –

    Specify a fixture to run with the certification process. Defaults to None.

  • enforce (bool, default: False ) –

    Run pipeline in enforce mode (block on risk detection). Defaults to False.

  • model (Optional[str], default: None ) –

    Model to use for Risk and Action Identification. The inference_engine param must support the model. If set to None, the default model for the inference engine will be used.

  • guardian_model (Optional[str], default: None ) –

    Model to use for Risk Assessment. The inference_engine param must support the model. If set to None, the default guardian model for the inference engine will be used.

  • inference_engine (InferenceEngineType, default: OLLAMA ) –

    Service to use for LLM inference. Defaults to InferenceEngineType.OLLAMA.

skill_pipeline
skill_pipeline(pipeline_dir: Path, fixture_id: str, enforce: bool = False, no_guardian: bool = False)

report

Generate a certification report with audit trail evidence showing which requirements are satisfied and how.

Functions:

Attributes:

EVIDENCE_EXTRACTORS module-attribute
EVIDENCE_EXTRACTORS: dict[str, Any] = {'pc-content-safety': _extract_guardian_evidence, 'pc-input-screening': _extract_io_evidence, 'pc-generation-audit': _extract_audit_evidence, 'pc-component-audit': _extract_audit_evidence, 'pc-validation-audit': _extract_audit_evidence, 'pc-latency-monitoring': _extract_performance_evidence, 'pc-incident-flagging': _extract_incident_evidence, 'pc-policy-traceability': _extract_audit_evidence}
generate_certification_report
generate_certification_report(manifest: PolicyManifest, compliance: ComplianceSummary, audit_trail: list[dict], audit_path: str = '') -> str

Generate a Markdown certification report with evidence.

Parameters:

Returns:

  • str ( str ) –

    description

load_audit_trail
load_audit_trail(path: str | Path) -> list[dict]

Load JSONL audit trail into a list of dicts.

cli

Functions:

  • certify

    Run full certification pipeline on a compiled skill.

  • compile

    Compile Mellea skill specification into a Mellea pipeline using mellea-fy Claude command.

  • export

    Export a compiled Mellea skill to a deployment target (langgraph, claude-code, or mcp).

  • ingest

    Run risk analysis on a skill specification.

  • main

    Mellea Skills Compiler CLI - Agent specification certification pipeline.

  • run

    Run Mellea Skill Pipeline.

  • signal_handler
  • validate

    Run Step 7 structural lints and (unless --no-run) the fixture smoke-check.

Attributes:

LOGGER module-attribute

LOGGER = configure_logger()

app module-attribute

app = Typer(no_args_is_help=True)

certify

certify(ctx: Context, pipeline_dir: Annotated[str, Argument(help='Compiled skill pipeline directory.', rich_help_panel=Arguments)], fixture: Annotated[Optional[str], Option(--fixture, -f, help='Run pipeline for a specific fixture.')] = None, enforce: Annotated[bool, Option(--enforce, -e, help='Run pipeline in enforce mode (block on risk detection)')] = False, model: Annotated[Optional[str], Option(--model, -m, help='Model to use for Risk and Action Identification. The `--inference-engine` option must support the model. If set to None, the default model for the inference engine will be used.')] = None, guardian_model: Annotated[Optional[str], Option(--guardian - model, -g, help='Model to use for Risk Assessment. The `--inference-engine` option must support the model. If set to None, the default guardian model for the inference engine will be used.')] = None, inference_engine: Annotated[Literal['ollama'], Option(--inference - engine, -i, callback=lambda x: upper(), help='Service to use for LLM inference. Supported: ollama')] = 'ollama')

Run full certification pipeline on a compiled skill.

Performs comprehensive certification including risk identification, compliance classification, and runtime Guardian checks.

compile

compile(ctx: Context, spec_path: Annotated[str, Argument(help='Path to the Mellea skill specification or the folder containing the skill specs.')], model: Annotated[str, Option(--model, -m, help='Claude model for compiling a skill. Default to Sonnet.')] = None, timeout: Annotated[int, Option(--timeout, -t, help='Claude session timeout in seconds. Default to 4500 (75min).')] = 4500, repair_mode: Annotated[bool, Option(-r, --repair - mode, help='Identify and correct any errors effectively in Mellea skill compilation.')] = False, no_run: Annotated[bool, Option(--no - run, help='Skip the post-compile fixture smoke-check (default ON).')] = False, refresh_cache: Annotated[bool, Option(--refresh - cache, help='Force refresh of grounding caches (~/.cache/mellea-skills-compiler/) before compile.')] = False, skill_backend: Annotated[Optional[str], Option(--skill - backend, help='Override the LLM backend used by the compiled skill at runtime (default from mellea_skills_compiler/compile/claude/data/runtime_defaults.json).')] = None, skill_model: Annotated[Optional[str], Option(--skill - model, help='Override the model id used by the compiled skill at runtime (default from mellea_skills_compiler/compile/claude/data/runtime_defaults.json).')] = None)

Compile Mellea skill specification into a Mellea pipeline using mellea-fy Claude command.

After mellea-fy returns successfully, automatically chains into validate: runs the structural lints (Step 7) and (unless --no-run) executes one fixture against the LLM backend. A green compile means compiled + lints passed + smoke-check passed (or skipped because backend was unreachable).

export

export(ctx: Context, package_path: Annotated[str, Argument(help='Path to the compiled skill directory (must contain melleafy.json).')], target: Annotated[str, Option(--target, -t, help='Deployment target: langgraph | claude-code | mcp')], force: Annotated[bool, Option(--force, -f, help='Overwrite output directory if it already exists.')] = False)

Export a compiled Mellea skill to a deployment target (langgraph, claude-code, or mcp).

Output is written to /- inside the skill directory.

NOTE: This command is experimental. Output structure and CLI interface may change in future releases without a deprecation period.

ingest

ingest(ctx: Context, spec_path: Annotated[str, Argument(help='Mellea Skill spec path', rich_help_panel=Arguments)], dry_run: Annotated[bool, Option(--dry - run, help='Preview without making LLM calls', show_default=True)] = False, model: Annotated[Optional[str], Option(--model, -m, help='Model to use for Risk and Action Identification. The `--inference-engine` option must support the model. If set to None, the default model for the inference engine will be used.')] = None, inference_engine: Annotated[Literal['ollama'], Option(--inference - engine, -i, callback=lambda x: upper(), help='Service to use for LLM inference. Supported: ollama')] = 'ollama')

Run risk analysis on a skill specification.

Analyzes the skill spec to identify risks and generate a policy document without running full certification.

main

main() -> None

Mellea Skills Compiler CLI - Agent specification certification pipeline.

Transform AI agent skill specifications into certified, governed pipelines with comprehensive risk analysis and compliance reporting.

run

run(ctx: Context, pipeline_dir: Annotated[str, Argument(help='Compiled skill pipeline directory.')], fixture: Annotated[str, Option(--fixture, -f, help='Run pipeline for a specific fixture.')], enforce: Annotated[bool, Option(--enforce, -e, help='Block execution when Guardian detects risks (default: audit-only).')] = False, no_guardian: Annotated[bool, Option(--no - guardian, -ng, help='Skip Guardian checks even if a policy manifest exists.')] = False)

Run Mellea Skill Pipeline.

signal_handler

signal_handler(sig, frame)

validate

validate(ctx: Context, pipeline_dir: Annotated[str, Argument(help='Compiled skill pipeline directory (e.g. skills/<name>/<name>_mellea/).')], no_run: Annotated[bool, Option(--no - run, help='Skip the post-lint fixture smoke-check (default ON).')] = False, all_fixtures: Annotated[bool, Option(--all, help='Run all fixtures (default: first only).')] = False)

Run Step 7 structural lints and (unless --no-run) the fixture smoke-check.

Exit codes

0 — lints passed, smoke-check passed or skipped (backend unreachable) 11 — at least one lint failed 12 — smoke-check failed (lint pass, but a fixture raised an exception)

compile

Modules:

Attributes:

CLAUDE_DIR module-attribute

CLAUDE_DIR = absolute() / 'claude'

claude_directives

Functions:

Attributes:

LOGGER module-attribute
LOGGER = configure_logger()
build_system_prompt
build_system_prompt(backend: str, model_id: str, source: str, package_name: str) -> str

Assemble the instruction string passed to the mellea-fy slash command.

Combines the existing autonomous-run directive with the resolved backend and model_id values, plus a notice listing the paths the wrapper renders authoritatively from JSON (the LLM cannot Write/Edit them — denied via --settings). This tells the LLM to emit the IR JSON instead of trying to write the rendered source itself, so the first attempt is correct rather than denied-then-retried.

package_name is the wrapper-derived Rule OUT-2 name (e.g. weather_mellea) substituted into the prompt body so the LLM uses the same name the wrapper computes for mirror_companion_dirs and the post-session *_mellea discovery. Previously the prompt carried the literal placeholder <package_name> and the LLM re-derived the name from the frontmatter — a long/complex name (e.g. gdpr-breach-sentinel-oliver-schmidt-prietz) could be mis-transcribed and produce a stray sibling directory.

derive_package_name
derive_package_name(spec_path: Path, frontmatter: dict | None) -> str

Apply Rule OUT-2 (lowercase, hyphens/spaces → underscores, append _mellea).

For .md sources, prefer the frontmatter name: field; fall back to the parent directory name. For directory inputs (multi-file runtimes), use the directory name.

mirror_companion_dirs
mirror_companion_dirs(skill_dir: Path, package_dir: Path) -> list[str]

Rule OUT-6 — mirror companion directories from skill root into the package.

Runs deterministically before mellea-fy so the LLM sees the mirrored assets in / when it generates code, reinforcing the package-relative path-resolution invariant. Returns the list of directory names actually mirrored (for logging).

resolve_runtime_defaults
resolve_runtime_defaults(skill_backend_override: Optional[str], skill_model_override: Optional[str]) -> tuple[str, str, str]

Pick the backend and model_id to bake into the compiled skill.

Precedence: CLI overrides win, then the defaults file, then a built-in fallback if the file is missing or unreadable. Returns (backend, model_id, source) where source records where the values came from for the compile log.

write_compile_settings
write_compile_settings(intermediate_dir: Path, package_dir: Path) -> Path

Write a per-invocation Claude Code --settings file with the deny rules.

Path-scoped Write/Edit denies prevent the LLM from clobbering files the wrapper will render after the slash command exits. acceptEdits permission mode does NOT override deny — Anthropic's evaluation order is deny → mode → allow (verified end-to-end in the synthetic deny-rule test).

Important: the deny rule paths must match what the LLM actually tries to write. Claude Code interprets relative paths against the subprocess cwd (the user's shell cwd, typically the repo root). So the deny path must be the full path-from-cwd to the package, not just the bare package name. Earlier versions used package_name/<rel>, which mis-scoped the deny to <cwd>/<package_name>/<rel> and confused the LLM into creating the package at the wrong location.

write_runtime_directive
write_runtime_directive(intermediate_dir: Path, backend: str, model_id: str, source: str) -> Path

Save the chosen backend and model_id alongside the package.

The post-compile lint reads this file to confirm the generated config.py actually used these values (and didn't drift from what we instructed).

grounding

Deterministic grounding artifact generation for the mellea-fy pipeline.

This module produces mellea_api_ref.json and mellea_doc_index.json in the compile pipeline's intermediate directory before the slash command runs. The slash command itself runs with --allowed-tools Read,Write,Edit and so cannot introspect the installed mellea package or fetch https://docs.mellea.ai/. Doing it here means Steps 2.5e and 2.5f of the slash command have real grounding data to consume rather than silently degrading to static fallbacks.

Both functions are idempotent and cache results under ~/.cache/mellea-skills-compiler/. The api_ref cache is keyed by the installed mellea version; the doc_index cache uses a configurable TTL (default 24h) with a stale-cache fallback if the network is unreachable.

Functions:

Attributes:

CACHE_DIR module-attribute
CACHE_DIR = home() / '.cache' / 'mellea-skills-compiler'
CORE_MODULES module-attribute
CORE_MODULES = {'mellea.stdlib.requirements', 'mellea.stdlib.sampling', 'mellea.backends.model_options'}
LOGGER module-attribute
LOGGER = configure_logger()
write_mellea_api_ref
write_mellea_api_ref(intermediate_dir: Path, refresh: bool = False) -> Path

Write mellea_api_ref.json to intermediate_dir.

Cached by installed mellea version under ~/.cache/mellea-skills-compiler/api_ref_<version>.json. If mellea is not installed, writes the grounding_unavailable: true shape and returns.

write_mellea_doc_index
write_mellea_doc_index(intermediate_dir: Path, refresh: bool = False, ttl_hours: int = 24) -> Path

Write mellea_doc_index.json to intermediate_dir.

Cached at ~/.cache/mellea-skills-compiler/doc_index.json with a ttl_hours TTL (default 24h). On fetch failure, reuses a stale cache if one exists; otherwise writes the static 2026-04-28 fallback list.

lints

Step 7 structural lints — Python implementations.

AST-precise structural checks where LLM application has proven unreliable and the drift modes have caused post-compile breakage in shipped skills. The remaining lints documented in .claude/commands/mellea-fy-validate.md are still applied by the LLM during Step 7 of the slash command.

Classes:

Functions:

Attributes:

LintFailure dataclass
LintFailure(file: str, line: Optional[int], column: Optional[int], message: str, rule_ref: Optional[str] = None)

Attributes:

column instance-attribute
column: Optional[int]
file instance-attribute
file: str
line instance-attribute
line: Optional[int]
message instance-attribute
message: str
rule_ref class-attribute instance-attribute
rule_ref: Optional[str] = None
LintResult dataclass
LintResult(lint_id: str, verdict: str, files_checked: int = 0, skipped_reason: Optional[str] = None, failures: List[LintFailure] = list())

Attributes:

failures class-attribute instance-attribute
failures: List[LintFailure] = field(default_factory=list)
files_checked class-attribute instance-attribute
files_checked: int = 0
lint_id instance-attribute
lint_id: str
skipped_reason class-attribute instance-attribute
skipped_reason: Optional[str] = None
verdict instance-attribute
verdict: str
LintRunResult dataclass
LintRunResult(overall_verdict: str, lints: List[LintResult], package_path: str, checked_at: str)

Attributes:

checked_at instance-attribute
checked_at: str
failed property
failed: bool
lints instance-attribute
lints: List[LintResult]
overall_verdict instance-attribute
overall_verdict: str
package_path instance-attribute
package_path: str
lint_bundled_asset_path_resolution
lint_bundled_asset_path_resolution(package_dir: Path) -> LintResult

Reject any path-join construction that resolves a bundled-asset path via anything other than Path(__file__).parent.

lint_fixture_pydantic_coercion
lint_fixture_pydantic_coercion(package_dir: Path) -> LintResult

Fixture inputs must not pass bare dicts where run_pipeline expects Pydantic models.

lint_fixtures_loader_contract
lint_fixtures_loader_contract(package_dir: Path) -> LintResult

<package>/fixtures/__init__.py must export ALL_FIXTURES or FIXTURES.

lint_format_annotation
lint_format_annotation(package_dir: Path) -> LintResult

m.instruct(...) calls whose result is parsed must include format=.

Detection
  1. AST-parse pipeline.py / slots.py / constrained_slots.py.
  2. For every Call that is m.instruct(...) assigned to a Name target (thunk = m.instruct(...)), record whether it has format= and which variable holds the result.
  3. Walk for subsequent uses of that variable as:
    • argument to <Model>.model_validate_json(<thunk>.value)
    • first positional arg to _parse_instruct_result(<thunk>, ...)
    • first positional arg to _safe_parse_with_fallback(<thunk>, ...)
  4. If any such parse-use is found AND the original instruct call had no format= kwarg → hard failure on the m.instruct call line.
lint_grounding_context_types
lint_grounding_context_types(package_dir: Path) -> LintResult

Every grounding_context= dict-literal value should be a string.

Verdict
  • fail if any definite-collection violation
  • warning if only ambiguous findings
  • pass otherwise
lint_import_side_effects
lint_import_side_effects(package_dir: Path) -> LintResult

No module-level Calls outside the documented allowlist.

Bare ast.Expr whose value is a Call (e.g. load_dotenv() at the top of a file) executes at import time; flagged. Module-level assignments where the RHS is a Call are flagged only when the callee is known to be side-effecting (network/env mutation) — allowing safe patterns like LOGGER = logging.getLogger(__name__).

lint_import_soundness
lint_import_soundness(package_dir: Path) -> LintResult

Every mellea.* import path must exist in mellea_api_ref.json:.modules.

lint_instruct_result_parse_before_access
lint_instruct_result_parse_before_access(package_dir: Path) -> LintResult

KB1: m.instruct(format=Model) returns a thunk, not a Pydantic model.

The thunk MUST be parsed before any field access. Exempt patterns: - _parse_instruct_result(thunk, M) - _safe_parse_with_fallback(thunk, M, **defaults) - M.model_validate_json(thunk.value)

lint_parseable
lint_parseable(package_dir: Path) -> LintResult

Every .py file under the package must parse, and <pkg>.pipeline must import.

Two checks, executed in order:

  1. Per-file parseability: ast.parse(p.read_text()) for every .py file directly under <package_dir>/ and <package_dir>/fixtures/.
  2. Package importability: python -c "import <package_name>.pipeline" executed as a clean subprocess from <package_dir>.parent. This surfaces wrong external import paths (e.g. mellea.stdlib.strategies instead of mellea.stdlib.sampling) that ast.parse() cannot detect.

Why a subprocess: importing the package directly in the lint process would pollute namespace, risk hanging on import side effects, or get masked by cached sys.modules entries. A clean child process surfaces exactly the error a downstream consumer (pip install + python -c) would see.

Missing pipeline.py: returns one ERROR failure naming the absent entry module. Tier 1 should hard-fail loudly so the slash-command gate halts before Tier ⅔ lints run on a package with no entry.

Scope: files directly under <package_dir>/ and <package_dir>/fixtures/ only.

lint_prefix_persona
lint_prefix_persona(package_dir: Path) -> LintResult

m.instruct(prefix=<config_constant>) is misuse (KB7).

prefix= is for structured-output continuation; not for persona / system-prompt injection.

lint_runtime_defaults_bound
lint_runtime_defaults_bound(package_dir: Path) -> LintResult

The compiled config.py must use the backend and model_id we instructed.

Compares the BACKEND and MODEL_ID constants in /config.py against the values recorded at /intermediate/runtime_directive.json (which captures what the compile pipeline injected via the system prompt). If the LLM ignored the directive and emitted different values, this fails loudly.

Skipped when the directive file is absent — usually because the package was compiled with an older pipeline that did not write the directive.

lint_session_boundary
lint_session_boundary(package_dir: Path) -> LintResult

Each start_session(...) block must use at most one distinct format type.

lint_session_method_arity
lint_session_method_arity(package_dir: Path) -> LintResult

MelleaSession method calls must supply required positional arguments.

lint_stdlib_arg_types
lint_stdlib_arg_types(package_dir: Path) -> LintResult

Flag clearly-non-dict arguments passed to Mellea API grounding_context= kwargs.

Narrow MVP scope: only checks grounding_context= on session-method calls in the instruct/chat/act family.

lint_validation_fn_not_called_directly
lint_validation_fn_not_called_directly(package_dir: Path) -> LintResult

Reject any direct call to <req>.validation_fn(...) in generated code.

run_lints
run_lints(package_dir: Path) -> LintRunResult

Run all implemented Step 7 structural lints; write step_7_report.json.

mellea_skills

Functions:

  • compile
  • validate

    Shared implementation for the validate command and the compile auto-chain.

Attributes:

LOGGER module-attribute
LOGGER = configure_logger()
console module-attribute
console = Console(log_time=True)
compile
compile(spec_path: Path, model: Optional[str] = None, timeout: int = 4500, repair_mode: bool = False, no_run: bool = False, refresh_cache: bool = False, skill_backend: Optional[str] = None, skill_model: Optional[str] = None) -> None
validate
validate(package_dir: Path, *, no_run: bool, all_fixtures: bool) -> None

Shared implementation for the validate command and the compile auto-chain.

models

Classes:

RenderResult dataclass
RenderResult(name: str, status: str, detail: Optional[str] = None, diff_lines_added: int = 0, diff_lines_removed: int = 0)

Attributes:

detail class-attribute instance-attribute
detail: Optional[str] = None
diff_lines_added class-attribute instance-attribute
diff_lines_added: int = 0
diff_lines_removed class-attribute instance-attribute
diff_lines_removed: int = 0
name instance-attribute
name: str
status instance-attribute
status: str
WriterSpec dataclass
WriterSpec(name: str, emission_relpath: str, output_relpath: str, writer: ModuleType, output_kind: str = 'file')

One renderable artifact: emission JSON in → Python source on disk out.

output_kind controls how the writer is invoked: - "file" (default): writer module exposes render(emission) -> str and the renderer writes the returned text to output_relpath. - "directory": writer module exposes write(emission, dir_path) -> list[Path] and the renderer wipes output_relpath (preserving the dir itself) before invoking, so the writer is the sole producer of the dir's contents.

Attributes:

emission_relpath instance-attribute
emission_relpath: str
name instance-attribute
name: str
output_kind class-attribute instance-attribute
output_kind: str = 'file'
output_relpath instance-attribute
output_relpath: str
writer instance-attribute
writer: ModuleType

proxy

Classes:

ContextMgmtStrippingProxy

Bases: BaseHTTPRequestHandler

Local proxy that strips context_management from Anthropic API requests.

The IBM LiteLLM proxy rejects context_management as an unknown parameter. Claude Code sends it automatically. This proxy intercepts requests from the claude subprocess, strips the field, and forwards to the real upstream.

Methods:

do_GET
do_GET()
do_POST
do_POST()
handle_error
handle_error(request, client_address)
log_message
log_message(fmt, *args)

smoke_check

Post-lint fixture smoke-check (Step 7's --run mode).

Executes one fixture (or all, if all_fixtures=True) from the compiled package and classifies the outcome into one of three verdicts:

  • passed — fixture executed without exception.
  • failed — fixture raised an exception other than backend-unreachable. Exit code 12 (distinct from lint failure exit code 11).
  • skipped — LLM backend unreachable (Ollama not running, timeout, auth error). Exit code 0 with a stderr warning so CI without an LLM stays green while still nudging local users.

Writes intermediate/step_7b_report.json. Does NOT trigger the repair loop — fixture failures require human review per mellea-fy-validate.md.

Classes:

Functions:

  • run_smoke_check

    Execute one (or all) fixtures and write step_7b_report.json.

Attributes:

LOGGER module-attribute
LOGGER = configure_logger()
SmokeFixtureResult dataclass
SmokeFixtureResult(fixture_id: str, verdict: str, duration_seconds: float, skipped_reason: Optional[str] = None, failure_message: Optional[str] = None, failure_traceback: Optional[str] = None)

Attributes:

duration_seconds instance-attribute
duration_seconds: float
failure_message class-attribute instance-attribute
failure_message: Optional[str] = None
failure_traceback class-attribute instance-attribute
failure_traceback: Optional[str] = None
fixture_id instance-attribute
fixture_id: str
skipped_reason class-attribute instance-attribute
skipped_reason: Optional[str] = None
verdict instance-attribute
verdict: str
SmokeRunResult dataclass
SmokeRunResult(overall_verdict: str, fixtures: List[SmokeFixtureResult] = list(), package_path: str = '', checked_at: str = '')

Attributes:

checked_at class-attribute instance-attribute
checked_at: str = ''
exit_code property
exit_code: int
fixtures class-attribute instance-attribute
fixtures: List[SmokeFixtureResult] = field(default_factory=list)
overall_verdict instance-attribute
overall_verdict: str
package_path class-attribute instance-attribute
package_path: str = ''
run_smoke_check
run_smoke_check(package_dir: Path, all_fixtures: bool = False) -> SmokeRunResult

Execute one (or all) fixtures and write step_7b_report.json.

Parameters:

  • package_dir (Path) –

    compiled package directory (e.g. <skill>/<name>_mellea).

  • all_fixtures (bool, default: False ) –

    if False (default), only the first fixture in ALL_FIXTURES is executed. If True, every fixture is run sequentially.

writers

Modules:

  • config_writer

    Deterministic writer: config_emission JSON → config.py Python source.

  • fixtures_writer

    Deterministic writer: fixtures_emission JSON → fixtures/ subpackage.

  • renderer

    The writers (config_writer.py, fixtures_writer.py, ...) were originally

config_writer

Deterministic writer: config_emission JSON → config.py Python source.

Input must conform to .claude/schemas/config_emission.schema.json. Output is ready to write directly to /config.py.

Functions:

  • render

    Render a config_emission JSON object as Python source.

  • write

    Write rendered config.py source to path.

render
render(emission: dict[str, Any] | str) -> str

Render a config_emission JSON object as Python source.

Parameters:

  • emission (dict[str, Any] | str) –

    dict conforming to config_emission.schema.json, or a JSON string.

Returns:

  • str

    Python source string with constants grouped by C-category and annotated

  • str

    with PROVENANCE comments where source location is available.

write
write(emission: dict[str, Any] | str, path: str | Path) -> None

Write rendered config.py source to path.

fixtures_writer

Deterministic writer: fixtures_emission JSON → fixtures/ subpackage.

Input must conform to .claude/schemas/fixtures_emission.schema.json. Output is a set of files written into /fixtures/:

fixtures/__init__.py        — exports ALL_FIXTURES: list[Callable]
fixtures/<id>.py            — one per fixture, defines make_<id>() factory

Each factory is a zero-arg callable that returns (inputs: dict, fixture_id: str, description: str) — the contract enforced by mellea_skills_compiler.toolkit.file_utils.load_fixtures.

The model never writes Python source for fixtures directly. This makes drift shapes (pytest-style tests, INPUT-only modules, etc.) structurally unreachable.

Functions:

  • render_fixture

    Render a single fixture spec as the contents of fixtures/.py.

  • render_init

    Render fixtures/init.py exporting ALL_FIXTURES.

  • write

    Write the rendered fixtures/ subpackage to fixtures_dir.

render_fixture
render_fixture(fixture: dict[str, Any]) -> str

Render a single fixture spec as the contents of fixtures/.py.

render_init
render_init(fixtures: list[dict[str, Any]], coverage_doc: str | None = None) -> str

Render fixtures/init.py exporting ALL_FIXTURES.

write
write(emission: dict[str, Any] | str, fixtures_dir: str | Path) -> list[Path]

Write the rendered fixtures/ subpackage to fixtures_dir.

Parameters:

  • emission (dict[str, Any] | str) –

    dict conforming to fixtures_emission.schema.json, or a JSON string.

  • fixtures_dir (str | Path) –

    target directory (typically /fixtures/). Created if absent.

Returns:

  • list[Path]

    List of paths written, in deterministic order: init.py first, then per-fixture files.

renderer

The writers (config_writer.py, fixtures_writer.py, ...) were originally documented as "the LLM follows them" — but the slash command runs with --allowed-tools Read,Write,Edit and cannot execute Python. So in practice the LLM was mentally rendering the writer's output, which is unreliable for anything more complex than config.py.

This module moves writer invocation into the compile pipeline, where Python actually runs. Two operating modes:

  • WARN (observation only): render via the writer, diff against the file the LLM wrote, log a warning if they differ, do NOT overwrite.
  • ENFORCE: render via the writer, write authoritatively, deletes whatever the LLM put there. Combine with Write(...) deny rules in --settings to prevent the LLM writing the path in the first place.
Migration plan
  1. WARN mode for config_writer (this commit) — gather evidence the diffs are stable and bounded across compiles.
  2. After ~5 clean compiles, flip config_writer to ENFORCE + add deny rule.
  3. Repeat for fixtures_writer.

Each writer module exposes a render(emission: dict) -> str function returning Python source text. The wrapper reads the emission JSON the LLM emitted, calls render(), and either compares-and-warns or writes-authoritatively.

Functions:

  • render_writers

    Render every writer in specs against the package and log outcomes.

Attributes:

DEFAULT_WRITER_SPECS module-attribute
DEFAULT_WRITER_SPECS = [WriterSpec(name='config.py', emission_relpath='intermediate/config_emission.json', output_relpath='config.py', writer=config_writer), WriterSpec(name='fixtures/', emission_relpath='intermediate/fixtures_emission.json', output_relpath='fixtures', output_kind='directory', writer=fixtures_writer)]
LOGGER module-attribute
LOGGER = configure_logger()
render_writers
render_writers(package_dir: Path, *, enforce: bool = False) -> List[RenderResult]

Render every writer in specs against the package and log outcomes.

With enforce=False (default for migration phase), only logs WARN on diff and never overwrites. With enforce=True, the writer's output becomes the authoritative source for the artifact.

enums

Classes:

ClaudeResponseMessageType

Bases: StrEnum

Attributes:

TEXT class-attribute instance-attribute
TEXT = auto()

ClaudeResponseType

Bases: StrEnum

Attributes:

ASSISTANT class-attribute instance-attribute
ASSISTANT = auto()
SYSTEM class-attribute instance-attribute
SYSTEM = auto()

CoverageLevel

Bases: Enum

Attributes:

AUTOMATED class-attribute instance-attribute
AUTOMATED = 'AUTOMATED'
MANUAL class-attribute instance-attribute
MANUAL = 'MANUAL'
PARTIAL class-attribute instance-attribute
PARTIAL = 'PARTIAL'

GovernanceTaxonomy

Bases: StrEnum

Default taxonomy identifiers

Methods:

Attributes:

CREDO_UCF class-attribute instance-attribute
CREDO_UCF = 'credo-ucf'
IBM_GRANITE_GUARDIAN class-attribute instance-attribute
IBM_GRANITE_GUARDIAN = 'ibm-granite-guardian'
NIST_AI_RMF class-attribute instance-attribute
NIST_AI_RMF = 'nist-ai-rmf'
list classmethod
list()

InferenceEngineType

Bases: Enum

Enum to contain possible values for inference engine types

Methods:

Attributes:

OLLAMA class-attribute instance-attribute
OLLAMA = auto()
list classmethod
list()

InferenceModel

Bases: StrEnum

Default model identifiers

Attributes:

CLAUDE_MODEL class-attribute instance-attribute
CLAUDE_MODEL = 'sonnet'
OLLAMA_GUARDIAN_MODEL class-attribute instance-attribute
OLLAMA_GUARDIAN_MODEL = 'ibm/granite3.3-guardian:8b'
OLLAMA_RISK_MODEL class-attribute instance-attribute
OLLAMA_RISK_MODEL = 'granite3.3:8b'

SpecFileFormat

Bases: StrEnum

Default spec file identifiers

Attributes:

SKILL_FILE_MD class-attribute instance-attribute
SKILL_FILE_MD = 'SKILL.md'
SPEC_FILE_MD class-attribute instance-attribute
SPEC_FILE_MD = 'spec.md'

export

Modules:

  • exporter

    Export pipeline: 5 stages (validate → load → translate → emit → lint).

  • targets

exporter

Export pipeline: 5 stages (validate → load → translate → emit → lint).

Classes:

Functions:

  • run_export
  • stage1_validate

    Read and validate melleafy.json. Returns raw manifest dict.

  • stage2_load

    Parse manifest fields, derive entry point, detect policy manifest.

  • stage3_translate

    Build TranslationPlan from LoadedContext. Dispatches to target module.

  • stage4_emit

    Write all files. Uses .partial/ pattern for atomic write.

  • stage5_lint

    Basic structural lint on emitted files. Halts with exit code 4 on failure.

Attributes:

EXPORT_VERSION module-attribute
EXPORT_VERSION = '0.1.0'
MIN_MANIFEST_VERSION module-attribute
MIN_MANIFEST_VERSION = Version('1.0.0')
SUPPORTED_TARGETS module-attribute
SUPPORTED_TARGETS = {'langgraph', 'claude-code', 'mcp'}
AdapterFile dataclass
AdapterFile(relative_path: str, content: str)

Attributes:

content instance-attribute
content: str
relative_path instance-attribute
relative_path: str
EmitResult dataclass
EmitResult(out_path: Path, files_written: int, bytes_written: int)

Attributes:

bytes_written instance-attribute
bytes_written: int
files_written instance-attribute
files_written: int
out_path instance-attribute
out_path: Path
Invocation dataclass
Invocation(package_path: Path, target: str, out_path: Optional[Path] = None, force: bool = False)

Attributes:

force class-attribute instance-attribute
force: bool = False
out_path class-attribute instance-attribute
out_path: Optional[Path] = None
package_path instance-attribute
package_path: Path
target instance-attribute
target: str
LoadedContext dataclass
LoadedContext(invocation: Invocation, manifest: dict, package_source_dir: Path, python_package_dir: Path, supporting_asset_dirs: list[Path], entry_module: str, sig: ParsedSignature, load_warnings: list[str] = list(), policy_manifest_path: Optional[Path] = None)

Attributes:

entry_module instance-attribute
entry_module: str
invocation instance-attribute
invocation: Invocation
load_warnings class-attribute instance-attribute
load_warnings: list[str] = field(default_factory=list)
manifest instance-attribute
manifest: dict
package_source_dir instance-attribute
package_source_dir: Path
policy_manifest_path class-attribute instance-attribute
policy_manifest_path: Optional[Path] = None
python_package_dir instance-attribute
python_package_dir: Path
sig instance-attribute
supporting_asset_dirs instance-attribute
supporting_asset_dirs: list[Path]
ParsedSignature dataclass
ParsedSignature(function_name: str, params: list[dict], return_type: str, pattern: str)

Parsed from manifest entry_signature string.

Attributes:

function_name instance-attribute
function_name: str
params instance-attribute
params: list[dict]
pattern instance-attribute
pattern: str
return_type instance-attribute
return_type: str
TranslationPlan dataclass
TranslationPlan(graph_name: str, adapter_files: list[AdapterFile], bundled_package_name: str, warnings: list[str] = list(), deployment_guidance: str = '')

Attributes:

adapter_files instance-attribute
adapter_files: list[AdapterFile]
bundled_package_name instance-attribute
bundled_package_name: str
deployment_guidance class-attribute instance-attribute
deployment_guidance: str = ''
graph_name instance-attribute
graph_name: str
warnings class-attribute instance-attribute
warnings: list[str] = field(default_factory=list)
run_export
run_export(inv: Invocation) -> EmitResult
stage1_validate
stage1_validate(inv: Invocation) -> dict

Read and validate melleafy.json. Returns raw manifest dict.

stage2_load
stage2_load(inv: Invocation, manifest: dict) -> LoadedContext

Parse manifest fields, derive entry point, detect policy manifest.

stage3_translate
stage3_translate(loaded: LoadedContext) -> TranslationPlan

Build TranslationPlan from LoadedContext. Dispatches to target module.

stage4_emit
stage4_emit(plan: TranslationPlan, loaded: LoadedContext) -> EmitResult

Write all files. Uses .partial/ pattern for atomic write.

stage5_lint
stage5_lint(result: EmitResult, loaded: LoadedContext, plan: TranslationPlan) -> None

Basic structural lint on emitted files. Halts with exit code 4 on failure.

targets

Modules:

  • claude_code

    Claude Code target: translate a LoadedContext into a TranslationPlan.

  • langgraph

    LangGraph target: translate a LoadedContext into a TranslationPlan.

  • mcp

    MCP target: translate a LoadedContext into a TranslationPlan.

claude_code

Claude Code target: translate a LoadedContext into a TranslationPlan.

Supported modalities

synchronous_oneshot — sync Python subprocess via scripts/run.sh streaming — unbuffered async streaming via scripts/run.sh conversational_session — session carry-forward via --session JSON arg

Functions:

Attributes:

SUPPORTED_MODALITIES module-attribute
SUPPORTED_MODALITIES = {'synchronous_oneshot', 'streaming', 'conversational_session'}
translate_claude_code
translate_claude_code(loaded: 'LoadedContext') -> 'TranslationPlan'
langgraph

LangGraph target: translate a LoadedContext into a TranslationPlan.

Supported modalities

synchronous_oneshot — async node wrapping sync pipeline via asyncio.to_thread streaming — async node with get_stream_writer(), graph.astream_events() conversational_session — async node, MemorySaver checkpointer, thread_id config scheduled — async node + langgraph.json schedules block event_triggered — async node, event dict state, platform webhook heartbeat — async node, MemorySaver + HEARTBEAT_THREAD_ID constant + cron schedule

All non-streaming nodes are generated as async def wrapping the synchronous Mellea pipeline via asyncio.to_thread. This prevents blocking LangGraph's ASGI event loop when the pipeline makes synchronous I/O calls (e.g. Ollama, file writes).

Functions:

Attributes:

SUPPORTED_MODALITIES module-attribute
SUPPORTED_MODALITIES = {'synchronous_oneshot', 'streaming', 'conversational_session', 'scheduled', 'event_triggered', 'heartbeat'}
translate_langgraph
translate_langgraph(loaded: 'LoadedContext') -> 'TranslationPlan'
mcp

MCP target: translate a LoadedContext into a TranslationPlan.

Supported modalities

synchronous_oneshot — sync @mcp.tool() wrapping run_pipeline streaming — async @mcp.tool() collecting async generator chunks conversational_session — sync @mcp.tool() (session state managed by pipeline) scheduled — sync @mcp.tool() (schedule is handled externally) event_triggered — sync @mcp.tool() with payload passthrough heartbeat — sync @mcp.tool() (heartbeat state managed by pipeline)

Functions:

translate_mcp
translate_mcp(loaded: 'LoadedContext') -> 'TranslationPlan'

guardian

Runtime governance hooks for Mellea pipelines.

Modules:

  • audit_trail

    Audit trail hook — records every pipeline event to a JSONL log.

  • guardian_hook

    Granite Guardian audit hook for Mellea pipelines.

Functions:

Attributes:

LOGGER module-attribute

LOGGER = configure_logger()

deregister_plugins

deregister_plugins(*plugins) -> None

Remove Guardian and Audit plugins from the global registry.

register_plugins

register_plugins(manifest: PolicyManifest, log_dir: Path = None, enforce: bool = False, guardian_model: Optional[str] = None, inference_engine: InferenceEngineType = OLLAMA) -> tuple

audit_trail

Audit trail hook — records every pipeline event to a JSONL log.

Captures generation calls, Guardian verdicts, component lifecycle events, and validation outcomes. Designed to compose with GuardianAuditPlugin (priority 40) — this hook runs at priority 100 so Guardian verdicts are available in user_metadata by the time we log.

Usage

from audit_trail_hook import AuditTrailPlugin plugin = AuditTrailPlugin(log_path="audit.jsonl") register(plugin)

Classes:

Attributes:

log module-attribute
log = configure_logger('mellea_skills_compiler.audit')
AuditTrailPlugin
AuditTrailPlugin(log_path: str | Path = 'audit_trail.jsonl', policy_id: str = '', guardian_ref: Any = None)

Bases: Plugin

Append-only JSONL audit trail for GraniteClawHub PoC.

Each entry contains
  • timestamp, hook_type, session_id, request_id
  • event-specific data (prompt preview, output preview, verdicts, etc.)
  • optional policy_id for traceability to organisational policy

The priority (100) ensures this runs after all transform/sequential hooks, so user_metadata is fully populated (e.g. Guardian verdicts at pri 40).

Methods:

Attributes:

log_path instance-attribute
log_path = Path(log_path)
policy_id instance-attribute
policy_id = policy_id
log_component_error async
log_component_error(payload: Any, ctx: Any) -> None
log_component_start async
log_component_start(payload: Any, ctx: Any) -> None
log_component_success async
log_component_success(payload: Any, ctx: Any) -> None
log_post_call async
log_post_call(payload: Any, ctx: Any) -> None

Log LLM output and any Guardian verdicts.

log_pre_call async
log_pre_call(payload: Any, ctx: Any) -> None

Log the prompt being sent to the LLM.

log_tool_post async
log_tool_post(payload: Any, ctx: Any) -> None

Log tool result after execution.

log_tool_pre async
log_tool_pre(payload: Any, ctx: Any) -> None

Log tool call before execution.

log_validation async
log_validation(payload: Any, ctx: Any) -> None
summary
summary() -> dict

Return a summary of the audit trail for display.

guardian_hook

Granite Guardian audit hook for Mellea pipelines.

Intercepts generation_post_call, sends the LLM output to Granite Guardian and records the safety verdict in user_metadata for downstream audit hooks.

Two modes
  • AUDIT (default): observe-only, verdicts are logged but generation proceeds.
  • ENFORCE: SEQUENTIAL mode, returns block() when any risk is flagged, raising PluginViolationError to halt the pipeline.

Usage (audit mode — observe only): from guardian_hook import GuardianAuditPlugin plugin = GuardianAuditPlugin(risks=["harm", "jailbreak"]) register(plugin)

Usage (enforce mode — blocks on risk): from guardian_hook import GuardianAuditPlugin plugin = GuardianAuditPlugin(risks=["harm", "jailbreak"], enforce=True) register(plugin)

Usage (Nexus-driven — risks from policy manifest): from nexus_policy import generate_policy_manifest manifest = generate_policy_manifest("An AI agent that...") plugin = GuardianAuditPlugin.from_manifest(manifest, enforce=True) register(plugin)

Classes:

Attributes:

log module-attribute
log = configure_logger('MelleaSkills.guardian_hook')
GuardianAuditPlugin
GuardianAuditPlugin(risks: Optional[list[str]] = None, risk_labels: Optional[list[str]] = None, guardian_model: Optional[str] = None, inference_engine: Optional[InferenceEngineType] = None)

Bases: _GuardianBase, Plugin

Observe-only Guardian hook (AUDIT mode).

Scans every LLM output against Granite Guardian risk checks. Verdicts are logged and stored but generation is never blocked.

For enforcement mode, use GuardianAuditPlugin.from_manifest(manifest, enforce=True) which returns a GuardianEnforcePlugin instead.

Methods:

  • check_input

    Pre-generation: assess input prompt for risks (observe-only).

  • check_output

    Post-generation: assess LLM output for risks (observe-only).

  • check_tool_input

    Pre-tool: log the tool call about to be executed (observe-only).

  • check_tool_output

    Post-tool: scan tool output for risks (observe-only).

  • from_manifest

    Create a plugin from a Nexus PolicyManifest.

Attributes:

all_verdicts instance-attribute
all_verdicts: list[GuardianVerdict] = []
guardian_model instance-attribute
guardian_model = guardian_model
inference_engine instance-attribute
inference_engine = inference_engine
risk_labels instance-attribute
risk_labels = risk_labels or risks
risks instance-attribute
risks = risks or ['harm', 'social_bias', 'jailbreak']
check_input async
check_input(payload: Any, ctx: Any) -> None

Pre-generation: assess input prompt for risks (observe-only).

check_output async
check_output(payload: Any, ctx: Any) -> None

Post-generation: assess LLM output for risks (observe-only).

check_tool_input async
check_tool_input(payload: Any, ctx: Any) -> None

Pre-tool: log the tool call about to be executed (observe-only).

For Pattern 3 (LLM-directed tool calls via ModelOption.TOOLS). Pattern 2 tool calls don't go through Mellea hooks — they use code-level governance instead.

check_tool_output async
check_tool_output(payload: Any, ctx: Any) -> None

Post-tool: scan tool output for risks (observe-only).

Sends the tool output through Guardian risk checks to detect harmful, biased, or sensitive content returned by external tools.

from_manifest classmethod
from_manifest(manifest: Any, enforce: bool = False, guardian_model: Optional[str] = None, inference_engine: InferenceEngineType = OLLAMA) -> GuardianAuditPlugin | GuardianEnforcePlugin

Create a plugin from a Nexus PolicyManifest.

Parameters:

  • manifest (Any) –

    A PolicyManifest with guardian_risks and risk_names.

  • enforce (bool, default: False ) –

    If True, returns a GuardianEnforcePlugin (SEQUENTIAL mode) that blocks generation when risks are detected.

  • guardian_model (Optional[str], default: None ) –

    The guardian model

  • inference_engine (InferenceEngineType, default: OLLAMA ) –

    The inference engine, defaults to Ollama

GuardianEnforcePlugin
GuardianEnforcePlugin(risks: Optional[list[str]] = None, risk_labels: Optional[list[str]] = None, guardian_model: Optional[str] = None, inference_engine: Optional[InferenceEngineType] = None)

Bases: _GuardianBase, Plugin

Enforcement Guardian hook (SEQUENTIAL mode).

Scans every LLM output against Granite Guardian risk checks. If any risk is flagged, returns block() to halt the pipeline with a PluginViolationError.

Methods:

Attributes:

all_verdicts instance-attribute
all_verdicts: list[GuardianVerdict] = []
guardian_model instance-attribute
guardian_model = guardian_model
inference_engine instance-attribute
inference_engine = inference_engine
risk_labels instance-attribute
risk_labels = risk_labels or risks
risks instance-attribute
risks = risks or ['harm', 'social_bias', 'jailbreak']
enforce_input async
enforce_input(payload: Any, ctx: Any) -> Any

Pre-generation: block if input prompt has risks.

enforce_output async
enforce_output(payload: Any, ctx: Any) -> Any

Post-generation: block if LLM output has risks.

enforce_tool_input async
enforce_tool_input(payload: Any, ctx: Any) -> Any

Pre-tool: log the tool call (enforcement reserved for post-invoke).

enforce_tool_output async
enforce_tool_output(payload: Any, ctx: Any) -> Any

Post-tool: block if tool output contains risks.

GuardianVerdict dataclass
GuardianVerdict(risk: str, label: str, raw_output: str = '', timestamp: str = (lambda: isoformat())())

Result of a single Guardian risk check.

Attributes:

label instance-attribute
label: str
raw_output class-attribute instance-attribute
raw_output: str = ''
risk instance-attribute
risk: str
timestamp class-attribute instance-attribute
timestamp: str = field(default_factory=lambda: isoformat())

inference

Classes:

Attributes:

INFERENCE_ENGINE_CACHE module-attribute

INFERENCE_ENGINE_CACHE = {}

OLLAMA_API_URL module-attribute

OLLAMA_API_URL = get('OLLAMA_API_URL', 'http://localhost:11434')

InferenceService

InferenceService(inference_engine_type: Optional[InferenceEngineType] = None)

Methods:

Attributes:

credentials property
credentials: Dict[str, Any]
guardian_model property
guardian_model: str
inference_engine_class property
inference_engine_class
inference_engine_type instance-attribute
inference_engine_type = inference_engine_type if inference_engine_type else OLLAMA
risk_model property
risk_model: str
guardian
guardian(model: Optional[str] = None, parameters: Optional[Dict[str, Any]] = None, **kwargs)
risk
risk(model: Optional[str] = None, parameters: Optional[Dict[str, Any]] = None, **kwargs)

models

Classes:

Attributes:

LOGGER module-attribute

LOGGER = configure_logger()

ComplianceSummary dataclass

ComplianceSummary(classifications: list[RequirementClassification])

Attributes:

automated property
classifications instance-attribute
classifications: list[RequirementClassification]
counts property
counts: dict[str, int]
manual property
partial property

GovernanceAction dataclass

GovernanceAction(id: str, name: str, description: str, source: str, category: str = '', via_risk: str = '', categorized_as: str | list[str] = '')

A governance action/mitigation from NIST, Credo UCF, or other taxonomy.

Attributes:

categorized_as class-attribute instance-attribute
categorized_as: str | list[str] = ''
category class-attribute instance-attribute
category: str = ''
description instance-attribute
description: str
id instance-attribute
id: str
name instance-attribute
name: str
source instance-attribute
source: str
via_risk class-attribute instance-attribute
via_risk: str = ''

NexusRisk dataclass

NexusRisk(name: str, description: str, guardian_prompt: str, is_native: bool = False, taxonomy: str = IBM_GRANITE_GUARDIAN)

A risk identified by AI Atlas Nexus for this use case.

Risks fall into two tiers based on their Nexus tag field: - Native (is_native=True): Guardian has a calibrated assessment path for this risk. guardian_prompt is the bare tag (e.g. "harm", "jailbreak", "social_bias"). - Custom (is_native=False): No built-in Guardian dimension. guardian_prompt is the description text, sent as custom criteria.

Attributes:

description instance-attribute
description: str
guardian_prompt instance-attribute
guardian_prompt: str
is_native class-attribute instance-attribute
is_native: bool = False
name instance-attribute
name: str
taxonomy class-attribute instance-attribute

PolicyManifest dataclass

PolicyManifest(use_case: str, taxonomy: str | list[str], risks: list[NexusRisk], additional_risks: list[NexusRisk], governance_actions: list[GovernanceAction] = list(), governance_taxonomies_used: list[str] = list(), governance_risks_identified: list[str] = list(), generated_at: str = (lambda: isoformat())(), model_used: str = str())

Policy manifest linking a use case to Guardian checks + governance guidance.

Methods:

Attributes:

additional_risks instance-attribute
additional_risks: list[NexusRisk]
generated_at class-attribute instance-attribute
generated_at: str = field(default_factory=lambda: isoformat())
governance_actions class-attribute instance-attribute
governance_actions: list[GovernanceAction] = field(default_factory=list)
governance_risks_identified class-attribute instance-attribute
governance_risks_identified: list[str] = field(default_factory=list)
governance_taxonomies_used class-attribute instance-attribute
governance_taxonomies_used: list[str] = field(default_factory=list)
guardian_risks property
guardian_risks: list[str]

List of Guardian system prompts for each identified risk.

model_used class-attribute instance-attribute
model_used: str = field(default_factory=str)
risk_names property
risk_names: list[str]

List of risk names for logging/display.

risks instance-attribute
risks: list[NexusRisk]
taxonomy instance-attribute
taxonomy: str | list[str]
use_case instance-attribute
use_case: str
from_json classmethod
from_json(path: str) -> PolicyManifest

Load a PolicyManifest from a JSON file produced by to_json().

to_dict
to_dict() -> dict
to_json
to_json(path: str | None = None) -> str

RequirementClassification dataclass

RequirementClassification(action: GovernanceAction, coverage: CoverageLevel, matched_controls: list[str] = list())

Attributes:

action instance-attribute
coverage instance-attribute
coverage: CoverageLevel
matched_controls class-attribute instance-attribute
matched_controls: list[str] = field(default_factory=list)

toolkit

Modules:

file_utils

Functions:

Attributes:

LOGGER module-attribute
LOGGER = configure_logger()
console module-attribute
console = Console()
load_fixtures
load_fixtures(pipeline_dir: Path) -> List[Dict]

Load fixtures from the skill's fixtures/ directory.

Supports two conventions: - ALL_FIXTURES: list of factory functions returning (inputs, id, description) - FIXTURES: list of dicts with 'id' and 'context' keys

load_skill_pipeline
load_skill_pipeline(pipeline_dir: Path)
parse_spec_file
parse_spec_file(spec_path: Path) -> dict

Parse a SKILL.md file into frontmatter dict + markdown body.

logging

Classes:

Functions:

Attributes:

COLORS module-attribute
COLORS = {'DEBUG': '\x1b[36m', 'INFO': '\x1b[32m', 'WARNING': '\x1b[33m', 'ERROR': '\x1b[31m', 'RESET': '\x1b[0m'}
ColoredFormatter

Bases: Formatter

Methods:

format
format(record)
configure_logger
configure_logger(name: str = 'MelleaSkills')