Guides

Extend with plugins and observability

Add guardrails and telemetry without forking the executor: lifecycle-hook plugins for PII masking, content filtering, and cost caps; OpenTelemetry traces and metrics; a zero-infrastructure audit log; and optional Latitude prompt management and tracing.

Route map

1. 1

   Load the built-in guardrails

   PIIMaskPlugin, ContentFilterPlugin, and CostCapPlugin through a PluginRegistry.

2. 2

   Write a custom plugin

   Subclass BasePlugin, override the hooks you need, and (optionally) publish it as an entry point.

3. 3

   Turn on OpenTelemetry

   Install the [otel] extra and point ObservabilityConfig at your collector.

4. 4

   Audit and trace

   Attach an AuditLogger, query it from the CLI, and optionally add Latitude tracing.

Lifecycle hooks, not inheritance

A plugin is a small object the executor calls at fixed points in every invocation's lifecycle. You never subclass or patch AgentExecutor itself — you implement hooks, and the executor dispatches to every loaded plugin in load order. The contract, defined by BasePlugin, has three required members and three optional hooks:

• Required: a metadata property returning PluginMetadata(name=..., version=...), plus setup() (called once on load) and teardown() (called once on unload).
• on_agent_run(agent_name, payload) — called before each invocation. Return a dict to replace the payload sent to the model, or None to pass it through unchanged.
• on_agent_result(agent_name, chunks, latency_ms) — called after a successful invocation. Return a list to replace the output chunks, or None to leave them alone.
• on_error(agent_name, error) — called when an invocation raises. Observational only: the exception always propagates regardless of what the hook returns.

Because plugins run sequentially in load order and each sees the previous plugin's output, transformations compose — load your PII mask before your logger and the logger only ever sees masked payloads. A hook that raises is logged and skipped; it cannot break the invocation.

Built-in guardrails

Three guardrail plugins ship with the package. Load them into a PluginRegistry and hand the registry to the executor:

from replicate_mcp.agents.execution import AgentExecutor
from replicate_mcp.plugins import (
    ContentFilterPlugin, CostCapPlugin, PIIMaskPlugin, PluginRegistry,
)

registry = PluginRegistry()
registry.load(PIIMaskPlugin())                          # no constructor params
registry.load(ContentFilterPlugin(deny_list=["acme-internal"]))
registry.load(CostCapPlugin(per_invocation_cap=1.0,     # defaults shown
                            session_cap=10.0))

executor = AgentExecutor(plugin_registry=registry)
print(registry.names)   # ['pii_mask', 'content_filter', 'cost_cap']

• PIIMaskPlugin — replaces SSN, email, phone, and credit-card patterns with placeholder tokens in payloads and results. No configuration.
• ContentFilterPlugin(deny_list=None) — case-insensitive keyword filter; on a match the prompt (or output text) is replaced with [CONTENT BLOCKED]. The deny list is optional and empty by default — it filters nothing until you supply terms.
• CostCapPlugin(per_invocation_cap=1.0, session_cap=10.0) — blocks invocations whose estimated_cost_usd exceeds the per-call cap or would push cumulative session spend (readable via .session_spend) past the session cap.

Plugins distributed as packages do not need explicit load() calls. The loader discovers anything registered under the replicate_mcp.plugins entry-point group — this is the exact TOML a third-party plugin puts in its own pyproject.toml:

[project.entry-points."replicate_mcp.plugins"]
my_plugin = "my_package.plugin:MyPlugin"

Then load_plugins() (from replicate_mcp.plugins) instantiates every discovered class, skips anything in skip_names, and returns instances ready for registry.load_many(...). Plugins that fail to instantiate are logged and skipped, never fatal.

Write a custom plugin

A complete, working example: stamp every payload with deployment metadata so downstream logs and traces can be filtered by environment. Only on_agent_run is overridden; the other hooks keep their pass-through defaults.

from typing import Any

from replicate_mcp.agents.execution import AgentExecutor
from replicate_mcp.plugins import BasePlugin, PluginMetadata, PluginRegistry, PIIMaskPlugin


class StampPlugin(BasePlugin):
    """Tag every payload with an environment marker."""

    @property
    def metadata(self) -> PluginMetadata:
        return PluginMetadata(
            name="stamp",
            version="1.0.0",
            description="Stamps payloads with deployment metadata.",
        )

    def setup(self) -> None:
        self._stamped = 0

    def teardown(self) -> None:
        print(f"stamp: {self._stamped} payload(s) stamped")

    def on_agent_run(
        self, agent_name: str, payload: dict[str, Any]
    ) -> dict[str, Any] | None:
        self._stamped += 1
        return {**payload, "deployment": "prod-eu", "via_agent": agent_name}


