Add examples

Author: peteski22

examples/auth_plugin/README.md

@@ -0,0 +1,33 @@
+# Auth Plugin Example
+
+A plugin that validates Bearer token authentication for incoming HTTP requests.
+
+## What it does
+
+- Checks for the `Authorization` header in requests
+- Validates that it contains a Bearer token
+- Rejects requests with invalid or missing tokens (returns 401 Unauthorized)
+- Allows valid requests to continue
+
+## Running the example
+
+```bash
+# Set the expected token (optional, defaults to "secret-token-123")
+export AUTH_TOKEN="my-secret-token"
+
+# From the repository root
+cd examples/auth_plugin
+uv run python main.py
+```
+
+## Configuration
+
+The plugin uses the `AUTH_TOKEN` environment variable to set the expected token (default: `secret-token-123`).
+
+## Key concepts demonstrated
+
+- Request validation and rejection
+- Environment variable configuration
+- Returning error responses with `continue_=False`
+- Setting HTTP status codes and response bodies
+- Using class initialization for plugin state

examples/auth_plugin/__init__.py

@@ -0,0 +1 @@
+"""Authentication plugin example."""

examples/auth_plugin/main.py

@@ -0,0 +1,80 @@
+"""Authentication plugin that validates Bearer tokens.
+
+This example demonstrates how to reject requests that don't meet security requirements.
+"""
+
+import asyncio
+import logging
+import os
+
+from google.protobuf.empty_pb2 import Empty
+from grpc import ServicerContext
+
+from mcpd_plugins import BasePlugin, serve
+from mcpd_plugins.v1.plugins.plugin_pb2 import (
+    FLOW_REQUEST,
+    Capabilities,
+    HTTPRequest,
+    HTTPResponse,
+    Metadata,
+)
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class AuthPlugin(BasePlugin):
+    """Plugin that validates Bearer token authentication."""
+
+    def __init__(self):
+        """Initialize the auth plugin with expected token."""
+        super().__init__()
+        self.expected_token = os.getenv("AUTH_TOKEN", "secret-token-123")
+
+    async def GetMetadata(self, request: Empty, context: ServicerContext) -> Metadata:
+        """Return plugin metadata."""
+        return Metadata(
+            name="auth-plugin",
+            version="1.0.0",
+            description="Validates Bearer token authentication",
+        )
+
+    async def GetCapabilities(self, request: Empty, context: ServicerContext) -> Capabilities:
+        """Declare support for request flow."""
+        return Capabilities(flows=[FLOW_REQUEST])
+
+    async def HandleRequest(self, request: HTTPRequest, context: ServicerContext) -> HTTPResponse:
+        """Validate Bearer token in Authorization header."""
+        logger.info(f"Authenticating request: {request.method} {request.url}")
+
+        # Check for Authorization header.
+        auth_header = request.headers.get("Authorization", "")
+
+        if not auth_header.startswith("Bearer "):
+            logger.warning("Missing or invalid Authorization header")
+            return self._unauthorized_response("Missing or invalid Authorization header")
+
+        # Extract and validate token.
+        token = auth_header[7:]  # Remove "Bearer " prefix
+        if token != self.expected_token:
+            logger.warning("Invalid token")
+            return self._unauthorized_response("Invalid token")
+
+        # Token is valid, allow request to continue.
+        logger.info("Authentication successful")
+        return HTTPResponse(**{"continue": True})
+
+    def _unauthorized_response(self, message: str) -> HTTPResponse:
+        """Create a 401 Unauthorized response."""
+        response = HTTPResponse(
+            status_code=401,
+            body=f'{{"error": "{message}"}}'.encode(),
+            **{"continue": False},
+        )
+        response.headers["Content-Type"] = "application/json"
+        response.headers["WWW-Authenticate"] = "Bearer"
+        return response
+
+
+if __name__ == "__main__":
+    asyncio.run(serve(AuthPlugin()))

examples/logging_plugin/README.md

@@ -0,0 +1,26 @@
+# Logging Plugin Example
+
+A plugin that logs HTTP request and response details for observability and debugging.
+
+## What it does
+
+- Logs details of incoming HTTP requests (method, URL, headers, body size)
+- Logs details of outgoing HTTP responses (status code, headers, body size)
+- Redacts sensitive headers (Authorization, Cookie) for security
+- Supports both REQUEST and RESPONSE flows
+
+## Running the example
+
+```bash
+# From the repository root
+cd examples/logging_plugin
+uv run python main.py
+```
+
+## Key concepts demonstrated
+
+- Implementing both `HandleRequest()` and `HandleResponse()` methods
+- Declaring multiple flows in `GetCapabilities()`
+- Structured logging for observability
+- Handling sensitive data (header redaction)
+- Pass-through behavior while gathering telemetry

