majiayu000 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
| Variable | Required | Description |
|----------|----------|-------------|
| API_KEY | Yes | Your API key |
| API_URL | No | API endpoint (default: https://api.example.com) |
## 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.