Plugin Framework

Plugin Framework

!!! success "Production Ready" The plugin framework is production ready with comprehensive hook coverage, robust error handling, and battle-tested implementations. Supports both native and external service plugins.

Overview

The MCP Context Forge Plugin Framework provides a comprehensive, production-grade system for extending gateway functionality through pre/post processing hooks at various points in the MCP request lifecycle. The framework supports both high-performance native plugins and sophisticated external AI service integrations.

Key Capabilities

• AI Safety Middleware - Integration with LlamaGuard, OpenAI Moderation, custom ML models
• Content Security - PII detection and masking, input validation, output sanitization
• Policy Enforcement - Business rules, compliance checking, audit trails
• Performance Protection - Timeout handling, resource limits, graceful degradation
• Operational Excellence - Health‑oriented design, clear errors, sensible defaults
• Enterprise Features - Multi-tenant isolation, conditional execution, sophisticated context management

Enabling Plugins

Environment Configuration

Enable the plugin framework in your .env file:

# Enable plugin framework
PLUGINS_ENABLED=true

# Optional: Custom plugin config path
PLUGIN_CONFIG_FILE=plugins/config.yaml

If deploying the gateway as a container, set these environment variables in your compose file or in the container's run command.

Architecture

!!! details "Plugin Framework Specification" Check the specification docs for a detailed design of the plugin system.

The plugin framework implements a hybrid architecture supporting both native and external service integrations:

Native Plugins

• In-Process Execution: Written in Python, run directly within the gateway process
• High Performance: Sub-millisecond latency, no network overhead
• Direct Access: Full access to gateway internals and context
• Use Cases: PII filtering, regex transformations, input validation, simple business rules
• Examples: PIIFilterPlugin, SearchReplacePlugin, DenyListPlugin

External Service Plugins

• Multiple Transports: External plugins can communicate via MCP (STDIO/HTTP) or high-performance gRPC
• Enterprise AI Support: LlamaGuard, OpenAI Moderation, custom ML models
• Independent Scaling: Services run outside the gateway and can scale separately
• Use Cases: Advanced AI safety, complex ML inference, policy engines (e.g., OPA)
• Examples: OPA external plugin server, LlamaGuard integration, OpenAI Moderation
• Performance: gRPC provides 8x throughput vs MCP/HTTP (4,700 vs ~600 calls/sec)

Gunicorn Workers and External Transports

When running the gateway under Gunicorn with multiple workers:

• STDIO: Each worker spawns its own plugin subprocess and maintains a separate session. This maximizes isolation but multiplies plugin processes and does not share state across workers.
• Streamable HTTP over UDS: Run the plugin server as a separate long‑lived process and point all workers to the same Unix socket. This reduces process count and allows shared plugin state, while avoiding TCP port exposure.

Unified Plugin Interface

Both plugin types implement the same interface, enabling seamless switching between deployment models:

class Plugin:
    async def prompt_pre_fetch(self, payload: PromptPrehookPayload,
                              context: PluginContext) -> PromptPrehookResult: ...
    async def prompt_post_fetch(self, payload: PromptPosthookPayload,
                               context: PluginContext) -> PromptPosthookResult: ...
    async def tool_pre_invoke(self, payload: ToolPreInvokePayload,
                             context: PluginContext) -> ToolPreInvokeResult: ...
    async def tool_post_invoke(self, payload: ToolPostInvokePayload,
                              context: PluginContext) -> ToolPostInvokeResult: ...
    async def resource_pre_fetch(self, payload: ResourcePreFetchPayload,
                                context: PluginContext) -> ResourcePreFetchResult: ...
    async def resource_post_fetch(self, payload: ResourcePostFetchPayload,
                                 context: PluginContext) -> ResourcePostFetchResult: ...
    # ... additional hook methods

Build Your First Plugin (Quickstart)

Decide between a native (in‑process) or external (MCP) plugin:

• Native: simplest path; write Python class extending Plugin, configure via plugins/config.yaml using fully‑qualified class path.
• External: runs as a separate MCP server (STDIO or Streamable HTTP); great for independent scaling and isolation.

!!! details "Plugins CLI" To bootstrap a plugin quickly (native or external), run mcpplugins bootstrap and follow the prompts.

