Building MCP Servers for Claude Code Development - Part 1

 1# Create a new project directory
 2mkdir my-mcp-server
 3cd my-mcp-server
 4
 5# Create virtual environment
 6python -m venv mcp-env
 7source mcp-env/bin/activate  # On Windows: mcp-env\Scripts\activate
 8
 9# Install MCP SDK
10pip install mcp
11
12# Install additional dependencies
13pip install asyncio aiofiles requests

 1# Create a new project directory
 2mkdir my-mcp-server-ts
 3cd my-mcp-server-ts
 4
 5# Initialize npm project
 6npm init -y
 7
 8# Install MCP SDK
 9npm install @modelcontextprotocol/sdk
10
11# Install development dependencies
12npm install -D typescript @types/node ts-node
13npm install axios fs-extra

  1#!/usr/bin/env python3
  2"""
  3Basic MCP Server - Part 1
  4Provides simple utility tools for Claude Code
  5"""
  6
  7import asyncio
  8import json
  9import logging
 10from datetime import datetime
 11from pathlib import Path
 12from typing import Dict, List, Any
 13
 14from mcp.server.models import InitializeResult
 15from mcp.server import Server, NotificationOptions
 16from mcp.server.stdio import stdio_server
 17from mcp.types import (
 18    Resource,
 19    Tool,
 20    TextContent,
 21    ImageContent,
 22    EmbeddedResource,
 23)
 24
 25# Configure logging
 26logging.basicConfig(level=logging.INFO)
 27logger = logging.getLogger("basic-mcp-server")
 28
 29# Create MCP server instance
 30server = Server("basic-mcp-server")
 31
 32@server.list_tools()
 33async def handle_list_tools() -> List[Tool]:
 34    """
 35    Define available tools that Claude Code can use
 36    """
 37    return [
 38        Tool(
 39            name="get_current_time",
 40            description="Get the current date and time in various formats",
 41            inputSchema={
 42                "type": "object",
 43                "properties": {
 44                    "format": {
 45                        "type": "string",
 46                        "description": "Time format: 'iso', 'human', or 'unix'",
 47                        "enum": ["iso", "human", "unix"]
 48                    },
 49                    "timezone": {
 50                        "type": "string",
 51                        "description": "Timezone (default: UTC)",
 52                        "default": "UTC"
 53                    }
 54                },
 55                "required": ["format"]
 56            }
 57        ),
 58        Tool(
 59            name="file_stats",
 60            description="Get detailed statistics about files and directories",
 61            inputSchema={
 62                "type": "object",
 63                "properties": {
 64                    "path": {
 65                        "type": "string",
 66                        "description": "Path to file or directory to analyze"
 67                    },
 68                    "recursive": {
 69                        "type": "boolean",
 70                        "description": "Include subdirectories in analysis",
 71                        "default": False
 72                    }
 73                },
 74                "required": ["path"]
 75            }
 76        ),
 77        Tool(
 78            name="generate_uuid",
 79            description="Generate UUIDs in various formats",
 80            inputSchema={
 81                "type": "object",
 82                "properties": {
 83                    "count": {
 84                        "type": "integer",
 85                        "description": "Number of UUIDs to generate",
 86                        "default": 1,
 87                        "minimum": 1,
 88                        "maximum": 100
 89                    },
 90                    "format": {
 91                        "type": "string",
 92                        "description": "UUID format: 'standard', 'compact', or 'upper'",
 93                        "enum": ["standard", "compact", "upper"],
 94                        "default": "standard"
 95                    }
 96                }
 97            }
 98        ),
 99        Tool(
100            name="system_info",
101            description="Get system information including OS, Python version, and environment",
102            inputSchema={
103                "type": "object",
104                "properties": {}
105            }
106        )
107    ]
108
109@server.call_tool()
110async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
111    """
112    Handle tool execution requests from Claude Code
113    """
114
115    if name == "get_current_time":
116        format_type = arguments.get("format", "iso")
117        timezone = arguments.get("timezone", "UTC")
118
119        now = datetime.now()
120
121        if format_type == "iso":
122            result = now.isoformat()
123        elif format_type == "human":
124            result = now.strftime("%A, %B %d, %Y at %I:%M:%S %p")
125        elif format_type == "unix":
126            result = str(int(now.timestamp()))
127        else:
128            result = now.isoformat()
129
130        return [TextContent(
131            type="text",
132            text=f"Current time ({format_type} format): {result}"
133        )]
134
135    elif name == "file_stats":
136        import os
137        import stat
138        from pathlib import Path
139
140        path_str = arguments.get("path")
141        recursive = arguments.get("recursive", False)
142
143        try:
144            path = Path(path_str)
145
146            if not path.exists():
147                return [TextContent(
148                    type="text",
149                    text=f"Error: Path '{path_str}' does not exist"
150                )]
151
152            stats = {
153                "path": str(path.absolute()),
154                "exists": path.exists(),
155                "is_file": path.is_file(),
156                "is_directory": path.is_dir(),
157                "is_symlink": path.is_symlink()
158            }
159
160            if path.exists():
161                stat_info = path.stat()
162                stats.update({
163                    "size_bytes": stat_info.st_size,
164                    "size_human": _human_readable_size(stat_info.st_size),
165                    "modified": datetime.fromtimestamp(stat_info.st_mtime).isoformat(),
166                    "created": datetime.fromtimestamp(stat_info.st_ctime).isoformat(),
167                    "permissions": oct(stat.S_IMODE(stat_info.st_mode))
168                })
169
170            if path.is_dir() and recursive:
171                file_count = 0
172                dir_count = 0
173                total_size = 0
174
175                for item in path.rglob("*"):
176                    if item.is_file():
177                        file_count += 1
178                        try:
179                            total_size += item.stat().st_size
180                        except (OSError, IOError):
181                            pass
182                    elif item.is_dir():
183                        dir_count += 1
184
185                stats["recursive_stats"] = {
186                    "total_files": file_count,
187                    "total_directories": dir_count,
188                    "total_size_bytes": total_size,
189                    "total_size_human": _human_readable_size(total_size)
190                }
191
192            return [TextContent(
193                type="text",
194                text=f"File statistics:\n{json.dumps(stats, indent=2)}"
195            )]
196
197        except Exception as e:
198            return [TextContent(
199                type="text",
200                text=f"Error analyzing path '{path_str}': {str(e)}"
201            )]
202
203    elif name == "generate_uuid":
204        import uuid
205
206        count = arguments.get("count", 1)
207        format_type = arguments.get("format", "standard")
208
209        uuids = []
210        for _ in range(count):
211            new_uuid = uuid.uuid4()
212
213            if format_type == "standard":
214                uuids.append(str(new_uuid))
215            elif format_type == "compact":
216                uuids.append(str(new_uuid).replace("-", ""))
217            elif format_type == "upper":
218                uuids.append(str(new_uuid).upper())
219
220        result = "\n".join(uuids) if count > 1 else uuids[0]
221
222        return [TextContent(
223            type="text",
224            text=f"Generated {count} UUID(s) in {format_type} format:\n{result}"
225        )]
226
227    elif name == "system_info":
228        import platform
229        import sys
230        import os
231
232        info = {
233            "system": platform.system(),
234            "platform": platform.platform(),
235            "architecture": platform.architecture(),
236            "processor": platform.processor(),
237            "python_version": sys.version,
238            "python_executable": sys.executable,
239            "current_directory": os.getcwd(),
240            "environment_variables": {
241                key: value for key, value in os.environ.items()
242                if key.startswith(('PATH', 'HOME', 'USER', 'SHELL', 'PYTHON'))
243            }
244        }
245
246        return [TextContent(
247            type="text",
248            text=f"System Information:\n{json.dumps(info, indent=2)}"
249        )]
250
251    else:
252        return [TextContent(
253            type="text",
254            text=f"Error: Unknown tool '{name}'"
255        )]
256
257def _human_readable_size(size_bytes: int) -> str:
258    """Convert bytes to human readable format"""
259    if size_bytes == 0:
260        return "0 B"
261
262    size_names = ["B", "KB", "MB", "GB", "TB"]
263    i = 0
264    size = float(size_bytes)
265
266    while size >= 1024.0 and i < len(size_names) - 1:
267        size /= 1024.0
268        i += 1
269
270    return f"{size:.1f} {size_names[i]}"
271
272async def main():
273    """
274    Main server entry point
275    """
276    logger.info("Starting Basic MCP Server...")
277
278    # Run the server using stdin/stdout for communication with Claude Code
279    async with stdio_server() as (read_stream, write_stream):
280        await server.run(
281            read_stream,
282            write_stream,
283            InitializeResult(
284                protocolVersion="2024-11-05",
285                capabilities=server.get_capabilities(
286                    notification_options=NotificationOptions(),
287                    experimental_capabilities={}
288                )
289            )
290        )
291
292if __name__ == "__main__":
293    asyncio.run(main())

 1"""
 2Utility functions for MCP server
 3"""
 4
 5import json
 6import logging
 7from pathlib import Path
 8from typing import Dict, Any, List
 9from datetime import datetime
