MCP Server and PostgreSQL Sample - Complete Walkthrough

Table of Contents

1. Overview
2. Architecture Deep Dive
3. Building the Solution
4. Component Breakdown
5. Deployment Guide
6. Using the Solution
7. Advanced Features
8. Troubleshooting
9. Best Practices

Overview

This walkthrough deconstructs how to build and utilize a production-ready Model Context Protocol (MCP) server that integrates with PostgreSQL and Azure AI services. The sample demonstrates enterprise-grade patterns including Row Level Security, semantic search, and multi-tenant data access.

What You'll Learn

• How to architect an MCP server with database integration
• Implementing Row Level Security for multi-tenant scenarios
• Building semantic search with Azure OpenAI embeddings
• Creating Docker-based development environments
• Deploying Azure infrastructure with Bicep templates
• Integrating with VS Code for AI-powered analytics

Technologies Used

• MCP Protocol: Model Context Protocol for AI tool integration
• FastMCP: Modern Python MCP server framework
• PostgreSQL: Database with pgvector extension for semantic search
• Azure OpenAI: Text embeddings and optional GPT models
• Docker: Containerization for consistent environments
• Bicep: Infrastructure as Code for Azure resources
• VS Code: Development environment with MCP integration

📚 Structured Learning Guide: /walkthrough

In addition to this technical walkthrough, this repository includes a comprehensive 12-module learning guide located in the /walkthrough directory. This structured approach breaks down the complex implementation into digestible learning modules, perfect for developers who want to understand each component step-by-step.

Learning Modules Overview

Module	Topic	Focus	Duration
00-Introduction	MCP Fundamentals	Core concepts, Zava Retail case study, architecture overview	30 min
01-Architecture	System Design	Technical architecture, design patterns, component relationships	45 min
02-Security	Enterprise Security	Azure authentication, Row Level Security, multi-tenant isolation	60 min
03-Setup	Environment Setup	Docker configuration, Azure CLI, project initialization	45 min
04-Database	Data Layer	PostgreSQL schema, pgvector setup, RLS policies, sample data	60 min
05-MCP-Server	Core Implementation	FastMCP framework, database integration, tool development	90 min
06-Tools	Tool Development	MCP tool creation, query validation, business intelligence	75 min
07-Semantic-Search	AI Integration	Azure OpenAI embeddings, vector search, hybrid queries	60 min
08-Testing	Quality Assurance	Testing strategies, debugging techniques, performance testing	75 min
09-VS-Code	Development Experience	VS Code configuration, AI Chat integration, debugging workflows	45 min
10-Deployment	Production Deployment	Containerization, Azure Container Apps, CI/CD pipelines	90 min
11-Monitoring	Observability	Application Insights, structured logging, performance metrics	60 min
12-Best-Practices	Production Excellence	Security hardening, optimization, enterprise patterns	45 min

How to Use the Learning Guide

📖 For Learning: The /walkthrough modules provide step-by-step instructions with explanations of why each component is designed the way it is. Start with Module 00 and progress sequentially.

🔧 For Implementation: This Sample_Walkthrough.md provides the technical deep-dive and code breakdown for developers who want to understand the complete implementation quickly.

🚀 For Production: Modules 02, 10, 11, and 12 focus specifically on production-ready deployment, security, and monitoring considerations.

📚 Complete Learning Path: Visit /walkthrough/README.md for the full learning guide overview with detailed learning objectives and prerequisites.

---

Architecture Deep Dive

High-Level Architecture

┌─────────────────┐    ┌─────────────────┐    ┌─────────────────┐
│   VS Code AI    │    │   MCP Server    │    │   PostgreSQL    │
│     Client      │◄──►│  (FastMCP)      │◄──►│   + pgvector    │
│                 │    │                 │    │                 │
└─────────────────┘    └─────────────────┘    └─────────────────┘
                                │
                                ▼
                       ┌─────────────────┐
                       │  Azure OpenAI   │
                       │   Embeddings    │
                       └─────────────────┘

