MODEL
CONTEXT
PROTOCOL

{
  "mcpServers": {
    "tars-tools": {
      "command": "python",
      "args": ["production_mcp_server.py"],
      "cwd": "/opt/tars-mcp",
      "env": {
        "PG_DSN": "postgresql://user:pass@host/db",
        "MINIO_ENDPOINT": "https://minio.local"
      },
      "permissions": [
        {
          "fs": {
            "read": ["/data/tars"],
            "write": ["/tmp"]
          }
        }
      ]
      // Note: "permissions" is a custom field for this example app
      // Not a standard MCP configuration field
    }
  }
}

FROM python:3.11-slim

WORKDIR /app

# Install dependencies
COPY requirements.txt .
RUN pip install -r requirements.txt

# Copy source code
COPY . .

# Create non-root user
RUN useradd --create-home app && chown -R app:app /app
USER app

# Expose port for HTTP mode
EXPOSE 8000

# Default command
CMD ["python", "production_mcp_server.py"]

import sqlite3
import json
import os
from typing import List, Dict, Any
from mcp.server.fastmcp import FastMCP
from mcp.types import Resource, Tool, TextContent

app = FastMCP("analytics-server")

# Database connection
DB_PATH = "analytics.db"

@app.tool()
def query_database(
    query: str,
    params: List[Any] = None
) -> Dict[str, Any]:
    """Execute SQL queries with safety checks"""
    # Safety: Only allow SELECT statements
    if not query.strip().upper().startswith('SELECT'):
        raise ValueError("Only SELECT queries are allowed")

    try:
        conn = sqlite3.connect(DB_PATH)
        conn.row_factory = sqlite3.Row
        cursor = conn.cursor()

        if params:
            cursor.execute(query, params)
        else:
            cursor.execute(query)

        results = [dict(row) for row in cursor.fetchall()]
        conn.close()

        return {
            "success": True,
            "data": results,
            "count": len(results)
        }
    except Exception as e:
        return {
            "success": False,
            "error": str(e)
        }

@app.tool()
def analyze_data(
    data: List[Dict[str, Any]],
    analysis_type: str = "summary"
) -> Dict[str, Any]:
    """Perform statistical analysis on data"""
    if not data:
        return {"error": "No data provided"}

    if analysis_type == "summary":
        numeric_fields = []
        for row in data:
            for key, value in row.items():
                if isinstance(value, (int, float)):
                    numeric_fields.append(key)

        summary = {}
        for field in set(numeric_fields):
            values = [row[field] for row in data if field in row]
            if values:
                summary[field] = {
                    "count": len(values),
                    "min": min(values),
                    "max": max(values),
                    "avg": sum(values) / len(values)
                }

        return {
            "type": "summary",
            "total_records": len(data),
            "numeric_analysis": summary
        }

    return {"error": f"Analysis type '{analysis_type}' not supported"}

@app.tool()
def create_report(
    title: str,
    data: Dict[str, Any],
    output_format: str = "json"
) -> Dict[str, Any]:
    """Generate and save analysis reports"""
    report = {
        "title": title,
        "timestamp": "2024-01-15T10:30:00Z",
        "data": data,
        "metadata": {
            "generated_by": "MCP Analytics Server",
            "format": output_format
        }
    }

    filename = f"reports/{title.lower().replace(' ', '_')}.{output_format}"
    os.makedirs("reports", exist_ok=True)

    if output_format == "json":
        with open(filename, 'w') as f:
            json.dump(report, f, indent=2)

    return {
        "success": True,
        "report_path": filename,
        "report_size": os.path.getsize(filename) if os.path.exists(filename) else 0
    }

# Resource: Expose database schema (FastMCP style)
@app.resource("schema://database", mime_type="application/json")
def get_database_schema() -> str:
    """Get database schema information"""
    try:
        conn = sqlite3.connect(DB_PATH)
        cursor = conn.cursor()

        # Get all tables
        cursor.execute("SELECT name FROM sqlite_master WHERE type='table'")
        tables = [row[0] for row in cursor.fetchall()]

        schema = {}
        for table in tables:
            cursor.execute(f"PRAGMA table_info({table})")
            columns = [
                {
                    "name": row[1],
                    "type": row[2],
                    "nullable": not row[3],
                    "primary_key": bool(row[5])
                }
                for row in cursor.fetchall()
            ]
            schema[table] = columns

        conn.close()
        return json.dumps(schema, indent=2)

    except Exception as e:
        return json.dumps({
            "error": f"Error accessing schema: {str(e)}"
        })