Quick native skeleton:

from mcpgateway.plugins.framework import Plugin, PluginConfig, PluginContext, PromptPrehookPayload, PromptPrehookResult

class MyPlugin(Plugin):
    def __init__(self, config: PluginConfig):
        super().__init__(config)

    async def prompt_pre_fetch(self, payload: PromptPrehookPayload, context: PluginContext) -> PromptPrehookResult:
        # modify
        return PromptPrehookResult(modified_payload=payload)

        # or block
        # return PromptPrehookResult(
        #     continue_processing=False,
        #     violation=PluginViolation(
        #         reason=f"Blocked by {self.name}",
        #         description="...",
        #         code="...",
        #         details="..."
        #     )
        # )

Register it in plugins/config.yaml:

plugins:

  - name: "MyPlugin"
    kind: "plugins.my_plugin.plugin.MyPlugin"
    hooks: ["prompt_pre_fetch"]
    mode: "permissive"
    priority: 120

External plugin quickstart:

!!! details "Plugins Lifecycle Guide" See the plugin lifecycle guide for building, testing, and serving external plugins.

plugins:
  # MCP/HTTP transport (default)
  - name: "MyExternal"
    kind: "external"
    priority: 10
    mcp:
      proto: STREAMABLEHTTP
      url: http://localhost:8000/mcp
      # uds: /var/run/mcp-plugin.sock  # use UDS instead of TCP

  # gRPC transport (high performance - ~8x faster)
  - name: "MyGrpcPlugin"
    kind: "external"
    priority: 10
    grpc:
      target: "localhost:50051"
      # uds: /var/run/grpc-plugin.sock  # use UDS instead of TCP

!!! tip "High-Performance Transports" For performance-critical deployments, consider gRPC transport (4,700 calls/sec) or Unix socket transport (9,000 calls/sec) compared to ~600 calls/sec with MCP/HTTP.

Plugin Configuration

The plugin configuration file is used to configure a set of plugins that implement hook functions used to register to hook points throughout the MCP Context Forge. An example configuration is below. It contains two main sections: plugins and plugin_settings.

!!! details "Plugin Configuration" Check here for detailed explanations of configuration options and fields.

# plugins/config.yaml
plugins:

  - name: "PIIFilterPlugin"                    # Unique plugin identifier
    kind: "plugins.pii_filter.pii_filter.PIIFilterPlugin"  # Plugin class path
    description: "Detects and masks PII"       # Human-readable description
    version: "1.0.0"                          # Plugin version
    author: "Security Team"                   # Plugin author
    hooks:                                    # Hook registration

      - "prompt_pre_fetch"
      - "tool_pre_invoke"
      - "tool_post_invoke"
    tags:                                     # Searchable tags

      - "security"
      - "pii"
      - "compliance"
    mode: "enforce"                           # enforce|enforce_ignore_error|permissive|disabled
    priority: 50                              # Execution priority (lower = higher)
    conditions:                               # Conditional execution

      - server_ids: ["prod-server"]
        tenant_ids: ["enterprise"]
        tools: ["sensitive-tool"]
    config:                                   # Plugin-specific configuration
      detect_ssn: true
      detect_credit_card: true
      mask_strategy: "partial"
      redaction_text: "[REDACTED]"

# Global plugin settings
plugin_settings:
  parallel_execution_within_band: false      # Execute same-priority plugins in parallel
  plugin_timeout: 30                         # Per-plugin timeout (seconds)
  fail_on_plugin_error: false                # Continue on plugin failures
  plugin_health_check_interval: 60           # Health check interval (seconds)

Getting Started (Native Plugins)

Use the native plugins out of the box:

1. Copy and adapt the example config (enable any subset):

