MCP Server Generator

Comprehensive system for creating high-quality Model Context Protocol (MCP) servers that integrate external APIs, databases, and services with Claude.

Core Concepts

What is MCP?

Model Context Protocol (MCP) is an open standard that enables AI applications to connect to external data sources and tools. MCP servers provide:

• Resources: File-like data (documents, logs, database schemas)

• Tools: Executable functions (API calls, calculations, searches)

• Prompts: Reusable prompt templates

• Sampling: LLM completion requests

MCP Architecture

┌─────────────┐ ┌─────────────┐ ┌──────────────┐ │ Claude │ ◄─────► │ MCP Server │ ◄─────► │ External API │ │ (Client) │ MCP │ (Bridge) │ │ or Service │ └─────────────┘ └─────────────┘ └──────────────┘

Implementation Patterns

1. TypeScript/Node.js MCP Server

Basic Structure:

#!/usr/bin/env node import { Server } from "@modelcontextprotocol/sdk/server/index.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; import { CallToolRequestSchema, ListToolsRequestSchema, ListResourcesRequestSchema, ReadResourceRequestSchema, } from "@modelcontextprotocol/sdk/types.js";

// Initialize server const server = new Server( { name: "my-mcp-server", version: "1.0.0", }, { capabilities: { tools: {}, resources: {}, prompts: {}, }, } );

// List available tools server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: [ { name: "example_tool", description: "An example tool that does something useful", inputSchema: { type: "object", properties: { query: { type: "string", description: "The query parameter", }, }, required: ["query"], }, }, ], }; });

// Handle tool calls server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

if (name === "example_tool") { const result = await performAction(args.query); return { content: [ { type: "text", text: JSON.stringify(result, null, 2), }, ], }; }

throw new Error(Unknown tool: ${name}); });

// List resources server.setRequestHandler(ListResourcesRequestSchema, async () => { return { resources: [ { uri: "example://resource/1", name: "Example Resource", description: "An example resource", mimeType: "application/json", }, ], }; });

// Read resource server.setRequestHandler(ReadResourceRequestSchema, async (request) => { const { uri } = request.params;

if (uri === "example://resource/1") { return { contents: [ { uri, mimeType: "application/json", text: JSON.stringify({ data: "example" }, null, 2), }, ], }; }

throw new Error(Resource not found: ${uri}); });

// Start server async function main() { const transport = new StdioServerTransport(); await server.connect(transport); console.error("MCP Server running on stdio"); }

main().catch(console.error);

package.json:

{ "name": "my-mcp-server", "version": "1.0.0", "type": "module", "bin": { "my-mcp-server": "./dist/index.js" }, "scripts": { "build": "tsc && chmod +x dist/index.js", "dev": "tsc --watch", "prepare": "npm run build" }, "dependencies": { "@modelcontextprotocol/sdk": "^0.5.0" }, "devDependencies": { "@types/node": "^20.0.0", "typescript": "^5.3.0" } }

tsconfig.json:

{ "compilerOptions": { "target": "ES2022", "module": "Node16", "moduleResolution": "Node16", "outDir": "./dist", "rootDir": "./src", "strict": true, "esModuleInterop": true, "skipLibCheck": true, "forceConsistentCasingInFileNames": true, "resolveJsonModule": true, "declaration": true }, "include": ["src/**/*"], "exclude": ["node_modules", "dist"] }

2. Python MCP Server

Basic Structure:

#!/usr/bin/env python3 import asyncio import logging from typing import Any

from mcp.server import Server from mcp.server.stdio import stdio_server from mcp.types import ( Tool, TextContent, Resource, INTERNAL_ERROR, )

Configure logging

logging.basicConfig(level=logging.INFO) logger = logging.getLogger(name)

Initialize server

app = Server("my-mcp-server")

@app.list_tools() async def list_tools() -> list[Tool]: """List available tools.""" return [ Tool( name="example_tool", description="An example tool that does something useful", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "The query parameter", }, }, "required": ["query"], }, ), ]