examples/logging_plugin/__init__.py

@@ -0,0 +1 @@
+"""Logging plugin example."""

examples/logging_plugin/main.py

@@ -0,0 +1,93 @@
+"""Logging plugin that logs HTTP requests and responses.
+
+This example demonstrates implementing both request and response flows
+for observability purposes.
+"""
+
+import asyncio
+import logging
+
+from google.protobuf.empty_pb2 import Empty
+from grpc import ServicerContext
+
+from mcpd_plugins import BasePlugin, serve
+from mcpd_plugins.v1.plugins.plugin_pb2 import (
+    FLOW_REQUEST,
+    FLOW_RESPONSE,
+    Capabilities,
+    HTTPRequest,
+    HTTPResponse,
+    Metadata,
+)
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+class LoggingPlugin(BasePlugin):
+    """Plugin that logs HTTP request and response details."""
+
+    async def GetMetadata(self, request: Empty, context: ServicerContext) -> Metadata:
+        """Return plugin metadata."""
+        return Metadata(
+            name="logging-plugin",
+            version="1.0.0",
+            description="Logs HTTP request and response details for observability",
+        )
+
+    async def GetCapabilities(self, request: Empty, context: ServicerContext) -> Capabilities:
+        """Declare support for both request and response flows."""
+        return Capabilities(flows=[FLOW_REQUEST, FLOW_RESPONSE])
+
+    async def HandleRequest(self, request: HTTPRequest, context: ServicerContext) -> HTTPResponse:
+        """Log incoming request details."""
+        logger.info("=" * 80)
+        logger.info("INCOMING REQUEST")
+        logger.info(f"Method: {request.method}")
+        logger.info(f"URL: {request.url}")
+        logger.info(f"Path: {request.path}")
+        logger.info(f"Remote Address: {request.remote_addr}")
+
+        # Log headers.
+        logger.info("Headers:")
+        for key, value in request.headers.items():
+            # Mask sensitive headers.
+            if key.lower() in ("authorization", "cookie"):
+                value = "***REDACTED***"
+            logger.info(f"  {key}: {value}")
+
+        # Log body size.
+        if request.body:
+            logger.info(f"Body size: {len(request.body)} bytes")
+
+        logger.info("=" * 80)
+
+        # Continue processing.
+        return HTTPResponse(**{"continue": True})
+
+    async def HandleResponse(self, request: HTTPResponse, context: ServicerContext) -> HTTPResponse:
+        """Log outgoing response details."""
+        logger.info("=" * 80)
+        logger.info("OUTGOING RESPONSE")
+        logger.info(f"Status Code: {request.status_code}")
+
+        # Log headers.
+        logger.info("Headers:")
+        for key, value in request.headers.items():
+            logger.info(f"  {key}: {value}")
+
+        # Log body size.
+        if request.body:
+            logger.info(f"Body size: {len(request.body)} bytes")
+
+        logger.info("=" * 80)
+
+        # Continue processing.
+        return HTTPResponse(**{"continue": True})
+
+
+if __name__ == "__main__":
+    asyncio.run(serve(LoggingPlugin()))

examples/rate_limit_plugin/README.md

@@ -0,0 +1,36 @@
+# Rate Limit Plugin Example
+
+A plugin that implements rate limiting using a token bucket algorithm to prevent clients from overwhelming the server.
+
+## What it does
+
+- Tracks requests per client IP address
+- Limits each client to a configurable number of requests per minute (default: 60)
+- Returns 429 Too Many Requests when limit is exceeded
+- Adds rate limit headers to responses (`X-RateLimit-Limit`, `X-RateLimit-Remaining`)
+- Includes `Retry-After` header when rate limit is exceeded
+
+## Running the example
+
+```bash
+# From the repository root
+cd examples/rate_limit_plugin
+uv run python main.py
+```
+
+## Configuration
+
+You can customize the rate limit by modifying the `requests_per_minute` parameter when creating the plugin:
+
+```python
+asyncio.run(serve(RateLimitPlugin(requests_per_minute=100)))
+```
+
+## Key concepts demonstrated
+
+- Stateful plugin implementation (tracking client buckets)
+- Token bucket algorithm for rate limiting
+- Custom response headers
+- Calculating and returning `Retry-After` header
+- Per-client tracking using remote IP address
+- Request rejection with appropriate HTTP status codes

