How to Create Claude MCP Server: The Complete Developer’s Guide

How to Create Claude MCP Server: Step-by-Step Implementation

Now we’ll dive into the practical aspects of how to create Claude MCP server implementations. This section provides concrete examples and code that you can adapt for your specific use cases. We’ll cover implementations in both Python and TypeScript, the two most popular languages for MCP server development.

Step 1: Environment Setup and Prerequisites

Before you begin building your MCP server, ensure your development environment has the necessary tools installed. For Python developers, you’ll need Python 3.8 or later. TypeScript developers should have Node.js 16 or later. Both approaches benefit from having Git installed for accessing the official MCP servers repository which contains valuable examples and reference implementations.

# For Python implementation
pip install mcp anthropic pydantic

# For TypeScript/Node.js implementation
npm install @modelcontextprotocol/sdk anthropic zod

Step 2: Creating a Basic MCP Server in Python

Let’s start with a simple Python implementation that demonstrates the core concepts. This example creates an MCP server that exposes a tool for querying weather information—a common use case that’s valuable whether you’re building applications for farmers in Punjab or delivery services in Manhattan.

from mcp.server import Server
from mcp.types import Tool, TextContent
import asyncio
import json

class WeatherMCPServer:
    def __init__(self):
        self.server = Server("weather-mcp-server")
        self.setup_tools()
    
    def setup_tools(self):
        # Define the weather query tool
        weather_tool = Tool(
            name="get_weather",
            description="Get current weather information for a specified location",
            inputSchema={
                "type": "object",
                "properties": {
                    "location": {
                        "type": "string",
                        "description": "City name or coordinates"
                    },
                    "units": {
                        "type": "string",
                        "enum": ["celsius", "fahrenheit"],
                        "description": "Temperature units",
                        "default": "celsius"
                    }
                },
                "required": ["location"]
            }
        )
        
        self.server.add_tool(weather_tool, self.handle_weather_request)
    
    async def handle_weather_request(self, arguments: dict) -> list[TextContent]:
        location = arguments.get("location")
        units = arguments.get("units", "celsius")
        
        # In production, you would call an actual weather API here
        # This example returns mock data for demonstration
        weather_data = {
            "location": location,
            "temperature": 25 if units == "celsius" else 77,
            "conditions": "Partly cloudy",
            "humidity": 65,
            "wind_speed": 12
        }
        
        response_text = f"Weather in {location}:\n"
        response_text += f"Temperature: {weather_data['temperature']}°{'C' if units == 'celsius' else 'F'}\n"
        response_text += f"Conditions: {weather_data['conditions']}\n"
        response_text += f"Humidity: {weather_data['humidity']}%\n"
        response_text += f"Wind Speed: {weather_data['wind_speed']} km/h"
        
        return [TextContent(type="text", text=response_text)]
    
    async def run(self):
        async with self.server:
            await self.server.run()

if __name__ == "__main__":
    server = WeatherMCPServer()
    asyncio.run(server.run())

Step 3: Creating an MCP Server with Database Integration

A more realistic example of how to create Claude MCP server implementations involves database integration. This TypeScript example shows how to create an MCP server that allows Claude to query a PostgreSQL database—a common scenario for enterprise applications managing customer data, inventory systems, or analytical databases.

import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import { Tool, TextContent } from '@modelcontextprotocol/sdk/types.js';
import { Pool } from 'pg';

class DatabaseMCPServer {
  private server: Server;
  private dbPool: Pool;

  constructor() {
    this.server = new Server({
      name: 'database-mcp-server',
      version: '1.0.0',
    });

    // Initialize database connection pool
    this.dbPool = new Pool({
      host: process.env.DB_HOST || 'localhost',
      port: parseInt(process.env.DB_PORT || '5432'),
      database: process.env.DB_NAME,
      user: process.env.DB_USER,
      password: process.env.DB_PASSWORD,
      max: 20,
      idleTimeoutMillis: 30000,
    });

    this.setupTools();
  }