@app.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: """Handle tool calls.""" if name == "example_tool": query = arguments.get("query") result = await perform_action(query)

    return [
        TextContent(
            type="text",
            text=str(result),
        )
    ]

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

@app.list_resources() async def list_resources() -> list[Resource]: """List available resources.""" return [ Resource( uri="example://resource/1", name="Example Resource", description="An example resource", mimeType="application/json", ), ]

@app.read_resource() async def read_resource(uri: str) -> str: """Read resource content.""" if uri == "example://resource/1": return '{"data": "example"}'

raise ValueError(f"Resource not found: {uri}")

async def perform_action(query: str) -> dict: """Perform the actual action.""" # Your implementation here return {"query": query, "result": "success"}

async def main(): """Run the MCP server.""" async with stdio_server() as (read_stream, write_stream): await app.run( read_stream, write_stream, app.create_initialization_options() )

if name == "main": asyncio.run(main())

pyproject.toml:

[project] name = "my-mcp-server" version = "1.0.0" description = "MCP server for..." requires-python = ">=3.10" dependencies = [ "mcp>=0.9.0", ]

[project.scripts] my-mcp-server = "my_mcp_server:main"

[build-system] requires = ["hatchling"] build-backend = "hatchling.build"

3. Common MCP Server Patterns

API Integration Server

// GitHub API MCP Server Example import { Octokit } from "@octokit/rest";

const octokit = new Octokit({ auth: process.env.GITHUB_TOKEN });

server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: [ { name: "search_repositories", description: "Search GitHub repositories", inputSchema: { type: "object", properties: { query: { type: "string", description: "Search query" }, language: { type: "string", description: "Filter by language" }, limit: { type: "number", description: "Max results", default: 10 }, }, required: ["query"], }, }, { name: "get_repository", description: "Get repository details", inputSchema: { type: "object", properties: { owner: { type: "string", description: "Repository owner" }, repo: { type: "string", description: "Repository name" }, }, required: ["owner", "repo"], }, }, { name: "create_issue", description: "Create a new issue", inputSchema: { type: "object", properties: { owner: { type: "string" }, repo: { type: "string" }, title: { type: "string" }, body: { type: "string" }, labels: { type: "array", items: { type: "string" } }, }, required: ["owner", "repo", "title"], }, }, ], }; });

server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

switch (name) { case "search_repositories": { const result = await octokit.search.repos({ q: ${args.query}${args.language ? language:${args.language} : ""}, per_page: args.limit || 10, }); return { content: [ { type: "text", text: JSON.stringify( result.data.items.map((item) => ({ name: item.full_name, description: item.description, stars: item.stargazers_count, url: item.html_url, })), null, 2 ), }, ], }; }

case "get_repository": {
  const result = await octokit.repos.get({
    owner: args.owner,
    repo: args.repo,
  });
  return {
    content: [
      { type: "text", text: JSON.stringify(result.data, null, 2) },
    ],
  };
}

case "create_issue": {
  const result = await octokit.issues.create({
    owner: args.owner,
    repo: args.repo,
    title: args.title,
    body: args.body,
    labels: args.labels,
  });
  return {
    content: [
      { type: "text", text: `Issue created: ${result.data.html_url}` },
    ],
  };
}

default:
  throw new Error(`Unknown tool: ${name}`);

} });

Database Integration Server

PostgreSQL MCP Server Example

import asyncpg from mcp.server import Server from mcp.types import Tool, TextContent

app = Server("postgres-mcp-server")

Connection pool

pool = None

async def init_db(): global pool pool = await asyncpg.create_pool( host=os.getenv("DB_HOST", "localhost"), port=int(os.getenv("DB_PORT", "5432")), database=os.getenv("DB_NAME"), user=os.getenv("DB_USER"), password=os.getenv("DB_PASSWORD"), )

@app.list_tools() async def list_tools() -> list[Tool]: return [ Tool( name="query_database", description="Execute a read-only SQL query", inputSchema={ "type": "object", "properties": { "query": { "type": "string", "description": "SQL SELECT query to execute", }, }, "required": ["query"], }, ), Tool( name="get_schema", description="Get database schema information", inputSchema={ "type": "object", "properties": { "table_name": { "type": "string", "description": "Optional table name to filter", }, }, }, ), ]