Core Components

1. MCP Server (sales_analysis.py)

• FastMCP Framework: Handles HTTP/SSE protocol communication
• Tool Registration: Exposes database query and schema tools
• Request Context: Manages RLS user identification
• Error Handling: Robust error management and logging

2. Database Layer (sales_analysis_postgres.py)

• Connection Pooling: Efficient asyncpg connection management
• Schema Provider: Dynamic table schema discovery
• Query Execution: Secure SQL execution with RLS
• Semantic Search: Vector similarity search with pgvector

3. Configuration Management (config.py)

• Environment Variables: Centralized configuration handling
• Connection Parameters: Database and Azure service configuration
• Validation: Required environment variable validation

4. Infrastructure (infra/)

• Bicep Templates: Declarative Azure resource provisioning
• Model Deployment: Automated AI model deployment
• Role Assignments: Security role configuration

Data Flow

1. VS Code AI Client sends query
2. MCP Server receives request with RLS headers
3. Server extracts user identity and sets context
4. Database queries execute with RLS filtering
5. Results return through MCP protocol
6. AI Client processes structured response

---

Building the Solution

Step 1: Project Structure Setup

project/
├── mcp_server/              # MCP server implementation
│   ├── __init__.py         # Package initialization
│   ├── sales_analysis.py   # Main MCP server
│   ├── sales_analysis_postgres.py  # Database layer
│   ├── sales_analysis_text_embeddings.py  # Semantic search
│   └── config.py           # Configuration management
├── infra/                  # Infrastructure as Code
│   ├── main.bicep          # Main deployment template
│   ├── foundry.bicep       # Azure AI Foundry setup
│   ├── deploy.ps1          # Windows deployment script
│   └── deploy.sh           # Unix deployment script
├── data/                   # Database backup and initialization
├── docker-init/            # Database initialization scripts
├── .vscode/                # VS Code MCP configuration
├── docker-compose.yml      # Development environment
├── Dockerfile             # MCP server container
└── requirements.lock.txt   # Python dependencies

Step 2: Core Dependencies

Python Requirements:

# MCP Framework
mcp[server]>=0.5.0
fastmcp>=0.4.0

# Database Integration
asyncpg>=0.29.0
asyncio-rlock>=0.3.0

# Azure Integration
azure-ai-projects>=1.0.0
azure-identity>=1.19.0
azure-monitor-opentelemetry>=1.7.0

# Data Processing
pydantic>=2.9.0
numpy>=1.24.0

# Development
python-dotenv>=1.0.0

System Requirements:

• Docker Desktop for containerization
• Azure CLI for deployment
• PostgreSQL with pgvector extension
• VS Code with AI extensions

Step 3: Database Schema Design

The sample uses a retail database with these key tables:

-- Core business entities
retail.stores          -- Store locations and metadata
retail.customers       -- Customer profiles
retail.categories      -- Product categorization
retail.product_types   -- Product type definitions
retail.products        -- Product catalog
retail.orders          -- Customer orders
retail.order_items     -- Order line items
retail.inventory       -- Stock levels

-- Semantic search support
retail.product_description_embeddings  -- Vector embeddings for products

Row Level Security (RLS) Implementation:

-- Enable RLS on tables
ALTER TABLE retail.orders ENABLE ROW LEVEL SECURITY;
ALTER TABLE retail.customers ENABLE ROW LEVEL SECURITY;

-- Create policies based on store association
CREATE POLICY store_policy ON retail.orders
  FOR ALL TO PUBLIC
  USING (store_id = get_user_store_id());

---

Component Breakdown

MCP Server Core (sales_analysis.py)

Tool Registration Pattern