# plugins/config.yaml
plugins:

  - name: "PIIFilterPlugin"
    kind: "plugins.pii_filter.pii_filter.PIIFilterPlugin"
    hooks: ["prompt_pre_fetch", "prompt_post_fetch", "tool_pre_invoke", "tool_post_invoke"]
    mode: "permissive"
    priority: 50
    config:
      detect_ssn: true
      detect_email: true
      detect_credit_card: true
      default_mask_strategy: "partial"

  - name: "ReplaceBadWordsPlugin"
    kind: "plugins.regex_filter.search_replace.SearchReplacePlugin"
    hooks: ["prompt_pre_fetch", "prompt_post_fetch", "tool_pre_invoke", "tool_post_invoke"]
    mode: "enforce"
    priority: 150
    config:
      words:

        - { search: "crap", replace: "crud" }
        - { search: "crud", replace: "yikes" }

  - name: "DenyListPlugin"
    kind: "plugins.deny_filter.deny.DenyListPlugin"
    hooks: ["prompt_pre_fetch"]
    mode: "enforce"
    priority: 100
    config:
      words: ["innovative", "groundbreaking", "revolutionary"]

  - name: "ResourceFilterExample"
    kind: "plugins.resource_filter.resource_filter.ResourceFilterPlugin"
    hooks: ["resource_pre_fetch", "resource_post_fetch"]
    mode: "enforce"
    priority: 75
    config:
      max_content_size: 1048576
      allowed_protocols: ["http", "https"]
      blocked_domains: ["malicious.example.com"]
      content_filters:

        - { pattern: "password\\s*[:=]\\s*\\S+", replacement: "password: [REDACTED]" }

plugin_settings:
  parallel_execution_within_band: false
  plugin_timeout: 30
  fail_on_plugin_error: false
  enable_plugin_api: true
  plugin_health_check_interval: 60

2. Ensure .env contains: PLUGINS_ENABLED=true and PLUGIN_CONFIG_FILE=plugins/config.yaml.
3. Start the gateway: make dev (or make serve).

That's it — the gateway now runs the enabled plugins at the selected hook points.

Plugin Configuration

The plugins section lists the set of configured plugins that will be loaded by the Context Forge at startup. Each plugin contains a set of standard configurations, and then a config section designed for plugin specific configurations. The attributes are defined as follows:

Hook Types

Available hook values for the hooks field:

MCP Protocol Hooks:

HTTP Authentication & Middleware Hooks:

See the HTTP Authentication Hooks Guide for detailed implementation examples.

Condition Fields

Users may only want plugins to be invoked on specific servers, tools, and prompts. To address this, a set of conditionals can be applied to a plugin. The attributes in a conditional combine together in as a set of and operations, while each attribute list item is ored with other items in the list.

The conditions array contains objects that specify when plugins should execute:

MCP Configuration Fields

For external plugins (kind: "external"), the mcp object configures the MCP server connection:

Global Plugin Settings

The plugin_settings are as follows:

Execution Modes

Each plugin can operate in one of four modes:

Priority and Execution Order

Plugins execute in priority order (ascending):

# Execution order example
plugins:

  - name: "Authentication"
    priority: 10      # Runs first

  - name: "RateLimiter"
    priority: 50      # Runs second

  - name: "ContentFilter"
    priority: 100     # Runs third

  - name: "Logger"
    priority: 200     # Runs last

Plugins with the same priority may execute in parallel if parallel_execution_within_band is enabled.

Available Hooks

The plugin framework provides comprehensive hook coverage across the entire MCP request lifecycle:

Production Hooks (Implemented)

!!! note "HTTP Authentication & Middleware Hooks" For detailed information on implementing custom authentication and authorization, see the HTTP Authentication Hooks Guide.

!!! note "Agent-to-Agent (A2A) Hooks" Agent hooks enable filtering and monitoring of Agent-to-Agent interactions. These hooks allow you to:

- Filter/transform messages before they reach agents
- Control which tools are available to agents
- Override model or system prompt settings
- Filter agent responses for safety/compliance
- Monitor tool invocations made by agents

See [A2A Documentation](../agents/a2a.md) for more information on Agent-to-Agent features.

Planned Hooks (Roadmap)

Prompt Hooks Details

The prompt hooks allow plugins to intercept and modify prompt retrieval and rendering:

• prompt_pre_fetch: Receives the prompt name and arguments before prompt template retrieval. Can modify the arguments.
• prompt_post_fetch: Receives the completed prompt after rendering. Can modify the prompt text or block it from being returned.

Example Use Cases:

• Detect prompt injection attacks
• Sanitize or anonymize prompts
• Search and replace