@app.call_tool() async def call_tool(name: str, arguments: Any) -> list[TextContent]: if name == "query_database": query = arguments.get("query", "")

    # Security: Only allow SELECT queries
    if not query.strip().upper().startswith("SELECT"):
        raise ValueError("Only SELECT queries are allowed")

    async with pool.acquire() as conn:
        results = await conn.fetch(query)
        return [
            TextContent(
                type="text",
                text=json.dumps([dict(row) for row in results], indent=2),
            )
        ]

elif name == "get_schema":
    table_filter = arguments.get("table_name")

    query = """
        SELECT table_name, column_name, data_type, is_nullable
        FROM information_schema.columns
        WHERE table_schema = 'public'
    """

    if table_filter:
        query += f" AND table_name = '{table_filter}'"

    async with pool.acquire() as conn:
        results = await conn.fetch(query)
        return [
            TextContent(
                type="text",
                text=json.dumps([dict(row) for row in results], indent=2),
            )
        ]

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

File System Server

// File System MCP Server import fs from "fs/promises"; import path from "path";

const ALLOWED_DIRECTORIES = [ process.env.DOCS_PATH || "./docs", process.env.DATA_PATH || "./data", ];

function isPathAllowed(filePath: string): boolean { const absPath = path.resolve(filePath); return ALLOWED_DIRECTORIES.some((dir) => absPath.startsWith(path.resolve(dir)) ); }

server.setRequestHandler(ListResourcesRequestSchema, async () => { const resources = [];

for (const dir of ALLOWED_DIRECTORIES) { const files = await fs.readdir(dir, { recursive: true }); for (const file of files) { const fullPath = path.join(dir, file); const stats = await fs.stat(fullPath);

  if (stats.isFile()) {
    resources.push({
      uri: `file://${fullPath}`,
      name: file,
      description: `File at ${fullPath}`,
      mimeType: getMimeType(fullPath),
    });
  }
}

}

return { resources }; });

server.setRequestHandler(ReadResourceRequestSchema, async (request) => { const { uri } = request.params; const filePath = uri.replace("file://", "");

if (!isPathAllowed(filePath)) { throw new Error("Access denied"); }

const content = await fs.readFile(filePath, "utf-8");

return { contents: [ { uri, mimeType: getMimeType(filePath), text: content, }, ], }; });

server.setRequestHandler(ListToolsRequestSchema, async () => { return { tools: [ { name: "search_files", description: "Search for files by name or content", inputSchema: { type: "object", properties: { query: { type: "string", description: "Search query" }, type: { type: "string", enum: ["name", "content"], description: "Search by filename or content", }, }, required: ["query", "type"], }, }, { name: "write_file", description: "Write content to a file", inputSchema: { type: "object", properties: { path: { type: "string", description: "File path" }, content: { type: "string", description: "File content" }, }, required: ["path", "content"], }, }, ], }; });

4. Configuration & Deployment

Claude Desktop Configuration:

On macOS: ~/Library/Application Support/Claude/claude_desktop_config.json

On Windows: %APPDATA%\Claude\claude_desktop_config.json

{ "mcpServers": { "my-mcp-server": { "command": "node", "args": ["/path/to/my-mcp-server/dist/index.js"], "env": { "API_KEY": "your-api-key", "LOG_LEVEL": "info" } }, "python-mcp-server": { "command": "python", "args": ["-m", "my_mcp_server"], "env": { "DATABASE_URL": "postgresql://localhost/mydb" } }, "npx-server": { "command": "npx", "args": ["-y", "@modelcontextprotocol/server-github"], "env": { "GITHUB_PERSONAL_ACCESS_TOKEN": "ghp_..." } } } }

Environment Variables Best Practices:

// Load from .env file import dotenv from "dotenv"; dotenv.config();

// Validate required env vars const requiredEnvVars = ["API_KEY", "DATABASE_URL"]; for (const varName of requiredEnvVars) { if (!process.env[varName]) { throw new Error(Missing required environment variable: ${varName}); } }