@mcp.tool()
async def execute_sales_query(
    ctx: Context,
    postgresql_query: Annotated[str, Field(description="A well-formed PostgreSQL query.")],
) -> str:
    """Execute PostgreSQL queries with Row Level Security."""
    rls_user_id = get_rls_user_id(ctx)
    
    try:
        return await db_provider.execute_query(
            postgresql_query, rls_user_id=rls_user_id
        )
    except Exception as e:
        logger.error("Error executing database query: %s", e)
        return f"Error executing database query: {e!s}"

Key Features:

• Type Annotations: Pydantic field descriptions for AI understanding
• Context Extraction: User identity from HTTP headers
• Error Handling: Graceful failure with informative messages
• Logging: Comprehensive operation logging

Request Context Management

def get_rls_user_id(ctx: Context) -> str:
    """Extract Row Level Security User ID from request context."""
    rls_user_id = get_header(ctx, "x-rls-user-id")
    if rls_user_id is None:
        # Default to placeholder if not provided
        rls_user_id = "00000000-0000-0000-0000-000000000000"
    return rls_user_id

Database Layer (sales_analysis_postgres.py)

Connection Pool Management

class PostgreSQLSchemaProvider:
    async def create_pool(self) -> None:
        """Create connection pool for better resource management."""
        if self.connection_pool is None:
            config_copy = dict(self.postgres_config)
            existing_server_settings = config_copy.pop("server_settings", {})
            
            merged_server_settings = {
                **existing_server_settings,
                "jit": "off",  # Disable JIT to reduce memory usage
                "work_mem": "4MB",  # Limit work memory per query
                "statement_timeout": "30s",  # 30 second statement timeout
            }
            
            self.connection_pool = await asyncpg.create_pool(
                **config_copy,
                min_size=1,
                max_size=3,  # Conservative pool size
                command_timeout=30,
                server_settings=merged_server_settings,
            )

Design Patterns:

• Resource Management: Proper pool lifecycle management
• Performance Tuning: Optimized PostgreSQL settings
• Error Recovery: Connection retry and fallback logic
• Security: RLS context setting per connection

Schema Introspection

async def get_table_schema(self, table_name: str, rls_user_id: str) -> Dict[str, Any]:
    """Return comprehensive schema information for a table."""
    conn = await self.get_connection()
    
    # Set RLS context
    await conn.execute(
        "SELECT set_config('app.current_rls_user_id', \$1, false)", 
        rls_user_id
    )
    
    # Get column information
    columns = await conn.fetch("""
        SELECT column_name, data_type, is_nullable, column_default
        FROM information_schema.columns 
        WHERE table_schema = \$1 AND table_name = \$2
        ORDER BY ordinal_position
    """, schema_name, table_name)
    
    # Get foreign key relationships
    foreign_keys = await conn.fetch("""
        SELECT kcu.column_name, ccu.table_name AS foreign_table_name
        FROM information_schema.table_constraints tc
        JOIN information_schema.key_column_usage kcu ON ...
    """)

Semantic Search Integration

Embedding Generation

class SemanticSearchTextEmbedding:
    def generate_query_embedding(self, query: str) -> Optional[List[float]]:
        """Generate embeddings using Azure OpenAI."""
        try:
            response = self.client.embeddings.create(
                input=[query],
                model=self.deployment_name
            )
            return response.data[0].embedding
        except Exception as e:
            logger.error("Embedding generation failed: %s", e)
            return None

Vector Similarity Search

async def search_products_by_similarity(
    self,
    query_embedding: List[float],
    rls_user_id: str,
    max_rows: int = 20,
    similarity_threshold: float = 30.0,
) -> str:
    """Search products using pgvector cosine similarity."""
    
    # Convert similarity percentage to distance threshold
    distance_threshold = 1.0 - (similarity_threshold / 100.0)
    
    query = f"""
        SELECT p.*, (pde.description_embedding <=> \$1::vector) as distance
        FROM {SCHEMA_NAME}.product_description_embeddings pde
        JOIN {SCHEMA_NAME}.products p ON pde.product_id = p.product_id
        WHERE (pde.description_embedding <=> \$1::vector) <= \$3
        ORDER BY distance
        LIMIT \$2
    """
    
    rows = await conn.fetch(query, embedding_str, max_rows, distance_threshold)