10
11logger = logging.getLogger(__name__)
12
13class MCPServerUtils:
14    """Utility class for common MCP server operations"""
15
16    @staticmethod
17    def validate_path(path_str: str) -> Path:
18        """Validate and return Path object"""
19        try:
20            path = Path(path_str).resolve()
21            return path
22        except Exception as e:
23            raise ValueError(f"Invalid path '{path_str}': {e}")
24
25    @staticmethod
26    def safe_json_dumps(data: Any, indent: int = 2) -> str:
27        """Safely serialize data to JSON"""
28        try:
29            return json.dumps(data, indent=indent, default=str)
30        except Exception as e:
31            logger.error(f"JSON serialization error: {e}")
32            return f"Error serializing data: {e}"
33
34    @staticmethod
35    def log_tool_call(tool_name: str, arguments: Dict[str, Any]):
36        """Log tool execution for debugging"""
37        logger.info(f"Tool called: {tool_name}")
38        logger.debug(f"Arguments: {arguments}")
39
40    @staticmethod
41    def create_error_response(error_message: str) -> str:
42        """Create standardized error response"""
43        return f"Error: {error_message}"
44
45    @staticmethod
46    def format_file_listing(path: Path, max_items: int = 50) -> Dict[str, Any]:
47        """Format directory listing with metadata"""
48        if not path.is_dir():
49            raise ValueError(f"Path '{path}' is not a directory")
50
51        items = []
52        count = 0
53
54        try:
55            for item in sorted(path.iterdir()):
56                if count >= max_items:
57                    break
58
59                try:
60                    stat = item.stat()
61                    items.append({
62                        "name": item.name,
63                        "type": "directory" if item.is_dir() else "file",
64                        "size": stat.st_size if item.is_file() else None,
65                        "modified": datetime.fromtimestamp(stat.st_mtime).isoformat(),
66                        "permissions": oct(stat.S_IMODE(stat.st_mode))
67                    })
68                    count += 1
69                except (OSError, IOError) as e:
70                    logger.warning(f"Could not access {item}: {e}")
71                    continue
72
73        except PermissionError:
74            raise ValueError(f"Permission denied accessing directory '{path}'")
75
76        return {
77            "directory": str(path),
78            "total_items_shown": len(items),
79            "items": items,
80            "truncated": count >= max_items
81        }

 1{
 2  "server": {
 3    "name": "basic-mcp-server",
 4    "version": "1.0.0",
 5    "description": "Basic MCP server for Claude Code development",
 6    "author": "Your Name"
 7  },
 8  "logging": {
 9    "level": "INFO",
10    "format": "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
11    "file": "mcp-server.log"
12  },
13  "tools": {
14    "enabled": [
15      "get_current_time",
16      "file_stats",
17      "generate_uuid",
18      "system_info"
19    ],
20    "rate_limits": {
21      "default": 100,
22      "file_stats": 50
23    }
24  },
25  "security": {
26    "allowed_paths": [
27      ".",
28      "~/",
29      "/tmp"
30    ],
31    "blocked_paths": [
32      "/etc/passwd",
33      "/etc/shadow",
34      "~/.ssh"
35    ]
36  }
37}

 1{
 2  "mcpServers": {
 3    "basic-tools": {
 4      "command": "python",
 5      "args": ["/path/to/your/server.py"],
 6      "env": {
 7        "PYTHONPATH": "/path/to/your/project"
 8      },
 9      "description": "Basic utility tools for development",
10      "enabled": true,
11      "timeout": 30000
12    }
13  }
14}

 1{
 2  "mcpServers": {
 3    "basic-tools-ts": {
 4      "command": "npm",
 5      "args": ["run", "start:mcp"],
 6      "cwd": "/path/to/your/typescript/project",
 7      "description": "TypeScript MCP server with basic tools",
 8      "enabled": true
 9    }
10  }
11}

