[User Input / Web UI] ──> [Next.js 16 App Router (Agent Orchestrator)]
                                    │
                                    ▼ (MCP JSON-RPC Protocol Layer)
                        [FastMCP Hardened Proxy Firewall]
                                    │
                       ┌────────────┴────────────┐
                       ▼                         ▼
            [DB Tool Server]          [API Integration Tool]

import os
import re
import json
import logging
from typing import Dict, Any, Optional
from mcp.server.fastmcp import FastMCP, Context
from pydantic import BaseModel, Field, validator

# Configure Audit Logging
logging.basicConfig(
    level=logging.INFO,
    format='%(asctime)s [%(levelname)s] [MCP-AUDIT] %(message)s'
)
logger = logging.getLogger("mcp_security")

# FastMCP Server Initialization with Strict Metadata
mcp = FastMCP(
    name="SecureDataQueryService",
    version="1.4.0",
    dependencies=["pydantic", "mcp"]
)

# Input Sanitization Rule Set
DANGEROUS_PATTERNS = re.compile(
    r'(;\s*DROP\s+TABLE|--|\bEXEC\b|<script|javascript:|sudo\s|/bin/sh)',
    re.IGNORECASE
)

class QueryInputSchema(BaseModel):
    query_id: str = Field(..., description="Alphanumeric query identifier", max_length=64)
    filter_tag: Optional[str] = Field(None, description="Category filter tag", max_length=32)

    @validator('query_id', 'filter_tag')
    def validate_safe_input(cls, v):
        if v and DANGEROUS_PATTERNS.search(v):
            logger.warning(f"Security Alert: Malicious pattern detected in input parameter: {v}")
            raise ValueError("Input parameter contains disallowed control sequences.")
        return v

@mcp.tool(
    name="execute_secure_search",
    description="Searches authorized documentation records by ID. Inputs are strictly validated."
)
def execute_secure_search(input_data: QueryInputSchema, ctx: Context) -> str:
    """
    Secure tool handler demonstrating context inspection and zero-trust validation.
    """
    # 1. Identity & Session Verification via Request Metadata
    client_id = ctx.request_context.meta.get("client_id") if ctx.request_context else "anonymous"
    user_role = ctx.request_context.meta.get("user_role") if ctx.request_context else "guest"

    logger.info(f"Tool Execution Requested | Client: {client_id} | Role: {user_role} | Query: {input_data.query_id}")

    # 2. Enforcement of Role-Based Authorization
    if user_role not in ["analyst", "admin"]:
        logger.error(f"Unauthorized Access Attempt | Client: {client_id} | Role: {user_role}")
        return json.dumps({
            "status": "error",
            "code": 403,
            "message": "Access Denied: Insufficient privilege level for tool execution."
        })

    # 3. Safe Execution Logic (Deterministic Output Processing)
    sanitized_id = re.sub(r'[^a-zA-Z0-9_-]', '', input_data.query_id)
    
    return json.dumps({
        "status": "success",
        "record_id": sanitized_id,
        "data": f"Validated record response for identifier: {sanitized_id}"
    })

if __name__ == "__main__":
    mcp.run()

import { NextRequest, NextResponse } from 'next/server';
import { z } from 'zod';

// Define Strict Schema for Agent Request Payload
const AgentRequestSchema = z.object({
  userMessage: z.string().min(1).max(2000),
  sessionToken: z.string().min(10),
});

// Output Defense: Sanitize MCP Tool Response prior to LLM Context Injection
function sanitizeToolOutputForContext(rawOutput: string): string {
  // Neutralize potential indirect prompt injection vectors
  return rawOutput
    .replace(/<\|im_start\|>/g, '')
    .replace(/<\|im_end\|>/g, '')
    .replace(/SYSTEM INSTRUCTION:/gi, '[FILTERED_INSTRUCTION_ATTEMPT]')
    .replace(/IGNORE ALL PREVIOUS INSTRUCTIONS/gi, '[FILTERED_INJECTION_ATTEMPT]');
}

export async function POST(req: NextRequest) {
  try {
    const body = await req.json();
    const validatedData = AgentRequestSchema.parse(body);

    // 1. Session Verification & RBAC Retrieval
    const userRole = req.headers.get('x-user-role') || 'guest';
    const clientId = req.headers.get('x-client-id') || 'unknown';

    // 2. Transmit Secure Metadata to MCP Host Execution Client
    const mcpHeaders = {
      'Content-Type': 'application/json',
      'X-Client-ID': clientId,
      'X-User-Role': userRole,
      'Authorization': `Bearer ${validatedData.sessionToken}`
    };

    // Simulated Safe Tool Call Execution to FastMCP Service
    const toolCallPayload = {
      jsonrpc: "2.0",
      id: 1,
      method: "tools/call",
      params: {
        name: "execute_secure_search",
        arguments: {
          query_id: "DOC-2026-9912",
          filter_tag: "security"
        }
      }
    };

    // Forward request with injected RBAC metadata headers
    // const mcpResponse = await fetch('http://localhost:8000/mcp', { method: 'POST', headers: mcpHeaders, body: JSON.stringify(toolCallPayload) });

    const simulatedMcpResult = '{"status": "success", "record_id": "DOC-2026-9912", "data": "Sample secure telemetry"}';
    const safeOutput = sanitizeToolOutputForContext(simulatedMcpResult);

    return NextResponse.json({
      success: true,
      agentResponse: `Processed response safely: ${safeOutput}`
    });

  } catch (error: any) {
    return NextResponse.json(
      { success: false, error: error.message || 'Invalid Request' },
      { status: 400 }
    );
  }
}

// Azure Sentinel / KQL Detection Query for MCP Indirect Prompt Injection
MCPTelemetryLogs_CL
| where ToolName_s == "execute_secure_search"
| where ParameterValue_s has_any ("IGNORE PREVIOUS", "SYSTEM PROMPT", "DROP TABLE", "<script>")
| summarize SuspiciousEvents = count() by ClientIP_s, UserRole_s, bin(TimeGenerated, 5m)
| where SuspiciousEvents > 3