// Use with defaults const config = { apiKey: process.env.API_KEY, apiUrl: process.env.API_URL || "https://api.example.com", timeout: parseInt(process.env.TIMEOUT || "30000"), maxRetries: parseInt(process.env.MAX_RETRIES || "3"), };

5. Error Handling & Logging

Comprehensive Error Handling:

import { McpError, ErrorCode } from "@modelcontextprotocol/sdk/types.js";

server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

try { // Validate arguments if (!args.query) { throw new McpError( ErrorCode.InvalidParams, "Missing required parameter: query" ); }

// Perform action
const result = await performAction(args.query);

return {
  content: [
    {
      type: "text",
      text: JSON.stringify(result, null, 2),
    },
  ],
};

} catch (error) { // Log error console.error(Error in tool ${name}:, error);

// Return user-friendly error
if (error instanceof McpError) {
  throw error;
}

throw new McpError(
  ErrorCode.InternalError,
  `Failed to execute tool: ${error.message}`
);

} });

Structured Logging:

import logging import json from datetime import datetime

Configure JSON logging

class JSONFormatter(logging.Formatter): def format(self, record): log_data = { "timestamp": datetime.utcnow().isoformat(), "level": record.levelname, "message": record.getMessage(), "module": record.module, "function": record.funcName, } if record.exc_info: log_data["exception"] = self.formatException(record.exc_info) return json.dumps(log_data)

handler = logging.StreamHandler() handler.setFormatter(JSONFormatter()) logger = logging.getLogger(name) logger.addHandler(handler) logger.setLevel(logging.INFO)

Use in code

logger.info("Tool called", extra={"tool_name": name, "args": arguments}) logger.error("API request failed", extra={"status_code": 500, "url": api_url})

6. Testing Strategies

Unit Testing:

import { describe, it, expect, beforeAll, afterAll } from "vitest"; import { Client } from "@modelcontextprotocol/sdk/client/index.js"; import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js";

describe("MCP Server", () => { let client: Client; let transport: StdioClientTransport;

beforeAll(async () => { transport = new StdioClientTransport({ command: "node", args: ["./dist/index.js"], });

client = new Client(
  {
    name: "test-client",
    version: "1.0.0",
  },
  {
    capabilities: {},
  }
);

await client.connect(transport);

});

afterAll(async () => { await client.close(); });

it("should list tools", async () => { const response = await client.listTools(); expect(response.tools).toHaveLength(3); expect(response.tools[0].name).toBe("example_tool"); });

it("should call tool successfully", async () => { const response = await client.callTool({ name: "example_tool", arguments: { query: "test" }, }); expect(response.content).toBeDefined(); });

it("should handle errors gracefully", async () => { await expect( client.callTool({ name: "non_existent_tool", arguments: {}, }) ).rejects.toThrow(); }); });

Integration Testing:

Test server manually

node dist/index.js & SERVER_PID=$!

Send test requests

echo '{"jsonrpc":"2.0","id":1,"method":"tools/list"}' | node dist/index.js

Cleanup

kill $SERVER_PID

7. Performance Optimization

Caching:

import NodeCache from "node-cache";

const cache = new NodeCache({ stdTTL: 600 }); // 10 minutes

server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

if (name === "expensive_operation") { const cacheKey = ${name}:${JSON.stringify(args)};

// Check cache
const cached = cache.get(cacheKey);
if (cached) {
  return {
    content: [{ type: "text", text: JSON.stringify(cached) }],
  };
}

// Perform operation
const result = await expensiveOperation(args);

// Store in cache
cache.set(cacheKey, result);

return {
  content: [{ type: "text", text: JSON.stringify(result) }],
};

} });

Rate Limiting:

import rateLimit from "express-rate-limit";

const limiter = new Map();

