Python Project Structure¶

import asyncio
from mcp.server import Server
from mcp.server.stdio import stdio_server
from .tools import register_tools
from .resources import register_resources

class MCPServer:
    def __init__(self, name: str):
        self.server = Server(name)
        self._register_handlers()
    
    def _register_handlers(self):
        register_tools(self.server)
        register_resources(self.server)
    
    async def run(self):
        async with stdio_server() as (read_stream, write_stream):
            await self.server.run(read_stream, write_stream)

def main():
    server = MCPServer("my-mcp-server")
    asyncio.run(server.run())

if __name__ == "__main__":
    main()

from mcp.server import Server
from .database import register_database_tools
from .file_ops import register_file_tools

def register_tools(server: Server):
    """Register all tools with the server."""
    register_database_tools(server)
    register_file_tools(server)

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

def register_database_tools(server: Server):
    """Register database-related tools."""
    
    @server.list_tools()
    async def list_tools():
        return [
            Tool(
                name="query_database",
                description="Execute a SQL query",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string"},
                        "limit": {"type": "integer", "default": 100}
                    },
                    "required": ["query"]
                }
            )
        ]
    
    @server.call_tool()
    async def call_tool(name: str, arguments: dict):
        if name == "query_database":
            return await _execute_query(arguments)
        raise ValueError(f"Unknown tool: {name}")

async def _execute_query(args: dict):
    """Execute database query (implementation)."""
    query = args["query"]
    limit = args.get("limit", 100)
    
    # Your database logic here
    result = {"query": query, "limit": limit, "results": []}
    
    return [TextContent(type="text", text=json.dumps(result))]

import os
from dataclasses import dataclass
from typing import Optional

@dataclass
class Settings:
    """Application settings from environment variables."""
    
    # Server settings
    server_name: str = "my-mcp-server"
    debug: bool = False
    
    # Database settings
    database_url: Optional[str] = None
    database_timeout: int = 30
    
    # API settings
    api_key: Optional[str] = None
    api_base_url: str = "https://api.example.com"
    
    @classmethod
    def from_env(cls) -> 'Settings':
        """Load settings from environment variables."""
        return cls(
            server_name=os.getenv("MCP_SERVER_NAME", cls.server_name),
            debug=os.getenv("MCP_DEBUG", "false").lower() == "true",
            database_url=os.getenv("DATABASE_URL"),
            database_timeout=int(os.getenv("DATABASE_TIMEOUT", "30")),
            api_key=os.getenv("API_KEY"),
            api_base_url=os.getenv("API_BASE_URL", cls.api_base_url)
        )

# Global settings instance
settings = Settings.from_env()

#!/usr/bin/env python3
"""Development server runner with hot reload."""

import sys
import os
from pathlib import Path

# Add src to path
src_path = Path(__file__).parent.parent / "src"
sys.path.insert(0, str(src_path))

from my_mcp_server.server import main

if __name__ == "__main__":
    # Set development environment
    os.environ.setdefault("MCP_DEBUG", "true")
    main()