  private setupTools(): void {
    // Define database query tool
    this.server.setRequestHandler('tools/list', async () => ({
      tools: [
        {
          name: 'query_customers',
          description: 'Query customer information from the database',
          inputSchema: {
            type: 'object',
            properties: {
              query_type: {
                type: 'string',
                enum: ['by_id', 'by_email', 'by_region', 'recent'],
                description: 'Type of customer query to perform',
              },
              value: {
                type: 'string',
                description: 'Search value (customer ID, email, or region name)',
},
              limit: {
                type: 'number',
                description: 'Maximum number of results to return',
                default: 10,
              },
            },
            required: ['query_type'],
          },
        } as Tool,
      ],
    }));

    // Handle tool execution
    this.server.setRequestHandler('tools/call', async (request) => {
      if (request.params.name === 'query_customers') {
        return await this.handleCustomerQuery(request.params.arguments);
      }
      throw new Error(`Unknown tool: ${request.params.name}`);
    });
  }

  private async handleCustomerQuery(args: any): Promise<{ content: TextContent[] }> {
    const { query_type, value, limit = 10 } = args;
    
    let query = '';
    let params: any[] = [];

    switch (query_type) {
      case 'by_id':
        query = 'SELECT * FROM customers WHERE customer_id = $1';
        params = [value];
        break;
      case 'by_email':
        query = 'SELECT * FROM customers WHERE email ILIKE $1 LIMIT $2';
        params = [`%${value}%`, limit];
        break;
      case 'by_region':
        query = 'SELECT * FROM customers WHERE region = $1 LIMIT $2';
        params = [value, limit];
        break;
      case 'recent':
        query = 'SELECT * FROM customers ORDER BY created_at DESC LIMIT $1';
        params = [limit];
        break;
      default:
        throw new Error('Invalid query type');
    }

    try {
      const result = await this.dbPool.query(query, params);
      const customers = result.rows;

      const responseText = `Found ${customers.length} customer(s):\n\n${
        customers.map((c, idx) => 
          `${idx + 1}. ${c.name} (${c.email})\n` +
          `   Region: ${c.region}\n` +
          `   Customer since: ${c.created_at}\n` +
          `   Status: ${c.status}`
        ).join('\n\n')
      }`;

      return {
        content: [
          {
            type: 'text',
            text: responseText,
          } as TextContent,
        ],
      };
    } catch (error) {
      return {
        content: [
          {
            type: 'text',
            text: `Database error: ${error.message}`,
          } as TextContent,
        ],
      };
    }
  }

  async run(): Promise {
    const transport = new StdioServerTransport();
    await this.server.connect(transport);
    console.error('Database MCP Server running on stdio');
  }
}

const server = new DatabaseMCPServer();
server.run().catch(console.error);

Step 4: Implementing Resource Providers

Beyond tools, MCP servers can expose resources—static or dynamic data that Claude can access. Understanding how to implement resource providers is essential when learning how to create Claude MCP server systems that provide comprehensive data access. Resources are particularly useful for scenarios like providing access to documentation, configuration files, or cached data.

from mcp.server import Server
from mcp.types import Resource, TextContent
import asyncio