1# Check Python path and dependencies
2python -c "import mcp; print('MCP installed successfully')"
3
4# Verify server script syntax
5python -m py_compile server.py
6
7# Check logs
8tail -f mcp-server.log

 1# Add these additional tools to your server
 2
 3@server.call_tool()
 4async def handle_call_tool(name: str, arguments: Dict[str, Any]) -> List[TextContent]:
 5    """Extended tool handler with more capabilities"""
 6
 7    # ... (previous tools) ...
 8
 9    elif name == "search_files":
10        pattern = arguments.get("pattern", "*")
11        directory = arguments.get("directory", ".")
12        case_sensitive = arguments.get("case_sensitive", False)
13
14        try:
15            path = Path(directory)
16            if not path.exists() or not path.is_dir():
17                return [TextContent(type="text", text=f"Directory '{directory}' not found")]
18
19            import fnmatch
20            matches = []
21
22            for file_path in path.rglob("*"):
23                if file_path.is_file():
24                    filename = file_path.name
25                    if not case_sensitive:
26                        if fnmatch.fnmatch(filename.lower(), pattern.lower()):
27                            matches.append(str(file_path.relative_to(path)))
28                    else:
29                        if fnmatch.fnmatch(filename, pattern):
30                            matches.append(str(file_path.relative_to(path)))
31
32            result = {
33                "search_pattern": pattern,
34                "directory": str(path.absolute()),
35                "matches_found": len(matches),
36                "files": matches[:50]  # Limit results
37            }
38
39            if len(matches) > 50:
40                result["note"] = "Results limited to first 50 matches"
41
42            return [TextContent(
43                type="text",
44                text=f"File search results:\n{json.dumps(result, indent=2)}"
45            )]
46
47        except Exception as e:
48            return [TextContent(type="text", text=f"Search error: {str(e)}")]
49
50    elif name == "calculate_expression":
51        expression = arguments.get("expression", "")
52
53        try:
54            # Safe evaluation of mathematical expressions
55            import ast
56            import operator
57
58            # Allowed operations for safety
59            operators = {
60                ast.Add: operator.add,
61                ast.Sub: operator.sub,
62                ast.Mult: operator.mul,
63                ast.Div: operator.truediv,
64                ast.Pow: operator.pow,
65                ast.USub: operator.neg,
66                ast.UAdd: operator.pos,
67            }
68
69            def eval_expr(node):
70                if isinstance(node, ast.Num):
71                    return node.n
72                elif isinstance(node, ast.BinOp):
73                    return operators[type(node.op)](eval_expr(node.left), eval_expr(node.right))
74                elif isinstance(node, ast.UnaryOp):
75                    return operators[type(node.op)](eval_expr(node.operand))
76                else:
77                    raise TypeError(f"Unsupported operation: {node}")
78
79            tree = ast.parse(expression, mode='eval')
80            result = eval_expr(tree.body)
81
82            return [TextContent(
83                type="text",
84                text=f"Expression: {expression}\nResult: {result}"
85            )]
86
87        except Exception as e:
88            return [TextContent(
89                type="text",
90                text=f"Calculation error for '{expression}': {str(e)}"
91            )]

 1import time
 2import psutil
 3from functools import wraps
 4from typing import Dict, Any
 5
 6class PerformanceMonitor:
 7    def __init__(self):
 8        self.tool_stats = {}
 9        self.server_start_time = time.time()
