Plugin System Overview

Argus MCP includes a composable plugin framework that intercepts MCP requests at well-defined hook points. Plugins run inside the middleware chain -- after recovery and before telemetry -- so every request passes through enabled plugins in priority order.

How Plugins Work

Each plugin extends PluginBase and implements one or more async hooks. The PluginManager runs hooks in priority order (lower number = runs first), enforces per-plugin timeouts, and applies optional conditions.

PluginManager.run_hook("tool_pre_invoke", context)

  ┌────────────────────────────────────────────────┐
  │  For each enabled plugin (sorted by priority)  │
  │    1. Evaluate conditions (method, backend)    │
  │    2. Copy context (copy-on-write)             │
  │    3. Run hook with asyncio.timeout            │
  │    4. Apply execution_mode on error            │
  └────────────────────────────────────────────────┘

Hook Points

Plugins receive context at eight points during request processing:

Execution Modes

Each plugin has an execution_mode that controls what happens on failure:

Conditions

Plugins can be scoped to specific methods or backends:

plugins:
  entries:
    - name: rate_limiter
      conditions:
        methods: ["call_tool"]           # Only tool calls
        backends: ["expensive-backend"]  # Only this backend

Configuration

Plugins are configured under the plugins top-level key:

plugins:
  enabled: true
  entries:
    - name: secrets_detection
      enabled: true
      execution_mode: enforce
      priority: 10
      timeout: 5.0
      settings:
        block: true

    - name: rate_limiter
      enabled: true
      priority: 50
      settings:
        max_requests: 100
        window: 60

Plugin Config Fields

Top-Level Plugin Config

Middleware Chain Position

The PluginMiddleware sits in the middleware chain between Recovery and Telemetry:

Auth → Recovery → PluginMiddleware → Telemetry → Audit → Routing

This means plugins run after crash protection (Recovery catches exceptions from plugins too) and before observability (Telemetry records timing including plugin overhead).

External Plugins

Beyond the built-in plugins, Argus MCP provides seven integration plugins for external policy engines, scanners, and content moderation services.

opa_policy

Evaluates tool requests against an Open Policy Agent instance. The plugin sends the request context to the OPA decision endpoint and blocks non-compliant calls.

- name: opa_policy
  execution_mode: enforce
  priority: 5
  settings:
    opa_url: "http://localhost:8181"

cedar_policy

Evaluates requests against an Amazon Cedar policy engine instance.

- name: cedar_policy
  execution_mode: enforce
  priority: 5
  settings:
    cedar_url: "http://localhost:8180"

clamav

Scans request/response payloads for malware using ClamAV.

- name: clamav
  execution_mode: enforce
  priority: 8
  settings:
    host: "127.0.0.1:3310"

virustotal

Checks content hashes against the VirusTotal database. The API key should be stored in the encrypted secret store and referenced via the VT_API_KEY environment variable.

- name: virustotal
  execution_mode: enforce
  priority: 8
  settings:
    threshold: 3

llmguard

Sends prompts to an LLM Guard instance for injection and toxicity detection.

- name: llmguard
  execution_mode: enforce
  priority: 12
  settings:
    api_url: "http://localhost:8800"
    threshold: 0.5

content_moderation

Routes content through a cloud moderation service. Supports multiple provider backends.

- name: content_moderation
  execution_mode: enforce
  priority: 12
  settings:
    provider: "openai"

unified_pdp

Combines multiple policy decision engines (OPA, Cedar, custom) into a single evaluation with configurable combination logic.

- name: unified_pdp
  execution_mode: enforce
  priority: 5
  settings:
    engines:
      - type: opa
        url: "http://localhost:8181"
      - type: cedar
        url: "http://localhost:8180"
    combination_mode: "all"

Writing a Custom Plugin

Create a Python file that extends PluginBase and implement the hooks you need. Place it somewhere importable or register it through the plugin discovery mechanism.

Minimal Example

from argus_mcp.plugins.base import PluginBase


class MyPlugin(PluginBase):
    """Example plugin that logs every tool call."""

    name = "my_logger"

    async def tool_pre_invoke(self, context: dict) -> dict:
        tool_name = context.get("tool_name", "unknown")
        self.logger.info("Tool called: %s", tool_name)
        return context

    async def tool_post_invoke(self, context: dict) -> dict:
        self.logger.info("Tool completed: %s", context.get("tool_name"))
        return context

Plugin Lifecycle

1. on_load -- Called once when the plugin manager initializes the plugin. Use for setup (open connections, load config).
2. Hook calls -- tool_pre_invoke, tool_post_invoke, etc. are called for each matching request. Context is copy-on-write: modifications to the dict are isolated to your plugin unless you return the modified context.
3. on_unload -- Called during shutdown. Close connections and flush buffers here.

Configuration Passthrough

Whatever you put under settings: in config.yaml is available as self.settings inside the plugin:

plugins:
  entries:
    - name: my_logger
      enabled: true
      priority: 50
      settings:
        log_level: "DEBUG"
        include_args: true

async def on_load(self, context: dict) -> dict:
    level = self.settings.get("log_level", "INFO")
    self.logger.setLevel(level)
    return context

Custom Plugin Conditions

Use conditions to restrict when the plugin runs:

- name: my_logger
  conditions:
    methods: ["call_tool"]
    backends: ["debug-server"]

The plugin will only execute for call_tool requests routed to debug-server.