# Initialize database
def init_database():
    """Create sample database with users table"""
    conn = sqlite3.connect(DB_PATH)
    cursor = conn.cursor()

    cursor.execute("""
        CREATE TABLE IF NOT EXISTS users (
            id INTEGER PRIMARY KEY,
            age INTEGER,
            signup_date TEXT,
            last_active TEXT,
            active INTEGER DEFAULT 1
        )
    """)

    # Insert sample data
    cursor.execute("DELETE FROM users")  # Clear existing
    sample_users = [
        (25, '2023-01-15', '2024-01-10', 1),
        (34, '2022-06-20', '2024-01-12', 1),
        (29, '2023-03-10', '2024-01-11', 1)
    ]
    cursor.executemany(
        "INSERT INTO users (age, signup_date, last_active, active) VALUES (?, ?, ?, ?)",
        sample_users
    )

    conn.commit()
    conn.close()

if __name__ == "__main__":
    init_database()
    app.run()

import sqlite3
import json
import asyncio
from typing import List, Dict, Any
from mcp.server.lowlevel import Server
from mcp.types import Resource, Tool, TextContent
import mcp.types as types

# Alternative: Official MCP SDK approach
server = Server("analytics-server")

@server.list_tools()
async def list_tools() -> List[Tool]:
    return [
        Tool(
            name="query_database",
            description="Execute SQL queries with safety checks",
            inputSchema={
                "type": "object",
                "properties": {
                    "query": {"type": "string"},
                    "params": {"type": "array", "items": {"type": "string"}}
                },
                "required": ["query"]
            }
        )
    ]

@server.call_tool()
async def call_tool(name: str, arguments: dict[str, Any]):
    if name == "query_database":
        query = arguments.get("query", "")
        params = arguments.get("params", [])

        # Safety check
        if not query.strip().upper().startswith('SELECT'):
            # soit du contenu texte
            return [types.TextContent(type="text", text=json.dumps({"success": False, "error": "Only SELECT queries allowed"}))]

        # Execute query logic here...
        # soit du contenu texte
        return [types.TextContent(type="text", text=json.dumps({"success": True, "data": [], "count": 0}))]
        # ou un dict (structured output) si tu déclares un outputSchema côté tool
        # return {"success": True, "data": [], "count": 0}

    raise ValueError(f"Unknown tool: {name}")

async def main():
    from mcp.server.stdio import stdio_server
    async with stdio_server() as (read_stream, write_stream):
        await server.run(read_stream, write_stream, server.create_initialization_options())

if __name__ == "__main__":
    asyncio.run(main())

import asyncio
import json
from mcp import ClientSession, StdioServerParameters
from mcp.client.stdio import stdio_client

