Skip to main content

IBM Mellea

A plugin for IBM Mellea pipelines, no pipeline rewrite. It hooks GENERATION_PRE_CALL, GENERATION_POST_CALL, TOOL_PRE_INVOKE and TOOL_POST_INVOKE and sends each surface to Alice. The tool_output hook is the one that catches indirect prompt injection: text returned by tools, files and URLs is evaluated before it re-enters the model. Audit mode records every verdict; enforce mode halts the pipeline on BLOCK and rewrites writable fields on MASK.

Status: Generally available · Evaluates: Prompts, Responses, Tool calls · Vendor: IBM

User prompt, model response, tool call arguments, tool output.

Setup​

pip install 'mellea-skills-compiler[alice]'

Configuration: ALICE_API_KEY, ALICE_APP_ID, ALICE_URL_OVERRIDE (optional), ALICE_API_TIMEOUT (optional)

Example​

usage (docs/alice_hook.md)

import os
from mellea.plugins import PluginViolationError
from mellea_skills_compiler.guardian import (
    register_alice_plugin, deregister_plugins,
)

plugin = register_alice_plugin(
    app_id=os.environ["ALICE_APP_ID"],
    enforce=True,           # SEQUENTIAL hooks; False = AUDIT-only
    fail_closed=False,      # SDK outage must not kill the skill
    model_name="granite3.3:8b",
    provider="ollama",
)
try:
    answer = run_pipeline(query)
except PluginViolationError as exc:
    answer = f"Blocked by policy: {exc.reason}"
finally:
    deregister_plugins(plugin)

register_alice_plugin (guardian/init.py)

def register_alice_plugin(
    app_id: str,
    api_key: Optional[str] = None,
    enforce: bool = False,
    user_id: str = "mellea-skill",
    model_name: Optional[str] = None,
    provider: Optional[str] = None,
    fail_closed: bool = False,
    log_dir: Optional[Path] = None,
) -> Union[AliceAuditPlugin, AliceEnforcePlugin]:
    """Register the Alice (Wonderfence) plugin against the global registry.

    Returns the plugin instance — pass it back to ``deregister_plugins(...)``
    when the skill exits.

    Args:
        app_id: Alice application UUID (provisioned in the Alice console).
        api_key: Wonderfence API key — falls back to ``ALICE_API_KEY`` env var.
        enforce: ``True`` for SEQUENTIAL hooks that can block/mask; ``False``
            for AUDIT-only verdict recording.
        user_id: User identifier threaded into every ``AnalysisContext``.
        model_name: Model name to report in the context.
        provider: Inference provider to report in the context.
        fail_closed: If ``True``, an SDK failure becomes a block; if ``False``
            (the default), failures log and let the pipeline through.
        log_dir: Reserved for future use; verdicts are auto-emitted to
            ``runtime_audit.jsonl`` whenever ``AuditTrailPlugin`` is also
            registered (it consumes ``user_metadata["alice_verdicts"]``).
    """
    plugin_cls = AliceEnforcePlugin if enforce else AliceAuditPlugin
    plugin = plugin_cls(
        app_id=app_id,
        api_key=api_key,
        user_id=user_id,
        model_name=model_name,
        provider=provider,
        fail_closed=fail_closed,
    )
    register(plugin)
    LOGGER.info(
        "Alice plugin registered (app_id=%s, mode=%s, fail_closed=%s)",
        app_id, "enforce" if enforce else "audit", fail_closed,
    )
    return plugin

AliceEnforcePlugin: tool output hook (guardian/alice_hook.py)

    @hook(HookType.TOOL_POST_INVOKE, mode=PluginMode.SEQUENTIAL)
    async def enforce_tool_output(self, payload: Any, ctx: Any) -> Any:
        text = str(getattr(payload, "tool_output", "") or "")
        session_id = getattr(payload, "session_id", None)
        v = await _evaluate_response(self, text, session_id, surface="tool_output")
        _log_verdict(v, prefix="alice-enforce")
        _stash_verdict(self, payload, v)

        fail = _failure_block_or_continue(self, v, code="alice_unreachable")
        if fail is not None:
            return fail

        if v.action == Actions.BLOCK:
            tool_name = getattr(
                getattr(payload, "model_tool_call", None), "name", "unknown",
            )
            return block(
                reason=f"Alice blocked tool output ({tool_name})",
                code="alice_tool_output_blocked",
                details={
                    "tool": tool_name,
                    "detections": v.detections,
                    "correlation_id": v.correlation_id,
                    "surface": "tool_output",
                },
            )
        if v.action == Actions.MASK and v.action_text is not None:
            return modify(payload, tool_output=v.action_text)
        return None

examples/web_summarizer/main.py: env-gated registration

    af_plugin = None
    app_id = os.environ.get("ALICE_APP_ID")
    if app_id:
        from mellea_skills_compiler.guardian import (
            deregister_plugins,
            register_alice_plugin,
        )
        af_plugin = register_alice_plugin(
            app_id=app_id,
            enforce=True,
            fail_closed=False,
            model_name=MODEL_ID,
            provider=BACKEND,
        )

    try:
        print(run_pipeline(question=args.question, target=args.target))
    except PluginViolationError as exc:
        print(f"Blocked by Alice policy: {exc.reason}", file=sys.stderr)
        sys.exit(2)
    except Exception as exc:
        print(f"Error: {exc}", file=sys.stderr)
        sys.exit(1)
    finally:
        if af_plugin is not None:
            deregister_plugins(af_plugin)

Good to know​

MASK is honored where the Mellea hook allows the field to be rewritten (model_tool_call, tool_output). On generation hooks the framework drops the mutation, so hard prompt and response masking belongs at a proxy layer. fail_closed=False by default so an Alice outage never kills the skill.