examples/rate_limit_plugin/__init__.py

@@ -0,0 +1 @@
+"""Rate limiting plugin example."""

examples/rate_limit_plugin/main.py

@@ -0,0 +1,107 @@
+"""Rate limiting plugin using a simple token bucket algorithm.
+
+This example demonstrates stateful request processing and rate limiting.
+"""
+
+import asyncio
+import logging
+import time
+from collections import defaultdict
+
+from google.protobuf.empty_pb2 import Empty
+from grpc import ServicerContext
+
+from mcpd_plugins import BasePlugin, serve
+from mcpd_plugins.v1.plugins.plugin_pb2 import (
+    FLOW_REQUEST,
+    Capabilities,
+    HTTPRequest,
+    HTTPResponse,
+    Metadata,
+)
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class RateLimitPlugin(BasePlugin):
+    """Plugin that implements rate limiting using token bucket algorithm."""
+
+    def __init__(self, requests_per_minute: int = 60):
+        """Initialize the rate limiter.
+
+        Args:
+            requests_per_minute: Maximum requests allowed per minute per client.
+        """
+        super().__init__()
+        self.requests_per_minute = requests_per_minute
+        self.rate_per_second = requests_per_minute / 60.0
+
+        # Track tokens for each client IP.
+        self.buckets: dict[str, float] = defaultdict(lambda: float(requests_per_minute))
+        self.last_update: dict[str, float] = defaultdict(time.time)
+
+    async def GetMetadata(self, request: Empty, context: ServicerContext) -> Metadata:
+        """Return plugin metadata."""
+        return Metadata(
+            name="rate-limit-plugin",
+            version="1.0.0",
+            description=f"Rate limits requests to {self.requests_per_minute} per minute per client",
+        )
+
+    async def GetCapabilities(self, request: Empty, context: ServicerContext) -> Capabilities:
+        """Declare support for request flow."""
+        return Capabilities(flows=[FLOW_REQUEST])
+
+    async def HandleRequest(self, request: HTTPRequest, context: ServicerContext) -> HTTPResponse:
+        """Apply rate limiting based on client IP."""
+        client_ip = request.remote_addr or "unknown"
+        logger.info(f"Rate limit check for {client_ip}: {request.method} {request.url}")
+
+        # Refill tokens based on time elapsed.
+        now = time.time()
+        elapsed = now - self.last_update[client_ip]
+        self.buckets[client_ip] = min(
+            self.requests_per_minute,
+            self.buckets[client_ip] + elapsed * self.rate_per_second,
+        )
+        self.last_update[client_ip] = now
+
+        # Check if client has tokens available.
+        if self.buckets[client_ip] < 1.0:
+            logger.warning(f"Rate limit exceeded for {client_ip}")
+            return self._rate_limit_response(client_ip)
+
+        # Consume one token.
+        self.buckets[client_ip] -= 1.0
+        logger.info(f"Request allowed for {client_ip} (tokens remaining: {self.buckets[client_ip]:.2f})")
+
+        # Add rate limit headers to response.
+        response = HTTPResponse(**{"continue": True})
+        response.modified_request.CopyFrom(request)
+        response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute)
+        response.headers["X-RateLimit-Remaining"] = str(int(self.buckets[client_ip]))
+
+        return response
+
+    def _rate_limit_response(self, client_ip: str) -> HTTPResponse:
+        """Create a 429 Too Many Requests response."""
+        # Calculate retry-after in seconds.
+        tokens_needed = 1.0 - self.buckets[client_ip]
+        retry_after = int(tokens_needed / self.rate_per_second) + 1
+
+        response = HTTPResponse(
+            **{"continue": False},
+            status_code=429,
+            body=b'{"error": "Rate limit exceeded"}',
+        )
+        response.headers["Content-Type"] = "application/json"
+        response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute)
+        response.headers["X-RateLimit-Remaining"] = "0"
+        response.headers["Retry-After"] = str(retry_after)
+
+        return response
+
+
+if __name__ == "__main__":
+    asyncio.run(serve(RateLimitPlugin(requests_per_minute=60)))