registry = PluginRegistry()
registry.load(PIIMaskPlugin())   # mask first ...
registry.load(StampPlugin())     # ... then stamp the masked payload

executor = AgentExecutor(plugin_registry=registry)

Loading order is the dispatch order, so put guardrails that scrub data before plugins that record it. Loading two plugins with the same metadata.name raises PluginError — unload the first one before replacing it.

OpenTelemetry observability

Telemetry is an optional extra so the core install stays light:

pip install "replicate-mcp-agents[otel]"

Successfully installed opentelemetry-sdk-1.x opentelemetry-exporter-otlp-proto-grpc-1.x ...

Configure the façade once at startup and pass it to the executor. Every field is optional; otlp_endpoint=None falls back to the OTEL_EXPORTER_OTLP_ENDPOINT environment variable, then to http://localhost:4317:

from replicate_mcp.agents.execution import AgentExecutor
from replicate_mcp.observability import Observability, ObservabilityConfig

obs = Observability(ObservabilityConfig(
    service_name="my-app",                        # default: "replicate-mcp-agents"
    otlp_endpoint="http://otel-collector:4317",   # or env var / localhost:4317
    console_fallback=True,                        # print locally if OTLP fails
))
obs.setup()   # once at startup; idempotent

executor = AgentExecutor(observability=obs)

Five instruments are emitted, all under the replicate_mcp. prefix:

Trace spans (e.g. agent.run) carry agent.id, model.id, latency_ms, cost_usd, success, and circuit.state attributes, with secrets redacted. If opentelemetry-sdk is not installed, every Observability method silently no-ops (spans become null objects) — you can leave the wiring in place on machines without the extra and nothing breaks.

Audit logging

The audit log answers "what did my agents do and what did it cost?" with zero infrastructure: an append-only JSONL file at ~/.replicate/audit.jsonl. Each record carries ts, agent, model, latency_ms, cost_usd, success, input_hash (SHA-256 of the payload), and session_id.

from replicate_mcp.agents.execution import AgentExecutor
from replicate_mcp.utils.audit import AuditLogger

executor = AgentExecutor(audit_logger=AuditLogger())  # default path: ~/.replicate/audit.jsonl

Once agents have run, query the log from the CLI:

replicate-agent audit tail --n 5
replicate-agent audit costs --period today

Audit Log — last 5 records
2026-06-12 14:02:11  llama_chat  meta/meta-llama-3-8b-instruct  812 ms  $0.0021  ✓  3f9c2a1b
...
Log path: ~/.replicate/audit.jsonl  (4,096 bytes)

Cost Dashboard — Today
meta/meta-llama-3-8b-instruct    5   100.0%   $0.0105
TOTAL                            5   100.0%   $0.0105

audit tail takes --agent to filter by agent name; audit costs and audit stats (latency percentiles) take --period today|week|month|all; audit clear deletes the file after a confirmation prompt.

Latitude integration (optional)

This entire section is optional — skip it unless you use Latitude for prompt management. Install the extra and set two environment variables:

pip install "replicate-mcp-agents[latitude]"
export LATITUDE_API_KEY=lat_xxxxxxxxxxxxxxxx
export LATITUDE_PROJECT_SLUG=my-project

Successfully installed httpx-0.27.x ...

LatitudeConfig() reads both variables automatically (LATITUDE_PROJECT_ID is the legacy numeric alternative to the slug). The client covers prompt management and multi-turn conversations; the plugin gives you zero-config tracing of every executor run:

from replicate_mcp.latitude import LatitudeClient, LatitudeConfig, LatitudePlugin
from replicate_mcp.plugins import PluginRegistry

config = LatitudeConfig()  # from env vars

# Direct client: fetch, run, and continue prompts
async with LatitudeClient(config) as client:
    prompt = await client.get_prompt("agents/system-prompt", version_uuid="live")
    result = await client.run_prompt("agents/system-prompt",
                                     parameters={"name": "World"})
    reply = await client.chat(result["uuid"], messages=[
        {"role": "user", "content": [{"type": "text", "text": "And in French?"}]},
    ])

# Zero-config tracing: every executor run becomes a Latitude trace
registry = PluginRegistry()
registry.load(LatitudePlugin(config))

Degradation is graceful by design: with no API key the plugin's hooks no-op, trace submission failures are logged but never break execution, and a 402 (trial ended) disables tracing for the rest of the process instead of spamming warnings.

Verification

Common mistakes

Next steps

• Python API reference — full signatures for BasePlugin, PluginRegistry, and Observability.
• CLI reference — every replicate-agent audit option and output format.
• Troubleshooting — what to check when telemetry goes quiet.