class DocumentationMCPServer:
    def __init__(self):
        self.server = Server("docs-mcp-server")
        self.setup_resources()
    
    def setup_resources(self):
        # Register resource provider
        self.server.add_resource_provider(
            "docs://",
            self.list_documentation,
            self.get_documentation
        )
    
    async def list_documentation(self) -> list[Resource]:
        """List all available documentation resources"""
        return [
            Resource(
                uri="docs://api/authentication",
                name="Authentication Guide",
                description="How to authenticate API requests",
                mimeType="text/markdown"
            ),
            Resource(
                uri="docs://api/rate-limits",
                name="Rate Limiting Guide",
                description="Understanding API rate limits",
                mimeType="text/markdown"
            ),
            Resource(
                uri="docs://deployment/production",
                name="Production Deployment",
                description="Best practices for production deployment",
                mimeType="text/markdown"
            )
        ]
    
    async def get_documentation(self, uri: str) -> str:
        """Retrieve specific documentation content"""
        docs_content = {
            "docs://api/authentication": """
# Authentication Guide

## Overview
All API requests require authentication using API keys.

## Getting Your API Key
1. Log into your dashboard
2. Navigate to Settings > API Keys
3. Click "Generate New Key"

## Using Your API Key
Include the key in request headers:

Authorization: Bearer YOUR_API_KEY
## Security Best Practices
- Never commit API keys to version control
- Rotate keys regularly
- Use environment variables
- Implement key-specific permissions
            """,
            "docs://api/rate-limits": """
# Rate Limiting Guide

## Rate Limits by Plan
- Free Tier: 100 requests/hour
- Professional: 1,000 requests/hour
- Enterprise: Custom limits

## Handling Rate Limits
When you exceed limits, the API returns HTTP 429.
The response includes `Retry-After` header indicating wait time.

## Best Practices for Indian Developers
Consider time zone differences when implementing retry logic.
Peak hours (9 AM - 6 PM IST) may have higher latency.
            """,
            "docs://deployment/production": """
# Production Deployment Best Practices

## Infrastructure Recommendations
- Use load balancers for high availability
- Implement health check endpoints
- Configure auto-scaling based on traffic

## Regional Considerations
- For Indian markets: Deploy in Mumbai or Bangalore regions
- For global apps: Use CDN for static assets
- Consider data residency requirements

## Monitoring and Logging
- Implement structured logging
- Use APM tools for performance monitoring
- Set up alerts for critical failures
            """
        }
        
        return docs_content.get(uri, "Documentation not found")
    
    async def run(self):
        async with self.server:
            await self.server.run()

if __name__ == "__main__":
    server = DocumentationMCPServer()
    asyncio.run(server.run())

Step 5: Implementing Authentication and Security

Security is paramount when you create Claude MCP server implementations that handle sensitive data. This example shows how to implement basic authentication and authorization mechanisms that ensure only authorized clients can access your MCP server.

import jwt
from mcp.server import Server
from mcp.types import Tool, TextContent
from datetime import datetime, timedelta
import asyncio
import os

class SecureMCPServer:
    def __init__(self):
        self.server = Server("secure-mcp-server")
        self.secret_key = os.getenv("JWT_SECRET_KEY", "your-secret-key-change-in-production")
        self.authorized_clients = set()
        self.setup_authentication()
        self.setup_tools()
    
    def verify_token(self, token: str) -> dict:
        """Verify JWT token and return payload"""
        try:
            payload = jwt.decode(token, self.secret_key, algorithms=["HS256"])
            return payload
        except jwt.ExpiredSignatureError:
            raise ValueError("Token has expired")
        except jwt.InvalidTokenError:
            raise ValueError("Invalid token")
    
    def setup_authentication(self):
        """Setup authentication handler"""
        @self.server.authentication_handler
        async def authenticate(credentials: dict) -> bool:
            token = credentials.get("token")
            if not token:
                return False
            
            try:
                payload = self.verify_token(token)
                client_id = payload.get("client_id")
                self.authorized_clients.add(client_id)
                return True
            except ValueError:
                return False
    
    def setup_tools(self):
        """Setup protected tools"""
        sensitive_tool = Tool(
            name="access_sensitive_data",
            description="Access sensitive business data (requires authentication)",
            inputSchema={
                "type": "object",
                "properties": {
                    "data_type": {
                        "type": "string",
                        "enum": ["financial", "customer", "analytics"],
                        "description": "Type of sensitive data to access"
                    }
                },
                "required": ["data_type"]
            }
        )
        
        self.server.add_tool(sensitive_tool, self.handle_sensitive_data)
    
    async def handle_sensitive_data(self, arguments: dict) -> list[TextContent]:
        """Handle requests for sensitive data with authorization check"""
        data_type = arguments.get("data_type")
        
        # In production, implement granular permission checks here
        # Check if client has permission for specific data types
        
        mock_data = {
            "financial": "Q4 Revenue: $2.5M (restricted data)",
            "customer": "Active Customers: 15,234 (restricted data)",
            "analytics": "Conversion Rate: 3.2% (restricted data)"
        }
        
        response = f"Sensitive Data Access - {data_type.title()}:\n{mock_data.get(data_type)}"
        
        return [TextContent(type="text", text=response)]
    
    async def run(self):
        async with self.server:
            await self.server.run()

