Merge pull request #4005 from IBM/3925-bug-pluginconditioncontent_types-field-is-defined-but-not-implemented

Fixes #3925: Implemented the `content_types` field functionality for `PluginCondition`

Author: ja8zyjits

.secrets.baseline

@@ -3,7 +3,7 @@
     "files": "^.secrets.baseline$",
     "lines": null
   },
-  "generated_at": "2026-04-02T08:08:58Z",
+  "generated_at": "2026-04-07T13:00:53Z",
   "plugins_used": [
     {
       "name": "AWSKeyDetector"
@@ -1462,15 +1462,15 @@
         "hashed_secret": "38297d822c960daea26f148a2fede848dcc2083b",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 619,
+        "line_number": 620,
         "type": "Secret Keyword",
         "verified_result": null
       },
       {
         "hashed_secret": "7373eb3c8f036e7f058bfc66fc27870f01231645",
         "is_secret": false,
         "is_verified": false,
-        "line_number": 1261,
+        "line_number": 1349,
         "type": "Secret Keyword",
         "verified_result": null
       }

docs/docs/architecture/plugins.md

@@ -1,3 +1,4 @@
+
 # Plugin Framework Specification
 
 **Version**: 1.0
@@ -1010,6 +1011,93 @@ class PluginCondition(BaseModel):
     content_types: Optional[list[str]] = None  # Execute for specific content types
 ```
 
+#### Content Type Filtering
+
+Plugins can be configured to execute only for specific content types using the `content_types` condition. This enables fine-grained control over when plugins process requests based on the HTTP `Content-Type` header.
+
+**Configuration Example:**
+
+```yaml
+plugins:
+  - name: "JsonValidator"
+    kind: "plugins.json_validator.JsonValidator"
+    hooks: ["tool_pre_invoke"]
+    conditions:
+      - content_types: ["application/json"]
+```
+
+**Matching Behavior:**
+
+- **Case-insensitive**: `APPLICATION/JSON` matches `application/json`
+- **Parameter stripping**: `application/json; charset=utf-8` matches `application/json`
+- **Multiple types**: Supports OR logic - plugin executes if any content type matches
+- **Permissive default**: If `content_types` is not specified or request has no Content-Type header, plugin executes normally
+
+**Common Content Types:**
+
+| Content Type | Description | Use Case |
+|--------------|-------------|----------|
+| `application/json` | JSON data | API requests, structured data validation |
+| `text/plain` | Plain text | Simple text processing, logging |
+| `text/html` | HTML documents | Web scraping, content extraction |
+| `application/xml` | XML data | Legacy API integration, SOAP services |
+| `multipart/form-data` | File uploads | File validation, virus scanning |
+| `application/x-www-form-urlencoded` | Form submissions | Form data validation |
+
+**Example: JSON-Only Security Plugin**
+
+```yaml
+plugins:
+  - name: "JsonSecurityScanner"
+    kind: "plugins.security.json_scanner.JsonSecurityScanner"
+    hooks: ["tool_pre_invoke", "tool_post_invoke"]
+    mode: "enforce"
+    priority: 10
+    conditions:
+      - content_types: ["application/json"]
+        server_ids: ["production-api"]
+```
+
+**Example: Multi-Format Data Validator**
+
+```yaml
+plugins:
+  - name: "DataValidator"
+    kind: "plugins.validation.data_validator.DataValidator"
+    hooks: ["tool_pre_invoke"]
+    conditions:
+      - content_types:
+          - "application/json"
+          - "application/xml"
+          - "text/csv"
+```
+
+**Combined Conditions:**
+
+Content type filtering works seamlessly with other conditions:
+
+```yaml
+plugins:
+  - name: "TeamJsonProcessor"
+    kind: "plugins.processors.json_processor.JsonProcessor"
+    hooks: ["tool_pre_invoke"]
+    conditions:
+      - content_types: ["application/json"]
+        tenant_ids: ["team-alpha", "team-beta"]
+        tools: ["data_analysis", "report_generation"]
+```
+
+**Security Considerations:**
+
+!!! warning "Content-Type Spoofing"
+    Clients can set arbitrary Content-Type headers. Use `content_types` for filtering and optimization, not as a security boundary. Combine with other conditions (server_ids, tenant_ids) for defense-in-depth.
+
+**Performance Benefits:**
+
+- **Reduced overhead**: Skip expensive processing for irrelevant content types
+- **Targeted validation**: Apply format-specific validators only when needed
+- **Resource optimization**: Prevent unnecessary plugin execution
+
 ## Hook Reference Documentation
 
 The plugin framework provides two main categories of hooks, each documented in detail in separate files:

docs/docs/using/plugins/index.md

@@ -1167,8 +1167,42 @@ conditions:
     server_ids: ["prod-server-1", "prod-server-2"]
     tenant_ids: ["enterprise-tenant"]
     user_patterns: ["admin-*", "support-*"]
+    content_types: ["application/json", "text/plain"]
 ```
 
+**Content Type Filtering:**
+
+Plugins can be configured to execute only for specific HTTP Content-Type headers:
+
+```yaml
+plugins:
+  - name: "JsonValidator"
+    kind: "plugins.validation.json_validator.JsonValidator"
+    hooks: ["tool_pre_invoke"]
+    mode: "enforce"
+    priority: 50
+    conditions:
+      - content_types: ["application/json"]
+    config:
+      strict_schema: true
+
+  - name: "MultiFormatProcessor"
+    kind: "plugins.processors.format_processor.FormatProcessor"
+    hooks: ["tool_pre_invoke", "tool_post_invoke"]
+    conditions:
+      - content_types:
+          - "application/json"
+          - "application/xml"
+          - "text/csv"
+        server_ids: ["data-api"]
+```
+
+**Key behaviors:**
+- Case-insensitive matching (`APPLICATION/JSON` matches `application/json`)
+- Ignores charset parameters (`application/json; charset=utf-8` matches `application/json`)
+- If `content_types` is not specified, plugin executes for all content types
+- If request has no Content-Type header, plugin executes normally (permissive default)
+
 ### 4. Logging and Monitoring
 
 Use appropriate log levels:

mcpgateway/auth.py

@@ -1188,11 +1188,14 @@ def _set_trace_for_user(user_obj: EmailUser, *, teams: Any = _UNSET, auth_method
             # Get plugin contexts from request state if available
             global_context = getattr(request.state, "plugin_global_context", None) if request else None
             if not global_context:
+                # Extract content type from headers
+                content_type = headers.get("content-type") if headers else None
                 # Create global context
                 global_context = GlobalContext(
                     request_id=request_id,
                     server_id=None,
                     tenant_id=None,
+                    content_type=content_type,
                 )
 
             context_table = getattr(request.state, "plugin_context_table", None) if request else None
@@ -1726,11 +1729,14 @@ def _inject_userinfo_instate(request: Optional[object] = None, user: Optional[Em
     # Get plugin contexts from request state if available
     global_context = getattr(request.state, "plugin_global_context", None) if request else None
     if not global_context:
+        # Extract content type from request headers
+        content_type = request.headers.get("content-type") if request and hasattr(request, "headers") else None
         # Create global context
         global_context = GlobalContext(
             request_id=request_id,
             server_id=None,
             tenant_id=None,
+            content_type=content_type,
         )
 
     if user:

mcpgateway/middleware/http_auth_middleware.py

@@ -57,7 +57,8 @@ async def run_pre_request_hooks(
 
     if global_context is None:
         request_id = get_correlation_id() or generate_correlation_id()
-        global_context = GlobalContext(request_id=request_id, server_id=None, tenant_id=None)
+        content_type = headers.get("content-type") if headers else None
+        global_context = GlobalContext(request_id=request_id, server_id=None, tenant_id=None, content_type=content_type)
 
     try:
         pre_result, context_table = await plugin_manager.invoke_hook(

mcpgateway/middleware/rbac.py

@@ -15,7 +15,7 @@
 import functools
 from functools import wraps
 import logging
-from typing import Callable, Generator, List, Optional
+from typing import Any, Callable, Generator, List, Optional
 import uuid
 import warnings
 
@@ -598,7 +598,7 @@ def decorator(func: Callable) -> Callable:
         """
 
         @wraps(func)
-        async def wrapper(*args, **kwargs):
+        async def wrapper(*args, **kwargs: dict[str, Any]):
             """Async wrapper function that performs permission check before calling original function.
 
             Args:
@@ -659,10 +659,13 @@ async def wrapper(*args, **kwargs):
                     global_context = plugin_global_context
                 else:
                     request_id = user_context.get("request_id") or uuid.uuid4().hex
+                    request: Optional[Request] = kwargs.get("request")
+                    content_type = request.headers.get("content-type") if request and hasattr(request, "headers") else None
                     global_context = GlobalContext(
                         request_id=request_id,
                         server_id=None,
                         tenant_id=None,
+                        content_type=content_type,
                     )
 
                 # Invoke permission check hook, passing plugin contexts from HTTP_PRE_REQUEST hook

mcpgateway/plugins/framework/models.py

@@ -14,7 +14,7 @@
 import logging
 import os
 from pathlib import Path
-from typing import Any, Generic, Optional, Self, TypeVar, Union
+from typing import Any, Generic, Literal, Optional, Self, TypeVar, Union
 
 # Third-Party
 from pydantic import BaseModel, ConfigDict, Field, field_serializer, field_validator, model_validator, PrivateAttr, ValidationInfo
@@ -222,12 +222,27 @@ def serialize_set(self, value: set[str] | None) -> list[str] | None:
             The set as a serializable list.
         """
         if value:
-            values = []
+            values: list[Any] = []
             for key in value:
                 values.append(key)
             return values
         return None
 
+    @field_validator("content_types")
+    @classmethod
+    def normalize_content_types(cls, value: str | None) -> list[str] | Literal[""] | None:
+        """Pre-normalize content types during initialization.
+
+        Args:
+            value: str of content type.
+
+        Returns:
+            Normalized content type.
+        """
+        if value:
+            return [each.split(sep=";")[0].strip().lower() for each in value]
+        return value
+
 
 class AppliedTo(BaseModel):
     """What tools/prompts/resources and fields the plugin will be applied to.
@@ -1382,6 +1397,7 @@ class GlobalContext(BaseModel):
             user (str): user ID associated with the request.
             tenant_id (str): tenant ID.
             server_id (str): server ID.
+            content_type (Optional[str]): Content-Type header from the request.
             metadata (Optional[dict[str,Any]]): a global shared metadata across plugins (Read-only from plugin's perspective).
             state (Optional[dict[str,Any]]): a global shared state across plugins.
 
@@ -1401,15 +1417,44 @@ class GlobalContext(BaseModel):
         '123'
         >>> c.server_id
         'srv1'
+        >>> ctx3 = GlobalContext(request_id="req-789", content_type="application/json")
+        >>> ctx3.content_type
+        'application/json'
+        >>> ctx4 = GlobalContext(request_id="req-999", content_type="application/json; charset=utf-8")
+        >>> ctx4.content_type
+        'application/json; charset=utf-8'
     """
 
     request_id: str
     user: Optional[Union[str, dict[str, Any]]] = None
     tenant_id: Optional[str] = None
     server_id: Optional[str] = None
+    content_type: Optional[str] = None
     state: dict[str, Any] = Field(default_factory=dict)
     metadata: dict[str, Any] = Field(default_factory=dict)
 
+    @field_validator("content_type")
+    @classmethod
+    def validate_content_type(cls, value: str | None) -> str | None:
+        """Pre Validate content types during initialization.
+
+        Args:
+            value: str of content type.
+
+        Raises:
+            ValueError: if name is length > 200 or not a valid character.
+
+        Returns:
+            validated content type.
+        """
+        if value is None:
+            return value
+        if len(value) > 200:
+            raise ValueError("Content-Type header too long")
+        if not value.isprintable():
+            raise ValueError("Content-Type contains invalid characters")
+        return value
+
 
 class PluginContext(BaseModel):
     """The plugin's context, which lasts a request lifecycle.

mcpgateway/plugins/framework/utils.py

@@ -184,6 +184,34 @@ def parse_class_name(name: str) -> tuple[str, str]:
     return ("", name)
 
 
+def normalize_content_type(content_type: str) -> str:
+    """Extract base content type without parameters.
+
+    Args:
+        content_type: Raw content type string (e.g., 'application/json; charset=utf-8')
+
+    Returns:
+        Normalized content type (e.g., 'application/json')
+
+    Examples:
+        >>> normalize_content_type('application/json; charset=utf-8')
+        'application/json'
+        >>> normalize_content_type('text/html')
+        'text/html'
+        >>> normalize_content_type('TEXT/PLAIN')
+        'text/plain'
+        >>> normalize_content_type('application/json;charset=utf-8')
+        'application/json'
+        >>> normalize_content_type('')
+        ''
+        >>> normalize_content_type('   ')
+        ''
+    """
+    if not isinstance(content_type, str):
+        return ""
+    return content_type.split(";")[0].strip().lower()
+
+
 def matches(condition: PluginCondition, context: GlobalContext) -> bool:
     """Check if conditions match the current context.
 
@@ -207,6 +235,16 @@ def matches(condition: PluginCondition, context: GlobalContext) -> bool:
         >>> ctx3 = GlobalContext(request_id="req3", user="admin_user")
         >>> matches(cond2, ctx3)
         True
+        >>> cond3 = PluginCondition(content_types=["application/json"])
+        >>> ctx4 = GlobalContext(request_id="req4", content_type="application/json")
+        >>> matches(cond3, ctx4)
+        True
+        >>> ctx5 = GlobalContext(request_id="req5", content_type="application/json; charset=utf-8")
+        >>> matches(cond3, ctx5)
+        True
+        >>> ctx6 = GlobalContext(request_id="req6", content_type="text/plain")
+        >>> matches(cond3, ctx6)
+        False
     """
     # Check server ID
     if condition.server_ids and context.server_id not in condition.server_ids:
@@ -220,6 +258,13 @@ def matches(condition: PluginCondition, context: GlobalContext) -> bool:
     if condition.user_patterns and context.user:
         if not any(pattern in context.user for pattern in condition.user_patterns):
             return False
+
+    # Check content types
+    if condition.content_types and context.content_type:
+        normalized_request = normalize_content_type(context.content_type)
+        if normalized_request not in condition.content_types:
+            return False
+
     return True