function checkRateLimit(key: string): boolean { const now = Date.now(); const record = limiter.get(key) || { count: 0, resetAt: now + 60000 };

if (now > record.resetAt) { record.count = 0; record.resetAt = now + 60000; }

record.count++; limiter.set(key, record);

return record.count <= 100; // 100 requests per minute }

server.setRequestHandler(CallToolRequestSchema, async (request) => { if (!checkRateLimit("global")) { throw new McpError(ErrorCode.InternalError, "Rate limit exceeded"); }

// Handle request... });

8. Security Best Practices

Input Validation:

import { z } from "zod";

const toolInputSchemas = { search_query: z.object({ query: z.string().min(1).max(500), limit: z.number().int().min(1).max(100).optional(), filters: z.record(z.string()).optional(), }), };

server.setRequestHandler(CallToolRequestSchema, async (request) => { const { name, arguments: args } = request.params;

// Validate input const schema = toolInputSchemas[name]; if (!schema) { throw new McpError(ErrorCode.InvalidParams, Unknown tool: ${name}); }

try { const validatedArgs = schema.parse(args); // Use validatedArgs... } catch (error) { throw new McpError( ErrorCode.InvalidParams, Invalid arguments: ${error.message} ); } });

Access Control:

function checkPermissions(userId: string, action: string): boolean { const permissions = { admin: ["read", "write", "delete"], user: ["read"], guest: [], };

const userRole = getUserRole(userId); return permissions[userRole]?.includes(action) || false; }

server.setRequestHandler(CallToolRequestSchema, async (request) => { const userId = request.params._meta?.userId; const action = getActionFromTool(request.params.name);

if (!checkPermissions(userId, action)) { throw new McpError(ErrorCode.InvalidParams, "Permission denied"); }

// Proceed with request... });

9. Publishing & Distribution

NPM Package:

{ "name": "@company/mcp-server-example", "version": "1.0.0", "description": "MCP server for...", "main": "dist/index.js", "bin": { "mcp-server-example": "dist/index.js" }, "files": ["dist", "README.md", "LICENSE"], "keywords": ["mcp", "claude", "mcp-server"], "repository": { "type": "git", "url": "https://github.com/company/mcp-server-example" }, "publishConfig": { "access": "public" } }

Docker Distribution:

FROM node:18-alpine WORKDIR /app COPY package*.json ./ RUN npm ci --only=production COPY dist ./dist USER node CMD ["node", "dist/index.js"]

10. Documentation Template

README.md:

My MCP Server

[Brief description of what this server does]

Features

• Feature 1
• Feature 2
• Feature 3

Installation

```bash npm install -g my-mcp-server ```

Configuration

Add to your Claude Desktop config:

```json { "mcpServers": { "my-server": { "command": "my-mcp-server", "env": { "API_KEY": "your-api-key" } } } } ```

Environment Variables

Available Tools

tool_name

Description of what this tool does.

Parameters:

• param1 (string, required): Description
• param2 (number, optional): Description

Example: ``` Use the tool_name tool with parameter "value" ```

Available Resources

resource://example/1

Description of this resource.

Development

```bash git clone https://github.com/company/my-mcp-server cd my-mcp-server npm install npm run build npm run dev ```

License

MIT

Quick Start Checklist

• Choose implementation language (TypeScript/Python)

• Set up project structure with proper tooling

• Implement tools with clear input schemas

• Add comprehensive error handling

• Implement caching for expensive operations

• Add input validation and sanitization

• Set up structured logging

• Write unit and integration tests

• Create detailed documentation

• Configure environment variables properly

• Add rate limiting if needed

• Implement access control if needed

• Test with Claude Desktop

• Publish to npm/PyPI or Docker

Common Use Cases

• API Integration: GitHub, Jira, Slack, Linear, Notion

• Database Access: PostgreSQL, MongoDB, Redis, Elasticsearch

• File Systems: Local files, S3, Google Drive, Dropbox

• Developer Tools: Git operations, CI/CD, deployment

• Business Tools: CRM, ERP, analytics platforms

• AI/ML: Vector databases, model APIs, embeddings

• Communication: Email, SMS, webhooks, notifications

• Monitoring: Logs, metrics, alerts, dashboards

When to Use This Skill:

Invoke when building new MCP servers, troubleshooting existing servers, or integrating external services with Claude.