# Example of generating tokens for clients
def generate_client_token(client_id: str, secret_key: str) -> str:
    """Generate JWT token for a client"""
    payload = {
        "client_id": client_id,
        "exp": datetime.utcnow() + timedelta(hours=24),
        "iat": datetime.utcnow()
    }
    return jwt.encode(payload, secret_key, algorithm="HS256")

if __name__ == "__main__":
    server = SecureMCPServer()
    asyncio.run(server.run())

Step 6: Testing Your MCP Server

Proper testing is crucial when you learn how to create Claude MCP server implementations. Here’s a Python script that demonstrates how to test your MCP server locally before deploying to production.

import asyncio
from mcp.client import Client
from mcp.transport import StdioTransport
import subprocess

async def test_mcp_server():
    """Test MCP server functionality"""
    
    # Start the server as a subprocess
    server_process = subprocess.Popen(
        ["python", "weather_server.py"],
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE
    )
    
    # Create client and connect to server
    transport = StdioTransport(server_process.stdout, server_process.stdin)
    client = Client("test-client")
    
    try:
        await client.connect(transport)
        
        # List available tools
        tools = await client.list_tools()
        print(f"Available tools: {[tool.name for tool in tools]}")
        
        # Test weather query
        result = await client.call_tool(
            "get_weather",
            {
                "location": "Mumbai",
                "units": "celsius"
            }
        )
        print(f"\nWeather Query Result:\n{result.content[0].text}")
        
        # Test with different location
        result = await client.call_tool(
            "get_weather",
            {
                "location": "San Francisco",
                "units": "fahrenheit"
            }
        )
        print(f"\nWeather Query Result:\n{result.content[0].text}")
        
        print("\n✅ All tests passed!")
        
    finally:
        await client.disconnect()
        server_process.terminate()
        server_process.wait()

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

Step 7: Deployment Strategies

Deploying your MCP server to production requires consideration of scalability, reliability, and monitoring. For developers deploying to cloud infrastructure in regions like AWS Mumbai, Google Cloud Singapore, or Azure regions serving European markets, here’s a Docker-based deployment configuration that works universally.

# Dockerfile for MCP Server
FROM python:3.11-slim

WORKDIR /app

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

# Copy application code
COPY . .

# Set environment variables
ENV PYTHONUNBUFFERED=1
ENV MCP_SERVER_PORT=8080

# Health check endpoint
HEALTHCHECK --interval=30s --timeout=3s --start-period=5s --retries=3 \
  CMD python -c "import requests; requests.get('http://localhost:8080/health')"

# Run the server
CMD ["python", "mcp_server.py"]

# docker-compose.yml for production deployment
version: '3.8'

services:
  mcp-server:
    build: .
    ports:
      - "8080:8080"
    environment:
      - DB_HOST=postgres
      - DB_NAME=production_db
      - DB_USER=${DB_USER}
      - DB_PASSWORD=${DB_PASSWORD}
      - JWT_SECRET_KEY=${JWT_SECRET}
    depends_on:
      - postgres
      - redis
    restart: unless-stopped
    deploy:
      replicas: 3
      resources:
        limits:
          cpus: '2'
          memory: 2G
        reservations:
          cpus: '1'
          memory: 1G
  
  postgres:
    image: postgres:15
    environment:
      - POSTGRES_DB=production_db
      - POSTGRES_USER=${DB_USER}
      - POSTGRES_PASSWORD=${DB_PASSWORD}
    volumes:
      - postgres_data:/var/lib/postgresql/data
    restart: unless-stopped
  
  redis:
    image: redis:7-alpine
    restart: unless-stopped
  
  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf
      - ./ssl:/etc/nginx/ssl
    depends_on:
      - mcp-server
    restart: unless-stopped

volumes:
  postgres_data:

This deployment configuration provides high availability, load balancing, and easy scaling—essential features whether you’re serving customers in Bangalore’s tech parks or supporting global enterprise clients. For more advanced deployment patterns and examples, refer to the official MCP servers repository.