class AnalyticsClient:
    def __init__(self):
        self.session = None

    async def connect(self):
        """Establish connection to MCP server"""
        # Correct pattern: use StdioServerParameters and context managers
        server = StdioServerParameters(command="python", args=["production_mcp_server.py"])
        async with stdio_client(server) as (read, write):
            async with ClientSession(read, write) as session:
                await session.initialize()
                self.session = session

        print("Connected to MCP server")

    async def discover_capabilities(self):
        """Discover available tools and resources"""
        # List available tools
        tools_result = await self.session.list_tools()
        print(f"Available tools: {len(tools_result.tools)}")

        for tool in tools_result.tools:
            print(f"  {tool.name}: {tool.description}")

        # List available resources
        resources_result = await self.session.list_resources()
        print(f"Available resources: {len(resources_result.resources)}")

        for resource in resources_result.resources:
            print(f"  {resource.name} ({resource.uri})")

        return tools_result.tools, resources_result.resources

    async def get_database_schema(self):
        """Fetch database schema resource"""
        schema_resource = await self.session.read_resource(
            "schema://database"
        )

        schema_data = json.loads(schema_resource.contents[0].text)
        print("Database Schema:")

        for table, columns in schema_data.items():
            print(f"  Table: {table}")
            for col in columns:
                pk_marker = " (PK)" if col["primary_key"] else ""
                print(f"    - {col['name']}: {col['type']}{pk_marker}")

        return schema_data

    async def analyze_users(self):
        """Complete workflow: query -> analyze -> report"""
        print("\nStarting user analysis workflow...")

        # Step 1: Query user data
        query_result = await self.session.call_tool(
            "query_database",
            {
                "query": "SELECT age, signup_date, last_active FROM users WHERE active = 1",
                "params": []
            }
        )

        # Handle different content types safely
        if not query_result.content or not query_result.content[0]:
            print("❌ Query failed")
            return

        content = query_result.content[0]
        if hasattr(content, 'text') and content.text:
            user_data = json.loads(content.text)
        else:
            # Handle direct JSON content if available
            user_data = content if isinstance(content, dict) else {}

        if not user_data.get('success', False):
            print(f"❌ Query error: {user_data.get('error', 'Unknown error')}")
            return

        print(f"Retrieved {user_data['count']} active users")

        # Step 2: Analyze the data
        analysis_result = await self.session.call_tool(
            "analyze_data",
            {
                "data": user_data["data"],
                "analysis_type": "summary"
            }
        )

        # Handle analysis result safely
        analysis_content = analysis_result.content[0]
        if hasattr(analysis_content, 'text') and analysis_content.text:
            analysis = json.loads(analysis_content.text)
        else:
            analysis = analysis_content if isinstance(analysis_content, dict) else {}
        print("Analysis completed:")
        print(f"  Total records: {analysis['total_records']}")

        for field, stats in analysis['numeric_analysis'].items():
            print(f"  {field}: avg={stats['avg']:.1f}, range={stats['min']}-{stats['max']}")

        # Step 3: Generate report
        report_result = await self.session.call_tool(
            "create_report",
            {
                "title": "User Analytics Report",
                "data": analysis,
                "output_format": "json"
            }
        )

        report_info = json.loads(report_result.content[0].text)
        print(f"Report saved: {report_info['report_path']}")
        print(f"   Size: {report_info['report_size']} bytes")

    async def disconnect(self):
        """Clean disconnect"""
        if self.session:
            await self.session.close()
            print("Disconnected from MCP server")

# Usage example with correct pattern
async def main():
    server = StdioServerParameters(command="python", args=["production_mcp_server.py"])
    async with stdio_client(server) as (read, write):
        async with ClientSession(read, write) as session:
            await session.initialize()

            # Discover capabilities
            tools = await session.list_tools()
            print(f"Available tools: {[tool.name for tool in tools.tools]}")

            # Get database schema
            resources = await session.list_resources()
            if resources.resources:
                schema = await session.read_resource(resources.resources[0].uri)
                print(f"Database schema loaded")

            # Analyze users
            result = await session.call_tool("query_database", {
                "query": "SELECT COUNT(*) as user_count FROM users",
                "params": []
            })
            print(f"👥 Analysis complete: {result.content[0].text}")

if __name__ == "__main__":
    asyncio.run(main())

import os
from fastapi import FastAPI, Header, HTTPException, Request
from fastapi.middleware.cors import CORSMiddleware
from mcp.server.fastmcp import FastMCP

app = FastAPI()
mcp_app = FastMCP("secure-analytics-server")

# Security configuration
AUTH_TOKEN = os.getenv("MCP_AUTH_TOKEN")
ALLOWED_ORIGINS = os.getenv("ALLOWED_ORIGINS", "*").split(",")

# CORS middleware for browser clients
app.add_middleware(
    CORSMiddleware,
    allow_origins=ALLOWED_ORIGINS,
    allow_credentials=True,
    allow_methods=["POST", "GET", "OPTIONS"],
    allow_headers=["*"],
)