---

Deployment Guide

Azure Infrastructure Deployment

1. Bicep Template Structure

Main Template (main.bicep):

targetScope = 'subscription'

// Core parameters
param resourcePrefix string
param location string
param models array = [
  {
    name: 'text-embedding-3-small'
    format: 'OpenAI'
    version: '1'
    capacity: 50
  }
]

// Deploy foundry and project resources
module foundry 'foundry.bicep' = {
  name: 'foundry-account-deployment'
  scope: rg
  params: {
    foundryResourceName: foundryResourceName
    location: location
  }
}

module foundryProject 'foundry-project.bicep' = {
  name: 'foundry-project-deployment'
  scope: rg
  dependsOn: [foundry]
  params: {
    foundryResourceName: foundry.outputs.accountName
    aiProjectName: aiProjectName
  }
}

2. Deployment Automation

PowerShell Deployment (deploy.ps1):

# Generate unique suffix for resources
$UNIQUE_SUFFIX = -join ((97..122) + (48..57) | Get-Random -Count 4 | ForEach-Object { [char]$_ })

# Deploy Azure resources
az deployment sub create `
  --name "$DEPLOYMENT_NAME" `
  --location "$RG_LOCATION" `
  --template-file main.bicep `
  --parameters location="$RG_LOCATION" `
  --parameters resourcePrefix="$RESOURCE_PREFIX" `
  --parameters uniqueSuffix="$UNIQUE_SUFFIX" `
  --parameters models="$modelsJson"

# Create service principal for authentication
$spResult = az ad sp create-for-rbac `
    --name "zava-mcp-server-sp" `
    --role "Cognitive Services OpenAI User" `
    --scopes "/subscriptions/$SubId/resourceGroups/$RESOURCE_GROUP_NAME"

# Generate .env file with configuration
@"
PROJECT_ENDPOINT=$PROJECTS_ENDPOINT
AZURE_OPENAI_ENDPOINT=$AZURE_OPENAI_ENDPOINT
EMBEDDING_MODEL_DEPLOYMENT_NAME="text-embedding-3-small"
AZURE_CLIENT_ID=$clientId
AZURE_CLIENT_SECRET=$clientSecret
AZURE_TENANT_ID=$tenantId
"@ | Out-File -FilePath "../.env"

Local Development Setup

1. Docker Compose Configuration

# docker-compose.yml
version: '3.8'
services:
  postgres:
    image: pgvector/pgvector:pg17
    environment:
      - POSTGRES_INITDB_ARGS=--auth-host=scram-sha-256
    volumes:
      - ./data:/backup_data:ro
      - ./docker-init:/docker-entrypoint-initdb.d:ro
    ports:
      - "5432:5432"
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U postgres -d zava"]
      interval: 15s
      retries: 5

  mcp_server:
    build: .
    depends_on:
      postgres:
        condition: service_healthy
    ports:
      - "8000:8000"
    env_file:
      - .env

2. Database Initialization

# docker-init/init-db.sh
#!/bin/bash
set -e

# Create extensions
psql -v ON_ERROR_STOP=1 --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" <<-EOSQL
    CREATE EXTENSION IF NOT EXISTS vector;
    CREATE EXTENSION IF NOT EXISTS "uuid-ossp";
EOSQL

# Restore database backup
if [ -f /backup_data/zava_retail_2025_07_21_postgres_rls.backup ]; then
    pg_restore --username "$POSTGRES_USER" --dbname "$POSTGRES_DB" \
               --verbose --clean --no-acl --no-owner \
               /backup_data/zava_retail_2025_07_21_postgres_rls.backup
fi

---

Using the Solution

VS Code Integration

1. MCP Configuration (.vscode/mcp.json)