Prompt Hook Payloads

PromptPrehookPayload: Payload for prompt pre-fetch hooks.

class PromptPrehookPayload(BaseModel):
    name: str                                    # Prompt template name
    args: Optional[dict[str, str]] = Field(default_factory=dict)  # Template arguments

Example:

payload = PromptPrehookPayload(
    name="user_greeting",
    args={"user_name": "Alice", "time_of_day": "morning"}
)

PromptPosthookPayload: Payload for prompt post-fetch hooks.

class PromptPosthookPayload(BaseModel):
    name: str                                    # Prompt name
    result: PromptResult                         # Rendered prompt result

Tool Hooks Details

The tool hooks enable plugins to intercept and modify tool invocations:

• tool_pre_invoke: Receives the tool name and arguments before execution. Can modify arguments or block the invocation entirely.
• tool_post_invoke: Receives the tool result after execution. Can modify the result or block it from being returned.

Example use cases:

• PII detection and masking in tool inputs/outputs
• Rate limiting specific tools
• Audit logging of tool usage
• Input validation and sanitization
• Output filtering and transformation

Tool Hook Payloads

ToolPreInvokePayload: Payload for tool pre-invoke hooks.

class ToolPreInvokePayload(BaseModel):
    name: str                                    # Tool name
    args: Optional[dict[str, Any]] = Field(default_factory=dict)  # Tool arguments
    headers: Optional[HttpHeaderPayload] = None  # HTTP pass-through headers

ToolPostInvokePayload: Payload for tool post-invoke hooks.

class ToolPostInvokePayload(BaseModel):
    name: str                                    # Tool name
    result: Any                                  # Tool execution result

The associated HttpHeaderPayload object for the ToolPreInvokePayload is as follows:

Special payload for HTTP header manipulation.

class HttpHeaderPayload(RootModel[dict[str, str]]):
    # Provides dictionary-like access to HTTP headers
    # Supports: __iter__, __getitem__, __setitem__, __len__

Usage:

headers = HttpHeaderPayload({"Authorization": "Bearer token", "Content-Type": "application/json"})
headers["X-Custom-Header"] = "custom_value"
auth_header = headers["Authorization"]

Resource Hooks Details

The resource hooks enable plugins to intercept and modify resource fetching:

• resource_pre_fetch: Receives the resource URI and metadata before fetching. Can modify the URI, add metadata, or block the fetch entirely.
• resource_post_fetch: Receives the resource content after fetching. Can modify the content, redact sensitive information, or block it from being returned.

Example use cases:

• Protocol validation (block non-HTTPS resources)
• Domain blocklisting/allowlisting
• Content size limiting
• Sensitive data redaction
• Content transformation and filtering
• Resource caching metadata

Resource Hook Payloads

ResourcePreFetchPayload: Payload for resource pre-fetch hooks.

class ResourcePreFetchPayload(BaseModel):
    uri: str                                     # Resource URI
    metadata: Optional[dict[str, Any]] = Field(default_factory=dict)  # Request metadata

ResourcePostFetchPayload: Payload for resource post-fetch hooks.

class ResourcePostFetchPayload(BaseModel):
    uri: str                                     # Resource URI
    content: Any                                 # Fetched resource content

HTTP Authentication & Middleware Hooks

For HTTP request processing and authentication, see the dedicated guide:

• HTTP Authentication Hooks Guide - Complete guide to http_pre_request, http_auth_resolve_user, http_auth_check_permission, and http_post_request hooks

Agent Hooks Details

The agent hooks allow plugins to intercept and modify Agent-to-Agent (A2A) interactions:

• agent_pre_invoke: Receives agent invocation details before the agent processes the request. Can filter messages, restrict tools, override model settings, or block the request entirely.
• agent_post_invoke: Receives the agent's response after processing. Can filter response content, redact sensitive information, or add audit metadata.

Example Use Cases:

• Filter offensive or sensitive content in messages
• Restrict which tools an agent can access
• Override model selection or system prompts
• Apply content moderation to agent responses
• Log all agent interactions for compliance
• Block agents from accessing certain resources

Agent Hook Payloads