@app.middleware("http")
async def auth_middleware(request: Request, call_next):
    """Enforce Bearer token authentication for MCP endpoints"""
    if request.url.path.startswith("/mcp"):
        if not AUTH_TOKEN:
            raise HTTPException(
                status_code=500,
                detail="Server misconfiguration: AUTH_TOKEN not set"
            )

        auth_header = request.headers.get("authorization", "")
        if not auth_header.startswith("Bearer "):
            raise HTTPException(
                status_code=401,
                detail="Missing or invalid authorization header"
            )

        token = auth_header.split(" ", 1)[1]
        if token != AUTH_TOKEN:
            # Log failed authentication attempt
            print(f"Failed auth attempt from {request.client.host}: {token[:8]}...")
            raise HTTPException(
                status_code=401,
                detail="Invalid authentication token"
            )

    response = await call_next(request)
    return response

@app.get("/health")
async def health_check():
    """Public health check endpoint"""
    return {"status": "healthy", "auth": "required" if AUTH_TOKEN else "disabled"}

# Mount MCP app with authentication
app.mount("/mcp", mcp_app.app)

import re
from typing import Dict, List, Any
from enum import Enum

class Permission(Enum):
    READ_PUBLIC = "read:public"
    READ_PRIVATE = "read:private"
    WRITE_REPORTS = "write:reports"
    ADMIN_QUERY = "admin:query"

class CapabilityChecker:
    def __init__(self, user_permissions: List[str]):
        self.permissions = set(user_permissions)

    def can_access_database(self, query: str, schema: str = None) -> bool:
        """Check if user can execute specific database operations"""
        query_upper = query.strip().upper()

        # Only SELECT allowed for most users
        if not query_upper.startswith('SELECT'):
            return Permission.ADMIN_QUERY.value in self.permissions

        # Schema-based restrictions
        if schema and schema == "private":
            return Permission.READ_PRIVATE.value in self.permissions

        # Public data access
        return Permission.READ_PUBLIC.value in self.permissions

    def can_write_files(self, path: str) -> bool:
        """Check file write permissions with path restrictions"""
        if not Permission.WRITE_REPORTS.value in self.permissions:
            return False

        # Restrict to specific directories
        allowed_patterns = [
            r'^/tmp/reports/.*',
            r'^./reports/.*',
            r'^reports/.*'
        ]

        return any(re.match(pattern, path) for pattern in allowed_patterns)

# Example usage in MCP tools
@mcp_app.tool()
def secure_query_database(
    query: str,
    schema: str = "public",
    user_permissions: List[str] = ["read:public"]
) -> Dict[str, Any]:
    """Database query with capability-based security"""
    checker = CapabilityChecker(user_permissions)

    if not checker.can_access_database(query, schema):
        return {
            "success": False,
            "error": "Insufficient permissions for this operation",
            "required_permission": "admin:query" if not query.upper().startswith('SELECT') else f"read:{schema}"
        }

    # Whitelist allowed tables/columns
    if schema == "public":
        allowed_tables = ["users", "reports", "analytics"]
        # Extract table name from query (simplified)
        table_match = re.search(r'FROM\s+(\w+)', query, re.IGNORECASE)
        if table_match and table_match.group(1) not in allowed_tables:
            return {
                "success": False,
                "error": f"Access denied to table: {table_match.group(1)}"
            }

    # Execute query logic here...
    return {"success": True, "data": [], "permissions_used": list(checker.permissions)}

@mcp_app.tool()
def secure_create_report(
    title: str,
    data: Dict[str, Any],
    output_path: str = None,
    user_permissions: List[str] = ["read:public"]
) -> Dict[str, Any]:
    """Report creation with file system permissions"""
    checker = CapabilityChecker(user_permissions)

    # Default safe path if none provided
    if not output_path:
        output_path = f"reports/{title.lower().replace(' ', '_')}.json"

    if not checker.can_write_files(output_path):
        return {
            "success": False,
            "error": f"Cannot write to path: {output_path}",
            "allowed_patterns": ["reports/*", "/tmp/reports/*"]
        }

    # Create report logic here...
    return {
        "success": True,
        "report_path": output_path,
        "permissions_validated": True
    }