10
11    def monitor_tool(self, func):
12        """Decorator to monitor tool performance"""
13        @wraps(func)
14        async def wrapper(name: str, arguments: Dict[str, Any], *args, **kwargs):
15            start_time = time.time()
16            memory_before = psutil.Process().memory_info().rss
17
18            try:
19                result = await func(name, arguments, *args, **kwargs)
20
21                end_time = time.time()
22                memory_after = psutil.Process().memory_info().rss
23
24                # Record statistics
25                self.record_tool_stats(name, {
26                    'execution_time': end_time - start_time,
27                    'memory_delta': memory_after - memory_before,
28                    'success': True
29                })
30
31                return result
32
33            except Exception as e:
34                end_time = time.time()
35                self.record_tool_stats(name, {
36                    'execution_time': end_time - start_time,
37                    'memory_delta': 0,
38                    'success': False,
39                    'error': str(e)
40                })
41                raise
42
43        return wrapper
44
45    def record_tool_stats(self, tool_name: str, stats: Dict[str, Any]):
46        """Record performance statistics for a tool"""
47        if tool_name not in self.tool_stats:
48            self.tool_stats[tool_name] = {
49                'call_count': 0,
50                'total_time': 0,
51                'avg_time': 0,
52                'success_count': 0,
53                'error_count': 0
54            }
55
56        tool_stats = self.tool_stats[tool_name]
57        tool_stats['call_count'] += 1
58        tool_stats['total_time'] += stats['execution_time']
59        tool_stats['avg_time'] = tool_stats['total_time'] / tool_stats['call_count']
60
61        if stats['success']:
62            tool_stats['success_count'] += 1
63        else:
64            tool_stats['error_count'] += 1
65
66    def get_stats_summary(self) -> Dict[str, Any]:
67        """Get performance summary"""
68        return {
69            'server_uptime': time.time() - self.server_start_time,
70            'tool_statistics': self.tool_stats,
71            'system_info': {
72                'cpu_percent': psutil.cpu_percent(),
73                'memory_percent': psutil.virtual_memory().percent,
74                'disk_usage': psutil.disk_usage('/').percent
75            }
76        }

 1def validate_file_path(path_str: str, allowed_paths: List[str]) -> bool:
 2    """Validate file paths against allowed directories"""
 3    try:
 4        path = Path(path_str).resolve()
 5
 6        for allowed in allowed_paths:
 7            allowed_path = Path(allowed).resolve()
 8            if path.is_relative_to(allowed_path):
 9                return True