AgentPreInvokePayload: Payload for agent pre-invoke hooks.

class AgentPreInvokePayload(BaseModel):
    agent_id: str                                    # Agent identifier (can be modified for routing)
    messages: List[Message]                          # Conversation messages (can be filtered/transformed)
    tools: Optional[List[str]] = None                # Available tools (can be restricted)
    headers: Optional[HttpHeaderPayload] = None      # HTTP headers
    model: Optional[str] = None                      # Model override
    system_prompt: Optional[str] = None              # System instructions override
    parameters: Optional[Dict[str, Any]] = None      # LLM parameters (temperature, max_tokens, etc.)

AgentPostInvokePayload: Payload for agent post-invoke hooks.

class AgentPostInvokePayload(BaseModel):
    agent_id: str                                    # Agent identifier
    messages: List[Message]                          # Response messages (can be filtered/transformed)
    tool_calls: Optional[List[Dict[str, Any]]] = None  # Tool invocations made by agent

Example Plugin:

from mcpgateway.plugins.framework import (
    Plugin,
    PluginContext,
    AgentPreInvokePayload,
    AgentPreInvokeResult,
    PluginViolation,
)

class AgentSafetyPlugin(Plugin):
    """Filter agent interactions for safety."""

    async def agent_pre_invoke(
        self,
        payload: AgentPreInvokePayload,
        context: PluginContext
    ) -> AgentPreInvokeResult:
        # Restrict dangerous tools
        if payload.tools:
            safe_tools = [t for t in payload.tools if t not in ["file_delete", "system_exec"]]
            if len(safe_tools) < len(payload.tools):
                payload.tools = safe_tools
                self.logger.info(f"Restricted tools for agent {payload.agent_id}")

        # Filter offensive content in messages
        for msg in payload.messages:
            if self._contains_offensive_content(msg.content):
                return AgentPreInvokeResult(
                    continue_processing=False,
                    violation=PluginViolation(
                        code="OFFENSIVE_CONTENT",
                        reason="Message contains offensive content",
                        description="Agent request blocked due to policy violation"
                    )
                )

        return AgentPreInvokeResult(
            modified_payload=payload,
            metadata={"safety_checked": True}
        )

Planned Hooks (Roadmap)

• server_pre_register / server_post_register - Server validation
• federation_pre_sync / federation_post_sync - Gateway federation

Writing 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.

from abc import ABC
from mcpgateway.plugins.framework import Plugin, PluginConfig

class MyPlugin(Plugin):
    """Your plugin must inherit from Plugin."""

    def __init__(self, config: PluginConfig):
        super().__init__(config)
        # Initialize plugin-specific configuration
        self.my_setting = config.config.get("my_setting", "default")

!!! important "Key Design Principle" Plugins implement only the hooks they need using one of three registration patterns. You don't need to implement all hooks - just the ones relevant to your plugin's purpose.

Three Hook Registration Patterns

Pattern 1: Convention-Based (Recommended)

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

from mcpgateway.plugins.framework import (
    Plugin,
    PluginContext,
    ToolPreInvokePayload,
    ToolPreInvokeResult,
)

