██████╗ ██╗ ██╗ ██████╗██╗ █████╗ ██████╗██╗ ██████╗ ██╗ ██╗██████╗
██╔══██╗██║ ██║██╔════╝██║ ██╔══██╗ ██╔════╝██║ ██╔═══██╗██║ ██║██╔══██╗
██║ ██║██║ ██║██║ ██║ ███████║ ██║ ██║ ██║ ██║██║ ██║██║ ██║
██║ ██║██║ ██║██║ ██║ ██╔══██║ ██║ ██║ ██║ ██║██║ ██║██║ ██║
██████╔╝╚██████╔╝╚██████╗███████╗██║ ██║ ╚██████╗███████╗╚██████╔╝╚██████╔╝██████╔╝
╚═════╝ ╚═════╝ ╚═════╝╚══════╝╚═╝ ╚═╝ ╚═════╝╚══════╝ ╚═════╝ ╚═════╝ ╚═════╝
Phát triển MCP server tùy chỉnh cho phép bạn tạo ra các tools và resources phù hợp với nhu cầu cụ thể của dự án. Hướng dẫn này sẽ đưa bạn qua toàn bộ quá trình từ thiết kế đến triển khai.
┌─────────────────────────────────────────────────────────────┐
│ MCP Server Architecture │
├─────────────────────────────────────────────────────────────┤
│ ┌─────────────┐ ┌─────────────┐ ┌─────────────────────┐ │
│ │ Tools │ │ Resources │ │ Message Handler │ │
│ │ │ │ │ │ │ │
│ │ • list │ │ • list │ │ • Request routing │ │
│ │ • call │ │ • read │ │ • Error handling │ │
│ │ • schema │ │ • subscribe │ │ • Validation │ │
│ └─────────────┘ └─────────────┘ └─────────────────────┘ │
├─────────────────────────────────────────────────────────────┤
│ Transport Layer │
│ (stdio, HTTP, WebSocket) │
└─────────────────────────────────────────────────────────────┘
# Tạo project directory
mkdir my-mcp-server
cd my-mcp-server
# Tạo virtual environment
python3 -m venv venv
source venv/bin/activate
# Cài đặt dependencies
pip install mcp#!/usr/bin/env python3
"""
Custom MCP Server Example
Provides tools for data processing and file operations
"""
import asyncio
import json
import logging
from typing import Any, Dict, List, Optional
from pathlib import Path
from mcp import Server, types
from mcp.server.models import InitializationOptions
from mcp.server import NotificationOptions, ServerRequestContext
# Configure logging
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("custom-mcp-server")
class CustomMCPServer:
def __init__(self, name: str = "custom-server", version: str = "1.0.0"):
self.server = Server(name)
self.name = name
self.version = version
self.setup_handlers()
def setup_handlers(self):
"""Setup all request handlers"""
@self.server.list_tools()
async def handle_list_tools() -> List[types.Tool]:
"""Return list of available tools"""
return [
types.Tool(
name="process_data",
description="Process and analyze data from various sources",
inputSchema={
"type": "object",
"properties": {
"data": {
"type": "string",
"description": "Data to process"
},
"operation": {
"type": "string",
"enum": ["analyze", "transform", "validate"],
"description": "Operation to perform"
}
},
"required": ["data", "operation"]
}
),
types.Tool(
name="file_operations",
description="Perform file operations with safety checks",
inputSchema={
"type": "object",
"properties": {
"action": {
"type": "string",
"enum": ["read", "write", "delete", "list"],
"description": "File operation to perform"
},
"path": {
"type": "string",
"description": "File or directory path"
},
"content": {
"type": "string",
"description": "Content for write operations"
}
},
"required": ["action", "path"]
}
),
types.Tool(
name="system_info",
description="Get system information and metrics",
inputSchema={
"type": "object",
"properties": {
"info_type": {
"type": "string",
"enum": ["cpu", "memory", "disk", "network"],
"description": "Type of system information"
}
},
"required": ["info_type"]
}
)
]
@self.server.call_tool()
async def handle_call_tool(
name: str, arguments: Dict[str, Any] | None
) -> List[types.TextContent]:
"""Handle tool calls"""
if arguments is None:
arguments = {}
try:
if name == "process_data":
return await self._process_data(arguments)
elif name == "file_operations":
return await self._file_operations(arguments)
elif name == "system_info":
return await self._system_info(arguments)
else:
raise ValueError(f"Unknown tool: {name}")
except Exception as e:
logger.error(f"Error in tool {name}: {str(e)}")
return [types.TextContent(
type="text",
text=f"Error: {str(e)}"
)]
async def _process_data(self, args: Dict[str, Any]) -> List[types.TextContent]:
"""Process data based on operation type"""
data = args.get("data", "")
operation = args.get("operation", "analyze")
if operation == "analyze":
# Perform data analysis
result = {
"length": len(data),
"words": len(data.split()),
"lines": len(data.splitlines()),
"characters": len(data.replace(" ", "")),
"type": "text_analysis"
}
elif operation == "transform":
# Transform data (example: uppercase)
result = {
"original": data,
"transformed": data.upper(),
"type": "text_transform"
}
elif operation == "validate":
# Validate data format
result = {
"is_json": self._is_valid_json(data),
"is_empty": len(data.strip()) == 0,
"encoding": "utf-8",
"type": "validation"
}
return [types.TextContent(
type="text",
text=f"Data processing result:\n{json.dumps(result, indent=2)}"
)]
async def _file_operations(self, args: Dict[str, Any]) -> List[types.TextContent]:
"""Handle file operations with safety checks"""
action = args.get("action")
path = Path(args.get("path", ""))
content = args.get("content", "")
# Safety check: only allow operations in safe directories
safe_dirs = ["/tmp", "/home/user/workspace", "/var/tmp"]
if not any(str(path).startswith(safe_dir) for safe_dir in safe_dirs):
return [types.TextContent(
type="text",
text=f"Error: Path {path} is not in allowed directories"
)]
try:
if action == "read":
if path.is_file():
content = path.read_text()
return [types.TextContent(
type="text",
text=f"File content:\n{content}"
)]
else:
return [types.TextContent(
type="text",
text=f"Error: {path} is not a file or doesn't exist"
)]
elif action == "write":
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(content)
return [types.TextContent(
type="text",
text=f"Successfully wrote to {path}"
)]
elif action == "delete":
if path.exists():
if path.is_file():
path.unlink()
else:
path.rmdir()
return [types.TextContent(
type="text",
text=f"Successfully deleted {path}"
)]
else:
return [types.TextContent(
type="text",
text=f"Error: {path} doesn't exist"
)]
elif action == "list":
if path.is_dir():
files = [str(p) for p in path.iterdir()]
return [types.TextContent(
type="text",
text=f"Directory contents:\n" + "\n".join(files)
)]
else:
return [types.TextContent(
type="text",
text=f"Error: {path} is not a directory"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"File operation error: {str(e)}"
)]
async def _system_info(self, args: Dict[str, Any]) -> List[types.TextContent]:
"""Get system information"""
import psutil
import platform
info_type = args.get("info_type", "cpu")
try:
if info_type == "cpu":
info = {
"cpu_percent": psutil.cpu_percent(interval=1),
"cpu_count": psutil.cpu_count(),
"cpu_freq": psutil.cpu_freq()._asdict() if psutil.cpu_freq() else None
}
elif info_type == "memory":
memory = psutil.virtual_memory()
info = {
"total": memory.total,
"available": memory.available,
"percent": memory.percent,
"used": memory.used
}
elif info_type == "disk":
disk = psutil.disk_usage('/')
info = {
"total": disk.total,
"used": disk.used,
"free": disk.free,
"percent": (disk.used / disk.total) * 100
}
elif info_type == "network":
info = {
"platform": platform.system(),
"hostname": platform.node(),
"architecture": platform.architecture()
}
return [types.TextContent(
type="text",
text=f"System {info_type} information:\n{json.dumps(info, indent=2)}"
)]
except Exception as e:
return [types.TextContent(
type="text",
text=f"System info error: {str(e)}"
)]
def _is_valid_json(self, data: str) -> bool:
"""Check if string is valid JSON"""
try:
json.loads(data)
return True
except json.JSONDecodeError:
return False
async def run(self):
"""Run the server"""
from mcp.server.stdio import stdio_server
async with stdio_server() as (read_stream, write_stream):
await self.server.run(
read_stream,
write_stream,
InitializationOptions(
server_name=self.name,
server_version=self.version
)
)
# Entry point
async def main():
server = CustomMCPServer()
await server.run()
if __name__ == "__main__":
asyncio.run(main())# requirements.txt
mcp>=1.0.0
psutil>=5.9.0
aiofiles>=23.0.0# setup.py
from setuptools import setup, find_packages
setup(
name="custom-mcp-server",
version="1.0.0",
description="Custom MCP Server for data processing and file operations",
packages=find_packages(),
install_requires=[
"mcp>=1.0.0",
"psutil>=5.9.0",
"aiofiles>=23.0.0"
],
entry_points={
"console_scripts": [
"custom-mcp-server=custom_mcp_server:main",
],
},
python_requires=">=3.8",
)# Tạo project
mkdir my-mcp-server-js
cd my-mcp-server-js
# Initialize npm project
npm init -y
# Install dependencies
npm install @modelcontextprotocol/sdk
npm install --save-dev @types/node typescript// src/server.ts
import { Server } from '@modelcontextprotocol/sdk/server/index.js';
import { StdioServerTransport } from '@modelcontextprotocol/sdk/server/stdio.js';
import {
CallToolRequestSchema,
ListToolsRequestSchema,
Tool,
TextContent,
} from '@modelcontextprotocol/sdk/types.js';
import * as fs from 'fs/promises';
import * as path from 'path';
import * as os from 'os';
interface ProcessDataArgs {
data: string;
operation: 'analyze' | 'transform' | 'validate';
}
interface FileOperationArgs {
action: 'read' | 'write' | 'delete' | 'list';
path: string;
content?: string;
}
interface SystemInfoArgs {
info_type: 'cpu' | 'memory' | 'disk' | 'network';
}
class CustomMCPServer {
private server: Server;
constructor() {
this.server = new Server(
{
name: 'custom-mcp-server-js',
version: '1.0.0',
},
{
capabilities: {
tools: {},
},
}
);
this.setupHandlers();
}
private setupHandlers(): void {
// List available tools
this.server.setRequestHandler(ListToolsRequestSchema, async () => {
return {
tools: [
{
name: 'process_data',
description: 'Process and analyze data from various sources',
inputSchema: {
type: 'object',
properties: {
data: {
type: 'string',
description: 'Data to process',
},
operation: {
type: 'string',
enum: ['analyze', 'transform', 'validate'],
description: 'Operation to perform',
},
},
required: ['data', 'operation'],
},
},
{
name: 'file_operations',
description: 'Perform file operations with safety checks',
inputSchema: {
type: 'object',
properties: {
action: {
type: 'string',
enum: ['read', 'write', 'delete', 'list'],
description: 'File operation to perform',
},
path: {
type: 'string',
description: 'File or directory path',
},
content: {
type: 'string',
description: 'Content for write operations',
},
},
required: ['action', 'path'],
},
},
{
name: 'system_info',
description: 'Get system information and metrics',
inputSchema: {
type: 'object',
properties: {
info_type: {
type: 'string',
enum: ['cpu', 'memory', 'disk', 'network'],
description: 'Type of system information',
},
},
required: ['info_type'],
},
},
] as Tool[],
};
});
// Handle tool calls
this.server.setRequestHandler(CallToolRequestSchema, async (request) => {
const { name, arguments: args } = request.params;
try {
switch (name) {
case 'process_data':
return await this.processData(args as ProcessDataArgs);
case 'file_operations':
return await this.fileOperations(args as FileOperationArgs);
case 'system_info':
return await this.systemInfo(args as SystemInfoArgs);
default:
throw new Error(`Unknown tool: ${name}`);
}
} catch (error) {
return {
content: [
{
type: 'text',
text: `Error: ${error instanceof Error ? error.message : String(error)}`,
} as TextContent,
],
};
}
});
}
private async processData(args: ProcessDataArgs) {
const { data, operation } = args;
let result: any;
switch (operation) {
case 'analyze':
result = {
length: data.length,
words: data.split(/\s+/).length,
lines: data.split('\n').length,
characters: data.replace(/\s/g, '').length,
type: 'text_analysis',
};
break;
case 'transform':
result = {
original: data,
transformed: data.toUpperCase(),
type: 'text_transform',
};
break;
case 'validate':
result = {
is_json: this.isValidJSON(data),
is_empty: data.trim().length === 0,
encoding: 'utf-8',
type: 'validation',
};
break;
}
return {
content: [
{
type: 'text',
text: `Data processing result:\n${JSON.stringify(result, null, 2)}`,
} as TextContent,
],
};
}
private async fileOperations(args: FileOperationArgs) {
const { action, path: filePath, content = '' } = args;
// Safety check
const safeDirs = ['/tmp', '/home/user/workspace', '/var/tmp'];
if (!safeDirs.some(dir => filePath.startsWith(dir))) {
return {
content: [
{
type: 'text',
text: `Error: Path ${filePath} is not in allowed directories`,
} as TextContent,
],
};
}
try {
switch (action) {
case 'read':
const fileContent = await fs.readFile(filePath, 'utf-8');
return {
content: [
{
type: 'text',
text: `File content:\n${fileContent}`,
} as TextContent,
],
};
case 'write':
await fs.mkdir(path.dirname(filePath), { recursive: true });
await fs.writeFile(filePath, content);
return {
content: [
{
type: 'text',
text: `Successfully wrote to ${filePath}`,
} as TextContent,
],
};
case 'delete':
await fs.unlink(filePath);
return {
content: [
{
type: 'text',
text: `Successfully deleted ${filePath}`,
} as TextContent,
],
};
case 'list':
const files = await fs.readdir(filePath);
return {
content: [
{
type: 'text',
text: `Directory contents:\n${files.join('\n')}`,
} as TextContent,
],
};
default:
throw new Error(`Unknown action: ${action}`);
}
} catch (error) {
return {
content: [
{
type: 'text',
text: `File operation error: ${error instanceof Error ? error.message : String(error)}`,
} as TextContent,
],
};
}
}
private async systemInfo(args: SystemInfoArgs) {
const { info_type } = args;
let info: any;
try {
switch (info_type) {
case 'cpu':
info = {
platform: os.platform(),
arch: os.arch(),
cpus: os.cpus().length,
loadavg: os.loadavg(),
};
break;
case 'memory':
info = {
total: os.totalmem(),
free: os.freemem(),
used: os.totalmem() - os.freemem(),
percent: ((os.totalmem() - os.freemem()) / os.totalmem()) * 100,
};
break;
case 'disk':
// Note: Node.js doesn't have built-in disk usage, would need additional package
info = {
message: 'Disk usage requires additional package like "diskusage"',
};
break;
case 'network':
info = {
hostname: os.hostname(),
platform: os.platform(),
networkInterfaces: Object.keys(os.networkInterfaces()),
};
break;
}
return {
content: [
{
type: 'text',
text: `System ${info_type} information:\n${JSON.stringify(info, null, 2)}`,
} as TextContent,
],
};
} catch (error) {
return {
content: [
{
type: 'text',
text: `System info error: ${error instanceof Error ? error.message : String(error)}`,
} as TextContent,
],
};
}
}
private isValidJSON(data: string): boolean {
try {
JSON.parse(data);
return true;
} catch {
return false;
}
}
async run(): Promise<void> {
const transport = new StdioServerTransport();
await this.server.connect(transport);
}
}
// Entry point
async function main() {
const server = new CustomMCPServer();
await server.run();
}
if (require.main === module) {
main().catch(console.error);
}{
"name": "custom-mcp-server-js",
"version": "1.0.0",
"description": "Custom MCP Server in TypeScript",
"main": "dist/server.js",
"type": "module",
"scripts": {
"build": "tsc",
"start": "node dist/server.js",
"dev": "tsc && node dist/server.js"
},
"bin": {
"custom-mcp-server-js": "./dist/server.js"
},
"dependencies": {
"@modelcontextprotocol/sdk": "^1.0.0"
},
"devDependencies": {
"@types/node": "^20.0.0",
"typescript": "^5.0.0"
}
}# tests/test_server.py
import pytest
import asyncio
from custom_mcp_server import CustomMCPServer
@pytest.fixture
async def server():
return CustomMCPServer()
@pytest.mark.asyncio
async def test_process_data_analyze(server):
args = {"data": "Hello world test", "operation": "analyze"}
result = await server._process_data(args)
assert len(result) == 1
assert "length" in result[0].text
assert "words" in result[0].text
@pytest.mark.asyncio
async def test_file_operations_safety(server):
args = {"action": "read", "path": "/etc/passwd"}
result = await server._file_operations(args)
assert "not in allowed directories" in result[0].text#!/bin/bash
# test_integration.sh
echo "Testing MCP server integration..."
# Start server in background
python3 custom_mcp_server.py &
SERVER_PID=$!
# Wait for server to start
sleep 2
# Test tools list
echo '{"jsonrpc": "2.0", "id": 1, "method": "tools/list"}' | python3 custom_mcp_server.py
# Test tool call
echo '{"jsonrpc": "2.0", "id": 2, "method": "tools/call", "params": {"name": "process_data", "arguments": {"data": "test", "operation": "analyze"}}}' | python3 custom_mcp_server.py
# Cleanup
kill $SERVER_PID# debug_server.py
import logging
import asyncio
from custom_mcp_server import CustomMCPServer
# Enable debug logging
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
async def debug_main():
server = CustomMCPServer()
# Add debug hooks
original_call_tool = server.server._request_handlers.get("tools/call")
async def debug_call_tool(request):
logging.debug(f"Tool call: {request.params}")
result = await original_call_tool(request)
logging.debug(f"Tool result: {result}")
return result
server.server.set_request_handler("tools/call", debug_call_tool)
await server.run()
if __name__ == "__main__":
asyncio.run(debug_main())# Dockerfile
FROM python:3.11-slim
WORKDIR /app
COPY requirements.txt .
RUN pip install -r requirements.txt
COPY . .
RUN pip install -e .
EXPOSE 8080
CMD ["python", "-m", "custom_mcp_server"]# /etc/systemd/user/custom-mcp-server.service
[Unit]
Description=Custom MCP Server
After=network.target
[Service]
Type=simple
ExecStart=/usr/local/bin/custom-mcp-server
Restart=always
RestartSec=5
Environment=MCP_LOG_LEVEL=info
[Install]
WantedBy=default.target# Build and distribute
python setup.py sdist bdist_wheel
twine upload dist/*
# Or for local installation
pip install -e .- Always wrap tool calls in try-catch blocks
- Provide meaningful error messages
- Log errors for debugging
- Return structured error responses
- Validate all inputs
- Implement path traversal protection
- Use allowlists for file operations
- Sanitize user data
- Use async/await for I/O operations
- Implement caching where appropriate
- Set reasonable timeouts
- Monitor resource usage
- Write comprehensive unit tests
- Test error conditions
- Use integration tests
- Mock external dependencies
Phát triển MCP server tùy chỉnh mở ra nhiều khả năng mới cho việc mở rộng Amazon Q CLI. Bằng cách tuân theo các best practices và sử dụng các patterns được đề xuất, bạn có thể tạo ra các servers mạnh mẽ và đáng tin cậy.
Tác giả: ducla-cloud