10
11        return False
12    except Exception:
13        return False

 1from collections import defaultdict
 2import time
 3
 4class RateLimiter:
 5    def __init__(self):
 6        self.calls = defaultdict(list)
 7        self.limits = {
 8            'default': 100,  # 100 calls per minute
 9            'file_operations': 50,
10            'system_operations': 20
11        }
12
13    def is_allowed(self, tool_name: str) -> bool:
14        now = time.time()
15        minute_ago = now - 60
16
17        # Clean old entries
18        self.calls[tool_name] = [
19            call_time for call_time in self.calls[tool_name]
20            if call_time > minute_ago
21        ]
22
23        # Check limit
24        limit = self.limits.get(tool_name, self.limits['default'])
25        return len(self.calls[tool_name]) < limit

 1def safe_tool_execution(func):
 2    """Decorator for safe tool execution with error handling"""
 3    @wraps(func)
 4    async def wrapper(*args, **kwargs):
 5        try:
 6            return await func(*args, **kwargs)
 7        except FileNotFoundError as e:
 8            return [TextContent(type="text", text=f"File not found: {e}")]
 9        except PermissionError as e:
10            return [TextContent(type="text", text=f"Permission denied: {e}")]
11        except Exception as e:
12            logger.error(f"Tool execution error: {e}")
13            return [TextContent(type="text", text=f"Unexpected error: {e}")]
14
15    return wrapper

  1#!/usr/bin/env python3
  2"""
  3Test script for MCP server functionality
  4"""
  5
  6import asyncio
  7import json
  8import subprocess
  9import sys
 10from pathlib import Path
 11
 12async def test_server_startup():
 13    """Test if the server starts correctly"""
 14    try:
 15        process = subprocess.Popen([
 16            sys.executable, 'server.py'
 17        ], stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
 18
 19        # Send initialize message
 20        init_message = {
 21            "jsonrpc": "2.0",
 22            "id": 1,
 23            "method": "initialize",
 24            "params": {
 25                "protocolVersion": "2024-11-05",
 26                "capabilities": {}
 27            }
 28        }
 29
 30        process.stdin.write((json.dumps(init_message) + '\n').encode())
 31        process.stdin.flush()
 32
 33        # Wait for response
 34        response = process.stdout.readline()
 35        process.terminate()
 36
 37        if response:
 38            print("✅ Server startup test passed")
 39            return True
 40        else:
 41            print("❌ Server startup test failed")
 42            return False
 43
 44    except Exception as e:
 45        print(f"❌ Server startup test error: {e}")
 46        return False
 47
 48def test_configuration():
 49    """Test configuration file validity"""
 50    config_file = Path("config.json")
 51
 52    if not config_file.exists():
 53        print("❌ Configuration file not found")
 54        return False
 55
 56    try:
 57        with open(config_file) as f:
 58            config = json.load(f)
 59
 60        required_keys = ['server', 'logging', 'tools']
 61        for key in required_keys:
 62            if key not in config:
 63                print(f"❌ Missing configuration key: {key}")
 64                return False
 65
 66        print("✅ Configuration test passed")
 67        return True
 68
 69    except json.JSONDecodeError as e:
 70        print(f"❌ Configuration JSON error: {e}")
 71        return False
 72    except Exception as e:
 73        print(f"❌ Configuration test error: {e}")
 74        return False
 75
 76def test_dependencies():
 77    """Test if all required dependencies are available"""
 78    required_modules = ['mcp', 'asyncio', 'pathlib', 'json']
 79
 80    for module in required_modules:
 81        try:
 82            __import__(module)
 83            print(f"✅ {module} available")
 84        except ImportError:
 85            print(f"❌ {module} not available")
 86            return False
 87
 88    return True
 89
 90async def main():
 91    """Run all tests"""
 92    print("🧪 Testing MCP Server Setup...\n")
 93
 94    tests = [
 95        ("Dependencies", test_dependencies),
 96        ("Configuration", test_configuration),
 97        ("Server Startup", test_server_startup),
 98    ]
 99
100    results = []
101    for test_name, test_func in tests:
102        print(f"Running {test_name} test...")
103        if asyncio.iscoroutinefunction(test_func):
104            result = await test_func()
105        else:
106            result = test_func()
107        results.append(result)
108        print()
109
110    # Summary
111    passed = sum(results)
112    total = len(results)
113
114    print(f"📊 Test Results: {passed}/{total} tests passed")
115
116    if passed == total:
117        print("🎉 All tests passed! Your MCP server is ready for Claude Code.")
118    else:
119        print("⚠️  Some tests failed. Please check the issues above.")
120
121if __name__ == "__main__":
122    asyncio.run(main())

 1# Start your MCP server (for testing)
 2python server.py
 3
 4# Test configuration
 5python test_mcp_server.py
 6
 7# Check logs
 8tail -f mcp-server.log
 9
10# Validate JSON configuration
11python -m json.tool config.json