class ContentFilterPlugin(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."""

        # Block dangerous operations
        if payload.name == "file_delete" and "system" in str(payload.args):
            from mcpgateway.plugins.framework import PluginViolation
            return ToolPreInvokeResult(
                continue_processing=False,
                violation=PluginViolation(
                    code="DANGEROUS_OP",
                    reason="Dangerous operation blocked",
                    description=f"Cannot delete system files"
                )
            )

        # Modify arguments
        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:

from mcpgateway.plugins.framework import Plugin, PluginContext
from mcpgateway.plugins.framework.decorator import hook
from mcpgateway.plugins.framework import (
    ToolHookType,
    ToolPostInvokePayload,
    ToolPostInvokeResult,
)

class AuditPlugin(Plugin):
    """Decorator-based hook with descriptive method name."""

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

        # Log tool execution
        self.logger.info(f"Tool executed: {payload.name}")

        # Filter sensitive data from results
        if isinstance(payload.result, dict) and "password" in payload.result:
            filtered_result = {**payload.result, "password": "[REDACTED]"}
            modified_payload = ToolPostInvokePayload(
                name=payload.name,
                result=filtered_result
            )
            return ToolPostInvokeResult(modified_payload=modified_payload)

        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:

from mcpgateway.plugins.framework import (
    Plugin,
    PluginContext,
    PluginPayload,
    PluginResult
)
from mcpgateway.plugins.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 EmailPlugin(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:
            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: Payload and result types must be properly typed for validation
4. Return appropriate result type: Each hook returns a PluginResult typed with the hook's payload type

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

Understanding Result Types:

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

# These are type aliases defined in the framework
ToolPreInvokeResult = PluginResult[ToolPreInvokePayload]
ToolPostInvokeResult = PluginResult[ToolPostInvokePayload]
PromptPrehookResult = PluginResult[PromptPrehookPayload]
HttpAuthResolveUserResult = PluginResult[dict]  # Special case for user dict
# ... and so on for each hook type

This means when you return a result, you're returning a PluginResult instance:

# 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}
)

Plugin Lifecycle Methods

Plugins can implement optional lifecycle methods:

class MyPlugin(Plugin):
    async def initialize(self):
        """Called when plugin is loaded."""
        # Set up resources, connections, etc.
        self._session = aiohttp.ClientSession()

    async def shutdown(self):
        """Called when plugin manager shuts down."""
        # Cleanup resources
        if hasattr(self, '_session'):
            await self._session.close()

Plugin Context and State

Each hook function has a context object of type PluginContext which is designed to allow plugins to pass state between one another across all hook types in a request, or for a plugin to pass state information to itself across different hooks. The plugin context looks as follows:

class GlobalContext(BaseModel):
    """The global context, which shared across all plugins.

    Attributes:
            request_id (str): ID of the HTTP request.
            user (str): user ID associated with the request.
            tenant_id (str): tenant ID.
            server_id (str): server ID.
            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.
    """

    request_id: str
    user: Optional[str] = None
    tenant_id: Optional[str] = None
    server_id: Optional[str] = None
    state: dict[str, Any] = Field(default_factory=dict)
    metadata: dict[str, Any] = Field(default_factory=dict)


class PluginContext(BaseModel):
    """The plugin's context, which lasts a request lifecycle.

    Attributes:
       state:  the inmemory state of the request.
       global_context: the context that is shared across plugins.
       metadata: plugin meta data.
    """

    state: dict[str, Any] = Field(default_factory=dict)
    global_context: GlobalContext
    metadata: dict[str, Any] = Field(default_factory=dict)

As can be seen, the PluginContext has both a state dictionary and a global_context object that also has a state dictionary. A single plugin can share state across all hooks in a request by using the PluginContext state dictionary. It can share state with other plugins using the context.global_context.state dictionary. Metadata for the specific hook site is passed in through the metadata dictionaries in the context.global_context.metadata. It is meant to be read-only. The context.metadata is plugin specific metadata and can be used to store metadata information such as timing information.

The following shows how plugins can maintain state across different hooks:

async def prompt_pre_fetch(self, payload, context):
    # Store state for later use
    context.set_state("request_time", time.time())
    context.set_state("original_args", payload.args.copy())

    return PromptPrehookResult()

async def prompt_post_fetch(self, payload, context):
    # Retrieve state from pre-hook
    elapsed = time.time() - context.get_state("request_time", 0)
    original = context.get_state("original_args", {})

    # Add timing metadata
    context.metadata["processing_time_ms"] = elapsed * 1000

    return PromptPosthookResult()

Tool and Gateway Metadata

Tool hooks have access to tool and gateway metadata through the global context metadata dictionary. They are accessible as follows:

It can be accessed inside of the tool hooks through:

from mcpgateway.plugins.framework.constants import GATEWAY_METADATA, TOOL_METADATA

tool_meta = context.global_context.metadata[TOOL_METADATA]
assert tool_meta.original_name == "test_tool"
assert tool_meta.url.host == "example.com"
assert tool_meta.integration_type == "REST" or tool_meta.integration_type == "MCP"

Note, if the integration type is MCP the gateway information may also be available as follows.

gateway_meta = context.global_context.metadata[GATEWAY_METADATA]
assert g

…(truncated)