# ContextForge Plugin Framework

> **Note:** The plugin framework is now provided by the [`cpex`](https://github.com/contextforge-org/contextforge-plugins-framework) package (ContextForge Plugin Extensions). All framework imports use the `cpex` namespace (e.g., `from cpex.framework import Plugin`).

ContextForge Plugin Framework provides a powerful, production-ready system for AI safety middleware, content security, policy enforcement, and operational excellence. Plugins run as middleware components that can intercept and transform requests and responses at various points in the gateway lifecycle.

## Quick Start

### Enable Plugins

1. Set environment variables in `.env`:
```bash
PLUGINS_ENABLED=true
PLUGINS_CONFIG_FILE=plugins/config.yaml
PLUGINS_CLI_COMPLETION=false
PLUGINS_CLI_MARKUP_MODE=rich
```

2. Configure plugins in `plugins/config.yaml` (see [Configuration](#configuration) section)

3. Restart the gateway: `make dev`

## Plugin Architecture

The framework supports two types of plugins:

### 1. Self-Contained Plugins
- Written in Python and run directly in the gateway process
- Sub-millisecond latency (<1ms)
- Perfect for high-frequency operations like PII filtering and regex transformations
- Examples: `pii_filter`, `regex_filter`, `deny_filter`, `resource_filter`

### 2. External Service Plugins
- Call external AI safety services via HTTP/MCP
- Support microservice integrations with authentication
- 10-100ms latency depending on service
- Examples: LlamaGuard, OpenAI Moderation, custom safety services

## Available Hooks

Plugins can implement hooks at these lifecycle points:

### HTTP Authentication & Middleware Hooks

| Hook | Description | Payload Type | Use Cases |
|------|-------------|--------------|-----------|
| `http_pre_request` | Before any authentication (middleware) | `HttpPreRequestPayload` | Header transformation (X-API-Key → Bearer), correlation IDs |
| `http_auth_resolve_user` | Custom user authentication (auth layer) | `HttpAuthResolveUserPayload` | LDAP, mTLS, token auth, external auth services |
| `http_auth_check_permission` | Custom permission checking (RBAC layer) | `HttpAuthCheckPermissionPayload` | Bypass RBAC, time-based access, IP restrictions |
| `http_post_request` | After request completion (middleware) | `HttpPostRequestPayload` | Audit logging, metrics, response headers |

**See**: [HTTP Authentication Hooks Guide](../docs/docs/using/plugins/http-auth-hooks.md) for detailed examples and flow diagrams.

### MCP Protocol Hooks

| Hook | Description | Payload Type | Use Cases |
|------|-------------|--------------|-----------|
| `prompt_pre_fetch` | Before prompt template retrieval | `PromptPrehookPayload` | Input validation, access control |
| `prompt_post_fetch` | After prompt template retrieval | `PromptPosthookPayload` | Content filtering, transformation |
| `tool_pre_invoke` | Before tool execution | `ToolPreInvokePayload` | Parameter validation, safety checks |
| `tool_post_invoke` | After tool execution | `ToolPostInvokePayload` | Result filtering, audit logging |
| `resource_pre_fetch` | Before resource retrieval | `ResourcePreFetchPayload` | Protocol/domain validation |
| `resource_post_fetch` | After resource retrieval | `ResourcePostFetchPayload` | Content scanning, size limits |
| `agent_pre_invoke` | Before agent invocation | `AgentPreInvokePayload` | Message filtering, access control |
| `agent_post_invoke` | After agent response | `AgentPostInvokePayload` | Response filtering, audit logging |

### Future Hooks (Planned)

- `server_pre_register` / `server_post_register` - Virtual server verification
- `federation_pre_sync` / `federation_post_sync` - Gateway federation

## Configuration

### Main Configuration File (`plugins/config.yaml`)

```yaml
plugins:
  - name: "PIIFilterPlugin"
    kind: "plugins.pii_filter.pii_filter.PIIFilterPlugin"
    description: "Detects and masks Personally Identifiable Information"
    version: "0.1.0"
    author: "Your Name"
    hooks: ["prompt_pre_fetch", "tool_pre_invoke"]
    tags: ["security", "pii", "compliance"]
    mode: "sequential"  # sequential | transform | disabled
    priority: 50     # Lower number = higher priority (runs first)
    conditions:
      - prompts: []     # Empty = apply to all prompts
        server_ids: []  # Apply to specific servers
        tenant_ids: []  # Apply to specific tenants
    config:
      detect_ssn: true
      detect_email: true
      default_mask_strategy: "partial"

# Global settings
plugin_settings:
  parallel_execution_within_band: true
  plugin_timeout: 30
  fail_on_plugin_error: false
  plugin_health_check_interval: 60
```

### Plugin Modes

- **`sequential`**: Blocks violations and prevents request processing (use `on_error: ignore` to block violations but allow errors to pass)
- **`transform`**: Logs violations but allows request to continue
- **`disabled`**: Plugin is not executed (useful for temporary disabling)

### Plugin Priority

Lower priority numbers run first (higher priority). Recommended ranges:
- **1-50**: Critical security plugins (PII, access control)
- **51-100**: Content filtering and validation
- **101-200**: Transformations and enhancements
- **201+**: Logging and monitoring

## Built-in Plugins

### PII Filter Plugin
Detects and masks Personally Identifiable Information (PII):

```yaml
- name: "PIIFilterPlugin"
  kind: "plugins.pii_filter.pii_filter.PIIFilterPlugin"
  config:
    detect_ssn: true
    detect_credit_card: true
    detect_email: true
    detect_phone: true
    detect_aws_keys: true
    default_mask_strategy: "partial"  # redact | partial | hash | tokenize
    block_on_detection: false
    whitelist_patterns:
      - "test@example.com"
```

### Regex Filter Plugin
Find and replace text patterns:

```yaml
- name: "ReplaceBadWordsPlugin"
  kind: "plugins.regex_filter.search_replace.SearchReplacePlugin"
  config:
    words:
      - search: "inappropriate_word"
        replace: "[FILTERED]"
```

### Deny List Plugin
Block requests containing specific terms:

```yaml
- name: "DenyListPlugin"
  kind: "plugins.deny_filter.deny.DenyListPlugin"
  config:
    words:
      - "blocked_term"
      - "another_blocked_term"
```

### Resource Filter Plugin
Validate and filter resource requests:

```yaml
- name: "ResourceFilterExample"
  kind: "plugins.resource_filter.resource_filter.ResourceFilterPlugin"
  config:
    max_content_size: 1048576  # 1MB
    allowed_protocols: ["http", "https"]
    blocked_domains: ["malicious.example.com"]
    content_filters:
      - pattern: "password\\s*[:=]\\s*\\S+"
        replacement: "password: [REDACTED]"
```

## Writing Custom Plugins

### Understanding the Plugin Base Class

The `Plugin` class is an abstract base class (ABC) that provides the foundation for all plugins. You **must** subclass it and implement at least one hook method to create a functional plugin.

```python
from abc import ABC
from cpex.framework import Plugin

class MyPlugin(Plugin):
    """Your plugin must inherit from Plugin."""
    # Implement hook methods (see patterns below)
```

### Three Hook Registration Patterns

The plugin framework supports three flexible patterns for registering hook methods:

#### Pattern 1: Convention-Based (Recommended for Standard Hooks)

The simplest approach - just name your method to match the hook type:

```python
from cpex.framework import (
    Plugin,
    PluginContext,
    ToolPreInvokePayload,
    ToolPreInvokeResult,
)

class MyPlugin(Plugin):
    """Convention-based hook - method name matches hook type."""

    async def tool_pre_invoke(
        self,
        payload: ToolPreInvokePayload,
        context: PluginContext
    ) -> ToolPreInvokeResult:
        """This hook is automatically discovered by its name."""

        # Your logic here
        modified_args = {**payload.args, "processed": True}

        modified_payload = ToolPreInvokePayload(
            name=payload.name,
            args=modified_args,
            headers=payload.headers
        )

        return ToolPreInvokeResult(
            modified_payload=modified_payload,
            metadata={"processed_by": self.name}
        )
```

**When to use:** Default choice for implementing standard framework hooks.

#### Pattern 2: Decorator-Based (Custom Method Names)

Use the `@hook` decorator to register a hook with a custom method name:

```python
from cpex.framework import Plugin, PluginContext
from cpex.framework.decorator import hook
from cpex.framework import (
    ToolHookType,
    ToolPostInvokePayload,
    ToolPostInvokeResult,
)

class MyPlugin(Plugin):
    """Decorator-based hook with custom method name."""

    @hook(ToolHookType.TOOL_POST_INVOKE)
    async def my_custom_handler_name(
        self,
        payload: ToolPostInvokePayload,
        context: PluginContext
    ) -> ToolPostInvokeResult:
        """Method name doesn't match hook type, but @hook decorator registers it."""

        # Your logic here
        return ToolPostInvokeResult(continue_processing=True)
```

**When to use:** When you want descriptive method names that better match your plugin's purpose.

#### Pattern 3: Custom Hooks (Advanced)

Register completely new hook types with custom payload and result types:

```python
from cpex.framework import Plugin, PluginContext, PluginPayload, PluginResult
from cpex.framework.decorator import hook

# Define custom payload type
class EmailPayload(PluginPayload):
    recipient: str
    subject: str
    body: str

# Define custom result type
class EmailResult(PluginResult[EmailPayload]):
    pass

class MyPlugin(Plugin):
    """Custom hook with new hook type."""

    @hook("email_pre_send", EmailPayload, EmailResult)
    async def validate_email(
        self,
        payload: EmailPayload,
        context: PluginContext
    ) -> EmailResult:
        """Completely new hook type: 'email_pre_send'"""

        # Validate email address
        if "@" not in payload.recipient:
            # Fix invalid email
            modified_payload = EmailPayload(
                recipient=f"{payload.recipient}@example.com",
                subject=payload.subject,
                body=payload.body
            )
            return EmailResult(
                modified_payload=modified_payload,
                metadata={"fixed_email": True}
            )

        return EmailResult(continue_processing=True)
```

**When to use:** When extending the framework with domain-specific hook points not covered by standard hooks.

### Hook Method Signature Requirements

All hook methods must follow these rules:

1. **Must be async**: All hooks are asynchronous
2. **Three parameters**: `self`, `payload`, `context`
3. **Type hints required** (for validation): Payload and result types must be properly typed
4. **Return appropriate result type**: Each hook returns a `PluginResult` typed with the hook's payload type

```python
async def hook_name(
    self,
    payload: PayloadType,           # Specific to the hook (e.g., ToolPreInvokePayload)
    context: PluginContext          # Always PluginContext
) -> PluginResult[PayloadType]:     # PluginResult generic, parameterized by the payload type
    """Hook implementation."""
    pass
```

**Understanding Result Types:**

Each hook has a corresponding result type that is actually a type alias for `PluginResult[PayloadType]`:

```python
# These are type aliases defined in the framework
ToolPreInvokeResult = PluginResult[ToolPreInvokePayload]
ToolPostInvokeResult = PluginResult[ToolPostInvokePayload]
PromptPrehookResult = PluginResult[PromptPrehookPayload]
# ... and so on for each hook type
```

This means when you return a result, you're returning a `PluginResult` instance that knows about the specific payload type:

```python
# All of these are valid ways to construct results:
return ToolPreInvokeResult(continue_processing=True)
return ToolPreInvokeResult(modified_payload=new_payload)
return ToolPreInvokeResult(
    modified_payload=new_payload,
    metadata={"processed": True}
)
```

### Complete Plugin Example

Here's a complete plugin showing all patterns:

```python
# plugins/my_plugin/my_plugin.py
from cpex.framework import (
    Plugin,
    PluginContext,
    PluginPayload,
    PluginResult,
    ToolPreInvokePayload,
    ToolPreInvokeResult,
    ToolPostInvokePayload,
    ToolPostInvokeResult,
    ToolHookType,
)
from cpex.framework.decorator import hook

class MyPlugin(Plugin):
    """Example plugin demonstrating all three patterns."""

    # Pattern 1: Convention-based
    async def tool_pre_invoke(
        self,
        payload: ToolPreInvokePayload,
        context: PluginContext
    ) -> ToolPreInvokeResult:
        """Pre-process tool invocation - found by naming convention."""

        # Access plugin configuration
        threshold = self.config.config.get("threshold", 0.5)

        # Modify payload
        modified_args = {**payload.args, "plugin_processed": True}
        modified_payload = ToolPreInvokePayload(
            name=payload.name,
            args=modified_args,
            headers=payload.headers
        )

        return ToolPreInvokeResult(
            modified_payload=modified_payload,
            metadata={"threshold": threshold}
        )

    # Pattern 2: Decorator with custom name
    @hook(ToolHookType.TOOL_POST_INVOKE)
    async def process_tool_result(
        self,
        payload: ToolPostInvokePayload,
        context: PluginContext
    ) -> ToolPostInvokeResult:
        """Post-process tool result - found via decorator."""

        # Transform result
        if isinstance(payload.result, dict):
            modified_result = {
                **payload.result,
                "processed_by": self.name
            }
            modified_payload = ToolPostInvokePayload(
                name=payload.name,
                result=modified_result
            )
            return ToolPostInvokeResult(modified_payload=modified_payload)

        return ToolPostInvokeResult(continue_processing=True)
```

### Plugin Structure

Create a new directory under `plugins/`:

```
plugins/my_plugin/
├── __init__.py
├── plugin-manifest.yaml
├── my_plugin.py
└── README.md
```

### Plugin Manifest (`plugin-manifest.yaml`)

```yaml
description: "My custom plugin"
author: "Your Name"
version: "1.0.0"
available_hooks:
  - "tool_pre_invoke"
  - "tool_post_invoke"
default_configs:
  threshold: 0.8
  enable_logging: true
```

### Register Your Plugin

Add to `plugins/config.yaml`:

```yaml
plugins:
  - name: "MyCustomPlugin"
    kind: "plugins.my_plugin.my_plugin.MyPlugin"
    description: "My custom plugin description"
    version: "1.0.0"
    author: "Your Name"
    hooks: ["tool_pre_invoke", "tool_post_invoke"]
    mode: "sequential"
    priority: 100
    config:
      threshold: 0.8
      enable_logging: true
```

## Plugin Development Best Practices

### Hook Results and Control Flow

Each hook returns a result object that controls execution flow:

```python
# Allow processing to continue
return ToolPreInvokeResult(continue_processing=True)

# Modify the payload
return ToolPreInvokeResult(
    modified_payload=modified_payload,
    metadata={"processed": True}
)

# Block execution with a violation
from cpex.framework import PluginViolation

return ToolPreInvokeResult(
    continue_processing=False,
    violation=PluginViolation(
        code="POLICY_VIOLATION",
        reason="Request blocked by security policy",
        description="Detected prohibited content"
    )
)
```

### Error Handling

Errors inside a plugin should be raised as exceptions. The plugin manager will catch the error, and its behavior depends on both the gateway's and plugin's configuration as follows:

1. If `plugin_settings.fail_on_plugin_error` in the plugin `config.yaml` is set to `true`, the exception is bubbled up as a PluginError and the error is passed to the client of ContextForge regardless of the plugin mode.
2. If `plugin_settings.fail_on_plugin_error` is set to false, the error is handled based off of the plugin mode in the plugin's config as follows:
   * If `mode` is `sequential`, both violations and errors are bubbled up as exceptions and the execution is blocked.
   * If `mode` is `sequential` with `on_error: ignore`, violations are bubbled up as exceptions and execution is blocked, but errors are logged and execution continues.
   * If `mode` is `transform`, execution is allowed to proceed whether there are errors or violations. Both are logged.

### Accessing Plugin Context

The `context` parameter provides access to request-scoped and global state:

```python
async def tool_pre_invoke(
    self,
    payload: ToolPreInvokePayload,
    context: PluginContext
) -> ToolPreInvokeResult:
    # Access request ID
    request_id = context.global_context.request_id

    # Access user information
    user = context.global_context.user
    tenant_id = context.global_context.tenant_id

    # Store plugin-specific state (persists across all hooks in the request)
    context.state["invocation_count"] = context.state.get("invocation_count", 0) + 1

    # Add metadata
    context.metadata["processing_time"] = 0.123

    return ToolPreInvokeResult(continue_processing=True)
```

### Logging and Monitoring

```python
def __init__(self, config: PluginConfig):
    super().__init__(config)
    self.logger.info(f"Initialized {self.name} v{self.version}")

async def tool_pre_invoke(self, payload: ToolPreInvokePayload, context: PluginContext) -> ToolPreInvokeResult:
    self.logger.debug(f"Processing tool: {payload.name}")
    # ... plugin logic
    self.metrics.increment("requests_processed")
```

### Configuration Validation

```python
def __init__(self, config: PluginConfig):
    super().__init__(config)
    self._validate_config()

def _validate_config(self) -> None:
    """Validate plugin configuration."""
    required_keys = ["threshold", "api_key"]
    for key in required_keys:
        if key not in self.config.config:
            raise ValueError(f"Missing required config key: {key}")

    threshold = self.config.config.get("threshold")
    if not 0 <= threshold <= 1:
        raise ValueError("threshold must be between 0 and 1")
```

## Performance Considerations

### Latency Guidelines
- **Self-contained plugins**: <1ms target
- **External service plugins**: <100ms target
- Use async/await for I/O operations
- Implement timeouts for external calls

### Resource Management
```python
class MyPlugin(Plugin):
    def __init__(self, config: PluginConfig):
        super().__init__(config)
        self._session = None

    async def initialize(self):
        """Called when plugin is loaded."""
        self._session = aiohttp.ClientSession()

    async def shutdown(self):
        """Called when plugin manager shuts down."""
        if self._session:
            await self._session.close()
```

## Testing Plugins

### Unit Testing

```python
import pytest
from cpex.framework import (
    PluginConfig,
    PluginContext,
    GlobalContext,
    ToolPreInvokePayload,
)
from plugins.my_plugin.my_plugin import MyPlugin

@pytest.fixture
def plugin():
    config = PluginConfig(
        name="test_plugin",
        description="Test",
        version="1.0",
        author="Test",
        kind="plugins.my_plugin.my_plugin.MyPlugin",
        hooks=["tool_pre_invoke"],
        config={"threshold": 0.8}
    )
    return MyPlugin(config)

@pytest.mark.asyncio
async def test_tool_pre_invoke(plugin):
    payload = ToolPreInvokePayload(
        name="test_tool",
        args={"arg1": "value1"}
    )
    context = PluginContext(
        global_context=GlobalContext(request_id="test-123")
    )

    result = await plugin.tool_pre_invoke(payload, context)

    assert result.continue_processing is True
    assert result.modified_payload.args["plugin_processed"] is True
```

### Integration Testing

```bash
# Test with live gateway
make dev
curl -X POST http://localhost:4444/tools/invoke \
  -H "Content-Type: application/json" \
  -d '{"name": "test_tool", "arguments": {}}'
```

## Troubleshooting

### Common Issues

1. **Plugin not loading**: Check `plugin_dirs` in config and Python import paths
2. **Configuration errors**: Validate YAML syntax and required fields
3. **Performance issues**: Profile plugin execution time and optimize bottlenecks
4. **Hook not triggering**: Verify hook name matches available hooks in manifest
5. **Method signature errors**: Ensure hooks have correct parameters (self, payload, context) and are async

### Debug Mode

```bash
LOG_LEVEL=DEBUG make serve # port 4444
# Or with reloading dev server:
LOG_LEVEL=DEBUG make dev # port 8000
```

### Testing Hook Discovery

To verify your hooks are properly registered:

```python
from cpex.framework import PluginManager

manager = PluginManager("path/to/config.yaml")
await manager.initialize()

# Check loaded plugins
for plugin_config in manager.config.plugins:
    print(f"Plugin: {plugin_config.name}")
    print(f"  Hooks: {plugin_config.hooks}")
```

## Documentation Links

- **Plugin Usage Guide**: https://ibm.github.io/mcp-context-forge/using/plugins/
- **Plugin Lifecycle**: https://ibm.github.io/mcp-context-forge/using/plugins/lifecycle/
- **API Reference**: Generated from code docstrings
- **Examples**: See `plugins/` directory for complete implementations
- **Hook Patterns Test**: `tests/unit/mcpgateway/plugins/framework/hooks/test_hook_patterns.py`

## Performance Metrics

The framework supports high-performance operations:
- **1,000+ requests/second** with 5 active plugins
- **Sub-millisecond latency** for self-contained plugins
- **Parallel execution** within priority bands
- **Resource isolation** and timeout protection

## Security Features

- Input validation and sanitization
- Timeout protection for external calls
- Resource limits and quota enforcement
- Error isolation between plugins
- Comprehensive audit logging
- Plugin configuration validation
- Hook signature validation at plugin load time
