ADR-009: Sequential Plugin Processing for Responses

Note: This ADR uses current terminology (process_response, modified_content, PluginResult). See ADR-022 for the unified result type.

Context

When implementing response modification capabilities (see ADR-008), we needed to determine how multiple plugins should interact when processing responses. The key question was whether plugins should process responses in parallel with conflict resolution, or sequentially with cumulative modifications.

This decision affects:

Use Case Driving the Decision

The tool allowlist plugin filters tools/list responses, potentially followed by other plugins that might:

Decision

We will implement sequential plugin processing where plugins process responses one after another, with each plugin receiving the output of the previous plugin as input.

async def process_response(self, response: MCPResponse) -> MCPResponse:
    """Process response through plugins sequentially."""
    current_response = response
    
    for plugin in self.response_plugins:
        decision = await plugin.process_response(current_response)
        
        if not decision.allowed:
            raise PluginBlockedError(decision.reason)
        
        if decision.modified_content is not None:
            current_response = decision.modified_content
            # Next plugin receives this modified response
    
    return current_response

Processing Order

Plugin processing order is determined by priority values (0-100, with 50 as default). All middleware and security plugins are sorted together in a unified pipeline:

plugins:
  middleware:
    _global:
      - handler: "tool_manager"       # Processes first (priority: 10)
        config:
          enabled: true
          priority: 10

  security:
    _global:
      - handler: "basic_pii_filter"   # Processes second (priority: 20)
        config:
          enabled: true
          priority: 20

Note: Auditing plugins do NOT participate in sequential response processing. They observe the final result via log_response() after the processing pipeline completes, and they execute in definition order (not by priority).

Alternatives Considered

Alternative 1: Parallel Processing with Conflict Resolution

async def process_response(self, response: MCPResponse) -> MCPResponse:
    """Process response through all plugins in parallel."""
    decisions = await asyncio.gather(*[
        plugin.process_response(response) 
        for plugin in self.response_plugins
    ])
    
    # Resolve conflicts between different modifications
    return resolve_response_conflicts(response, decisions)

Rejected because:

Alternative 3: Plugin Priority System

plugins:
  security:
    _global:
      - handler: "tool_allowlist"
        config:
          enabled: true
          priority: 10  # Higher priority (lower number)

      - handler: "content_filter"
        config:
          enabled: true
          priority: 50  # Default priority

Rejected because:

Note: This alternative was later reconsidered and adopted in a subsequent decision.

Alternative 4: Plugin Dependency Declaration

class ContentFilterPlugin:
    depends_on = ["tool_allowlist"]  # Must run after allowlist

Rejected because:

Note: This alternative was later reconsidered and rejected in favor of a simple priority system.

Alternative 5: Response Accumulation with Original

async def process_response(self, response: MCPResponse) -> MCPResponse:
    """Each plugin sees original response, accumulate changes."""
    modifications = []
    
    for plugin in self.response_plugins:
        decision = await plugin.process_response(response)  # Always original
        if decision.modified_content:
            modifications.append(decision.modified_content)
    
    return merge_modifications(response, modifications)

Rejected because:

Consequences

Positive

Negative

Mitigation Strategies

  1. Clear Documentation: Explain priority system and its implications
  2. Plugin Guidelines: Best practices for plugin design and priority assignment
  3. Configuration Validation: Warn about potentially problematic plugin priorities
  4. Audit Logging: Log each plugin's decision for transparency

Implementation Details

Plugin Manager Sequential Processing

The actual implementation returns a ProcessingPipeline object that provides full visibility into each processing stage:

class PluginManager:
    async def process_response(
        self, request: MCPRequest, response: MCPResponse, server_name: Optional[str] = None
    ) -> ProcessingPipeline:
        """Process response through all enabled plugins sequentially."""
        # Creates a ProcessingPipeline with PipelineStage for each plugin
        # Middleware and security plugins are sorted together by priority
        # Auditing plugins are NOT included - they observe via log_response() after

Plugin Ordering Strategy

def _get_processing_pipeline(self, server_name: str) -> List[PluginInterface]:
    """Get plugins in processing order: sorted by priority (0-100, lower = higher priority)."""
    all_plugins = []

    # Collect middleware plugins
    all_plugins.extend(self._resolve_plugins_for_upstream(self.middleware_plugins, server_name))

    # Collect security plugins
    all_plugins.extend(self._resolve_plugins_for_upstream(self.security_plugins, server_name))

    # Sort by priority (ascending - lower numbers first)
    all_plugins.sort(key=lambda p: getattr(p, "priority", 50))
    return all_plugins

Key points:

This sequential processing approach provides predictable, debuggable plugin behavior while enabling powerful plugin composition for response modification use cases.