Build Your Own MCP Server

mkdir my-mcp-server
cd my-mcp-server

# Create files
touch server.py requirements.txt README.md

# Project structure
my-mcp-server/
├── server.py
├── requirements.txt
├── tools/
│   ├── __init__.py
│   └── customer_tools.py
└── README.md

fastapi==0.104.1
uvicorn==0.24.0
pydantic==2.5.0
httpx==0.25.1
python-dotenv==1.0.0

pip install -r requirements.txt

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
from typing import Optional, Dict, Any, List
import os

app = FastAPI(title="My Custom MCP Server")

# Tool definitions
TOOLS = [
    {
        "name": "get_customer_data",
        "description": "Retrieve customer information by ID",
        "parameters": {
            "type": "object",
            "properties": {
                "customer_id": {
                    "type": "string",
                    "description": "Customer ID (e.g., CUST-12345)"
                }
            },
            "required": ["customer_id"]
        }
    },
    {
        "name": "calculate_discount",
        "description": "Calculate discount based on customer tier",
        "parameters": {
            "type": "object",
            "properties": {
                "amount": {"type": "number"},
                "tier": {"type": "string", "enum": ["basic", "premium", "enterprise"]}
            },
            "required": ["amount", "tier"]
        }
    }
]

class MCPRequest(BaseModel):
    method: str
    params: Optional[Dict[str, Any]] = {}

@app.post("/")
async def handle_mcp_request(request: MCPRequest):
    """Main MCP endpoint"""
    if request.method == "list_tools":
        return {"tools": TOOLS}

    elif request.method == "call_tool":
        tool_name = request.params.get("tool_name")
        arguments = request.params.get("arguments", {})

        if tool_name == "get_customer_data":
            return await get_customer_data(arguments["customer_id"])

        elif tool_name == "calculate_discount":
            return await calculate_discount(
                arguments["amount"],
                arguments["tier"]
            )

        else:
            raise HTTPException(status_code=404, detail=f"Tool {tool_name} not found")

    else:
        raise HTTPException(status_code=400, detail=f"Unknown method: {request.method}")

async def get_customer_data(customer_id: str):
    """Tool implementation: Get customer data"""
    # TODO: Replace with your actual API call
    mock_data = {
        "CUST-12345": {
            "customer_id": "CUST-12345",
            "name": "Acme Corp",
            "tier": "enterprise",
            "mrr": 5000,
            "industry": "Technology"
        }
    }

    if customer_id in mock_data:
        return {"result": mock_data[customer_id]}
    else:
        raise HTTPException(status_code=404, detail="Customer not found")

async def calculate_discount(amount: float, tier: str):
    """Tool implementation: Calculate discount"""
    discount_rates = {
        "basic": 0.05,
        "premium": 0.10,
        "enterprise": 0.20
    }

    discount = amount * discount_rates.get(tier, 0)
    final_amount = amount - discount

    return {
        "result": {
            "original_amount": amount,
            "discount_rate": discount_rates[tier],
            "discount": discount,
            "final_amount": final_amount
        }
    }

@app.get("/health")
async def health_check():
    """Health check endpoint"""
    return {"status": "healthy", "tools_count": len(TOOLS)}

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=8766)

python server.py

# List tools
curl -X POST http://localhost:8766/ \
  -H "Content-Type: application/json" \
  -d '{"method": "list_tools", "params": {}}'

# Call a tool
curl -X POST http://localhost:8766/ \
  -H "Content-Type: application/json" \
  -d '{
    "method": "call_tool",
    "params": {
      "tool_name": "get_customer_data",
      "arguments": {"customer_id": "CUST-12345"}
    }
  }'

mkdir my-mcp-server-js
cd my-mcp-server-js
npm init -y
npm install express body-parser axios dotenv

const express = require('express');
const bodyParser = require('body-parser');
require('dotenv').config();

const app = express();
app.use(bodyParser.json());

// Tool definitions
const TOOLS = [
  {
    name: 'get_product_info',
    description: 'Retrieve product information by SKU',
    parameters: {
      type: 'object',
      properties: {
        sku: {
          type: 'string',
          description: 'Product SKU code'
        }
      },
      required: ['sku']
    }
  }
];

// Main MCP endpoint
app.post('/', async (req, res) => {
  const { method, params } = req.body;

  try {
    if (method === 'list_tools') {
      res.json({ tools: TOOLS });
    } else if (method === 'call_tool') {
      const { tool_name, arguments: args } = params;

      if (tool_name === 'get_product_info') {
        const result = await getProductInfo(args.sku);
        res.json({ result });
      } else {
        res.status(404).json({ error: `Tool ${tool_name} not found` });
      }
    } else {
      res.status(400).json({ error: `Unknown method: ${method}` });
    }
  } catch (error) {
    res.status(500).json({ error: error.message });
  }
});

// Tool implementation
async function getProductInfo(sku) {
  // TODO: Replace with your actual API call
  const mockData = {
    'SKU-001': {
      sku: 'SKU-001',
      name: 'Premium Widget',
      price: 99.99,
      stock: 150,
      category: 'Electronics'
    }
  };

  if (mockData[sku]) {
    return mockData[sku];
  } else {
    throw new Error('Product not found');
  }
}

// Health check
app.get('/health', (req, res) => {
  res.json({ status: 'healthy', tools_count: TOOLS.length });
});

const PORT = process.env.PORT || 8766;
app.listen(PORT, () => {
  console.log(`MCP Server running on port ${PORT}`);
});