{
    "servers": {
        "zava-sales-analysis-headoffice": {
            "url": "http://127.0.0.1:8000/mcp",
            "type": "http",
            "headers": {"x-rls-user-id": "00000000-0000-0000-0000-000000000000"}
        },
        "zava-sales-analysis-seattle": {
            "url": "http://127.0.0.1:8000/mcp",
            "type": "http", 
            "headers": {"x-rls-user-id": "f47ac10b-58cc-4372-a567-0e02b2c3d479"}
        }
    }
}

2. Query Examples

Schema Discovery:

AI: #zava What tables are available in the database?

The MCP server uses get_multiple_table_schemas to return table structures

Sales Analysis:

AI: #zava Show me the top 10 products by revenue last quarter

Generates SQL with proper joins and date filtering

Semantic Search:

AI: #zava Find products similar to "waterproof electrical connectors"

Uses embeddings to find semantically similar products

Multi-Store Analysis:

# Switch to Seattle store manager context
AI: #zava-seattle What are our best-selling categories this month?

RLS ensures only Seattle store data is accessed

Advanced Query Patterns

1. Time-Series Analysis

-- Generated by AI through MCP server
SELECT 
    DATE_TRUNC('month', o.order_date) as month,
    SUM(oi.total_amount) as revenue,
    COUNT(DISTINCT o.order_id) as order_count
FROM retail.orders o
JOIN retail.order_items oi ON o.order_id = oi.order_id
WHERE o.order_date >= CURRENT_DATE - INTERVAL '12 months'
GROUP BY DATE_TRUNC('month', o.order_date)
ORDER BY month;

2. Product Performance with Categories

-- AI generates complex joins using schema information
SELECT 
    c.category_name,
    pt.type_name,
    COUNT(DISTINCT p.product_id) as product_count,
    SUM(oi.total_amount) as total_revenue,
    AVG(oi.unit_price) as avg_price
FROM retail.products p
JOIN retail.categories c ON p.category_id = c.category_id
JOIN retail.product_types pt ON p.product_type_id = pt.product_type_id
JOIN retail.order_items oi ON p.product_id = oi.product_id
GROUP BY c.category_name, pt.type_name
ORDER BY total_revenue DESC;

---

Advanced Features

Row Level Security Implementation

1. Policy Creation

-- Store-based access control
CREATE POLICY customer_store_policy ON retail.customers
  FOR ALL TO PUBLIC
  USING (store_id = get_current_store_id());

CREATE POLICY order_store_policy ON retail.orders  
  FOR ALL TO PUBLIC
  USING (store_id = get_current_store_id());

-- Function to get current user's store
CREATE OR REPLACE FUNCTION get_current_store_id()
RETURNS uuid AS $$
BEGIN
  RETURN current_setting('app.current_rls_user_id')::uuid;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER;

2. Context Setting in MCP Server

async def execute_query(self, sql_query: str, rls_user_id: str) -> str:
    """Execute query with RLS context."""
    conn = await self.get_connection()
    
    # Set RLS context for this connection
    await conn.execute(
        "SELECT set_config('app.current_rls_user_id', \$1, false)", 
        rls_user_id
    )
    
    # Execute user query with RLS filtering
    rows = await conn.fetch(sql_query)
    return self.format_results(rows)

Semantic Search Deep Dive

1. Embedding Pipeline

# Generate embeddings for product descriptions
async def generate_product_embeddings():
    products = await get_all_products()
    
    for product in products:
        description = f"{product.name} {product.description} {product.category}"
        embedding = embedding_client.generate_embedding(description)
        
        await store_embedding(product.id, embedding)

2. Similarity Search Optimization

-- Create vector index for performance
CREATE INDEX idx_product_embeddings_vector 
ON retail.product_description_embeddings 
USING ivfflat (description_embedding vector_cosine_ops);

-- Optimized similarity query
SELECT p.*, 
       (pde.description_embedding <=> \$1::vector) as distance,
       (1 - (pde.description_embedding <=> \$1::vector)) * 100 as similarity_percent
FROM retail.product_description_embeddings pde
JOIN retail.products p ON pde.product_id = p.product_id
WHERE (pde.description_embedding <=> \$1::vector) < 0.7  -- 30% similarity threshold
ORDER BY distance
LIMIT 20;

Monitoring and Observability

1. Azure Application Insights Integration

# Configure telemetry
from azure.monitor.opentelemetry import configure_azure_monitor
from opentelemetry.instrumentation.starlette import StarletteInstrumentor

# Enable monitoring if configured
if config.applicationinsights_connection_string:
    configure_azure_monitor(
        connection_string=config.applicationinsights_connection_string
    )
    StarletteInstrumentor().instrument_app(mcp.sse_app())

2. Custom Metrics and Logging

# Query execution tracking
@contextmanager
async def track_query_execution(query_type: str):
    start_time = time.time()
    try:
        yield
        duration = time.time() - start_time
        logger.info("Query executed", extra={
            "query_type": query_type,
            "duration_ms": duration * 1000,
            "status": "success"
        })
    except Exception as e:
        duration = time.time() - start_time
        logger.error("Query failed", extra={
            "query_type": query_type,
            "duration_ms": duration * 1000,
            "status": "error",
            "error": str(e)
        })
        raise

---

Troubleshooting

Common Issues and Solutions

1. Database Connection Issues

# Connection diagnostics
async def diagnose_connection():
    try:
        pool = await asyncpg.create_pool(**connection_params, min_size=1)
        conn = await pool.acquire()
        result = await conn.fetchval("SELECT 1")
        await pool.release(conn)
        await pool.close()
        return True
    except Exception as e:
        logger.error("Connection failed: %s", e)
        return False

Common Fixes:

• Verify PostgreSQL is running: docker ps
• Check connection parameters in .env
• Validate database exists: docker exec -it pg17 psql -U postgres -l
• Test network connectivity: telnet localhost 5432

2. RLS Policy Issues

-- Debug RLS policies
SELECT schemaname, tablename, policyname, cmd, qual 
FROM pg_policies 
WHERE schemaname = 'retail';

-- Check current RLS setting
SELECT current_setting('app.current_rls_user_id');

-- Temporarily disable RLS for debugging
ALTER TABLE retail.orders DISABLE ROW LEVEL SECURITY;

3. Embedding Service Issues

# Test embedding generation
async def test_embeddings():
    try:
        test_text = "waterproof electrical connector"
        embedding = embedding_client.generate_embedding(test_text)
        logger.info("Embedding generated successfully: %d dimensions", len(embedding))
        return True
    except Exception as e:
        logger.error("Embedding test failed: %s", e)
        return False

Performance Optimization

1. Connection Pool Tuning

# Optimize for your workload
connection_pool = await asyncpg.create_pool(
    min_size=2,          # Minimum connections
    max_size=10,         # Maximum connections  
    max_inactive_connection_lifetime=300,  # 5 minutes
    command_timeout=30,   # Query timeout
    server_settings={
        "application_name": "mcp-server",
        "work_mem": "4MB",
        "shared_preload_libraries": "pg_stat_statements"
    }
)

2. Query Optimization

-- Add indexes for common query patterns
CREATE INDEX idx_orders_store_date 
ON retail.orders (store_id, order_date);

CREATE INDEX idx_order_items_product 
ON retail.order_items (product_id);

-- Analyze query performance
EXPLAIN (ANALYZE, BUFFERS) 
SELECT ... FROM retail.orders o JOIN retail.order_items oi ...;

---

Best Practices

Security Best Practices

1. Environment Variable Management

# Use strong, unique passwords
POSTGRES_PASSWORD=$(openssl rand -base64 32)

# Rotate service principal credentials regularly
az ad sp credential reset --id $SP_ID --credential-description "Rotated $(date)"

# Use Azure Key Vault in production
az keyvault secret set --vault-name $VAULT_NAME --name "db-password" --value $PASSWORD

2. RLS Implementation Guidelines

• Default Deny: Start with restrictive policies
• Audit Regularly: Monitor policy effectiveness
• Test Thoroughly: Validate access patterns
• Document Policies: Maintain clear documentation

3. Network Security

# Production docker-compose with network isolation
networks:
  internal:
    driver: bridge
    internal: true
  external:
    driver: bridge

services:
  postgres:
    networks:
      - internal
    # No external ports in production
  
  mcp_server:
    networks:
      - internal
      - external
    ports:
      - "127.0.0.1:8000:8000"  # Bind to localhost only

Development Best Practices

1. Error Handling Patterns

# Structured error responses
class MCPError(Exception):
    def __init__(self, message: str, error_type: str = "general"):
        self.message = message
        self.error_type = error_type
        super().__init__(message)

async def safe_execute_query(query: str, rls_user_id: str) -> str:
    try:
        return await db_provider.execute_query(query, rls_user_id)
    except asyncpg.PostgresError as e:
        logger.error("Database error: %s", e)
        return json.dumps({"error": "Database query failed", "type": "database"})
    except Exception as e:
        logger.error("Unexpected error: %s", e)
        return json.dumps({"error": "Internal server error", "type": "server"})

2. Testing Strategies

# Unit test example
@pytest.mark.asyncio
async def test_rls_isolation():
    """Test that RLS properly isolates store data."""
    
    # Test Seattle store manager
    seattle_results = await db_provider.execute_query(
        "SELECT COUNT(*) FROM retail.orders",
        rls_user_id="f47ac10b-58cc-4372-a567-0e02b2c3d479"
    )
    
    # Test Redmond store manager  
    redmond_results = await db_provider.execute_query(
        "SELECT COUNT(*) FROM retail.orders", 
        rls_user_id="e7f8a9b0-c1d2-3e4f-5678-90abcdef1234"
    )
    
    # Results should be different due to RLS
    assert seattle_results != redmond_results

3. Monitoring and Alerts

# Custom metrics for monitoring
from prometheus_client import Counter, Histogram, start_http_server

query_counter = Counter('mcp_queries_total', 'Total queries executed', ['query_type'])
query_duration = Histogram('mcp_query_duration_seconds', 'Query execution time')

@query_duration.time()
async def execute_query_with_metrics(query: str, rls_user_id: str):
    query_counter.labels(query_type='sales_analysis').inc()
    return await db_provider.execute_query(query, rls_user_id)

Deployment Best Practices

1. Infrastructure as Code

// Use parameter files for different environments
param environment string = 'dev'
param location string = 'westus2'

// Apply consistent naming conventions
var resourcePrefix = 'zava-mcp-${environment}'
var resourceGroupName = 'rg-${resourcePrefix}-${uniqueSuffix}'

// Use tags for resource management
var commonTags = {
  Environment: environment
  Project: 'zava-mcp-server'
  ManagedBy: 'bicep'
  CreatedDate: utcNow('yyyy-MM-dd')
}

2. CI/CD Pipeline Integration

# Azure DevOps pipeline example
- task: AzureCLI@2
  displayName: 'Deploy Infrastructure'
  inputs:
    azureSubscription: $(azureServiceConnection)
    scriptType: 'bash'
    scriptLocation: 'inlineScript'
    inlineScript: |
      az deployment sub create \
        --name "mcp-server-$(Build.BuildId)" \
        --location $(location) \
        --template-file infra/main.bicep \
        --parameters environment=$(environment)

- task: Docker@2
  displayName: 'Build and Push MCP Server'
  inputs:
    command: 'buildAndPush'
    repository: 'zava-mcp-server'
    tags: '$(Build.BuildId)'

---

This comprehensive walkthrough provides the foundation for building, deploying, and operating a production-ready MCP server with PostgreSQL integration. The patterns and practices demonstrated here can be extended to other domains and use cases while maintaining security, performance, and maintainability.