# Create a new SQLite database
sqlite3 ~/${YOUR_PATH}/test.db <<EOF
CREATE TABLE products (
  id INTEGER PRIMARY KEY,
  name TEXT,
  price REAL
);

INSERT INTO products (name, price) VALUES
  ('Widget', 19.99),
  ('Gadget', 29.99),
  ('Gizmo', 39.99),
  ('Smart Watch', 199.99),
  ('Wireless Earbuds', 89.99),
  ('Portable Charger', 24.99),
  ('Bluetooth Speaker', 79.99),
  ('Phone Stand', 15.99),
  ('Laptop Sleeve', 34.99),
  ('Mini Drone', 299.99),
  ('LED Desk Lamp', 45.99),
  ('Keyboard', 129.99),
  ('Mouse Pad', 12.99),
  ('USB Hub', 49.99),
  ('Webcam', 69.99),
  ('Screen Protector', 9.99),
  ('Travel Adapter', 27.99),
  ('Gaming Headset', 159.99),
  ('Fitness Tracker', 119.99),
  ('Portable SSD', 179.99);
EOF

{
  "mcpServers": {
    "sqlite": {
      "command": "uvx",
      "args": ["mcp-server-sqlite", "--db-path", "/Users/${YOUR_USERNAME}/${YOUR_PATH}/test.db"]
    }
  }
}

[project]
name = "mcp-server-sqlite"
version = "0.6.2"
description = "A simple SQLite MCP server"
readme = "README.md"
requires-python = ">=3.10"
dependencies = ["mcp>=1.0.0"]

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

[tool.uv]
dev-dependencies = ["pyright>=1.1.389"]

[project.scripts]
mcp-server-sqlite = "mcp_server_sqlite:main"

ls src/mcp_server_sqlite
__init__.py
server.py

from . import server
import asyncio
import argparse


def main():
    """Main entry point for the package."""
    parser = argparse.ArgumentParser(description='SQLite MCP Server')
    parser.add_argument('--db-path', 
                       default="./sqlite_mcp_server.db",
                       help='Path to SQLite database file')
    
    args = parser.parse_args()
    asyncio.run(server.main(args.db_path))


# Optionally expose other important items at package level
__all__ = ["main", "server"]

class SqliteDatabase:
    def __init__(self, db_path: str):
        self.db_path = str(Path(db_path).expanduser())
        ...

    def _init_database(self):
        """Initialize connection to the SQLite database"""
        ...

    def _synthesize_memo(self) -> str:
        """Synthesizes business insights into a formatted memo"""
        logger.debug(f"Synthesizing memo with {len(self.insights)} insights")
        ...

    def _execute_query(self, query: str, params: dict[str, Any] | None = None) -> list[dict[str, Any]]:
        """Execute a SQL query and return results as a list of dictionaries"""
        logger.debug(f"Executing query: {query}")
        ...

...
from mcp.server.models import InitializationOptions
import mcp.types as types
from mcp.server import NotificationOptions, Server
import mcp.server.stdio
...

PROMPT_TEMPLATE = """
The assistants goal is to walkthrough an informative demo of MCP. To demonstrate the Model Context Protocol (MCP) we will leverage this example server to interact with an SQLite database.
....
"""


async def main(db_path: str):
    logger.info(f"Starting SQLite MCP Server with DB path: {db_path}")

    db = SqliteDatabase(db_path)
    server = Server("sqlite-manager")

    ...

    @server.list_tools()
    async def handle_list_tools() -> list[types.Tool]:
        """List available tools"""
        return [
            types.Tool(
                name="read-query",
                description="Execute a SELECT query on the SQLite database",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "SELECT SQL query to execute"},
                    },
                    "required": ["query"],
                },
            ),
            types.Tool(
                name="write-query",
                description="Execute an INSERT, UPDATE, or DELETE query on the SQLite database",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "SQL query to execute"},
                    },
                    "required": ["query"],
                },
            ),
            types.Tool(
                name="create-table",
                description="Create a new table in the SQLite database",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "CREATE TABLE SQL statement"},
                    },
                    "required": ["query"],
                },
            ),
            ...
        ]

    @server.call_tool()
    async def handle_call_tool(
        name: str, arguments: dict[str, Any] | None
    ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
        """Handle tool execution requests"""
        try:
            ...
            if name == "read-query":
                if not arguments["query"].strip().upper().startswith("SELECT"):
                    raise ValueError("Only SELECT queries are allowed for read-query")
                results = db._execute_query(arguments["query"])
                return [types.TextContent(type="text", text=str(results))]

            elif name == "write-query":
                if arguments["query"].strip().upper().startswith("SELECT"):
                    raise ValueError("SELECT queries are not allowed for write-query")
                results = db._execute_query(arguments["query"])
                return [types.TextContent(type="text", text=str(results))]

            elif name == "create-table":
                if not arguments["query"].strip().upper().startswith("CREATE TABLE"):
                    raise ValueError("Only CREATE TABLE statements are allowed")
                db._execute_query(arguments["query"])
                return [types.TextContent(type="text", text="Table created successfully")]

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

        except sqlite3.Error as e:
            return [types.TextContent(type="text", text=f"Database error: {str(e)}")]
        except Exception as e:
            return [types.TextContent(type="text", text=f"Error: {str(e)}")]

    async with mcp.server.stdio.stdio_server() as (read_stream, write_stream):
        logger.info("Server running with stdio transport")
        await server.run(
            read_stream,
            write_stream,
            InitializationOptions(
                server_name="sqlite",
                server_version="0.1.0",
                capabilities=server.get_capabilities(
                    notification_options=NotificationOptions(),
                    experimental_capabilities={},
                ),
            ),
        )

class Server:
    ...
    def list_tools(self):
        def decorator(func: Callable[[], Awaitable[list[types.Tool]]]):
            logger.debug("Registering handler for ListToolsRequest")

            async def handler(_: Any):
                tools = await func()
                return types.ServerResult(types.ListToolsResult(tools=tools))

            self.request_handlers[types.ListToolsRequest] = handler
            return func

        return decorator

class ListToolsResult(PaginatedResult):
    """The server's response to a tools/list request from the client."""

    tools: list[Tool]

class Tool(BaseModel):
    """Definition for a tool the client can call."""

    name: str
    """The name of the tool."""
    description: str | None = None
    """A human-readable description of the tool."""
    inputSchema: dict[str, Any]
    """A JSON Schema object defining the expected parameters for the tool."""
    model_config = ConfigDict(extra="allow")

            types.Tool(
                name="read-query",
                description="Execute a SELECT query on the SQLite database",
                inputSchema={
                    "type": "object",
                    "properties": {
                        "query": {"type": "string", "description": "SELECT SQL query to execute"},
                    },
                    "required": ["query"],
                },
            ),

class Server:
    ...
    def call_tool(self):
        def decorator(
            func: Callable[
                ...,
                Awaitable[
                    Sequence[
                        types.TextContent | types.ImageContent | types.EmbeddedResource
                    ]
                ],
            ],
        ):
            logger.debug("Registering handler for CallToolRequest")

            async def handler(req: types.CallToolRequest):
                try:
                    results = await func(req.params.name, (req.params.arguments or {}))
                    return types.ServerResult(
                        types.CallToolResult(content=list(results), isError=False)
                    )
                except Exception as e:
                    return types.ServerResult(
                        types.CallToolResult(
                            content=[types.TextContent(type="text", text=str(e))],
                            isError=True,
                        )
                    )

            self.request_handlers[types.CallToolRequest] = handler
            return func

        return decorator

@server.call_tool()
    async def handle_call_tool(
        name: str, arguments: dict[str, Any] | None
    ) -> list[types.TextContent | types.ImageContent | types.EmbeddedResource]:
        """Handle tool execution requests"""
        try:
            ...
            if name == "read-query":
                if not arguments["query"].strip().upper().startswith("SELECT"):
                    raise ValueError("Only SELECT queries are allowed for read-query")
                results = db._execute_query(arguments["query"])
                return [types.TextContent(type="text", text=str(results))]
            ...

npx @cloudflare/mcp-server-cloudflare init

export async function init(accountTag: string | undefined) {
  ...
  try {
    getAuthTokens()
  } catch (e: any) {
    ...
  }

  if (isAccessTokenExpired()) {
    updateStatus(`Access token expired, refreshing...`, false)
    if (await refreshToken()) {
      updateStatus('Successfully refreshed access token')
    } else {
      throw new Error('Failed to refresh access token')
    }
  }

  const { result: accounts } = await fetchInternal<FetchResult<AccountInfo[]>>('/accounts')
  ...

  const claudeConfigPath = path.join(
    os.homedir(),
    'Library',
    'Application Support',
    'Claude',
    'claude_desktop_config.json',
  )
  const cloudflareConfig = {
    command: (await which('node')).trim(),
    args: [__filename, 'run', account],
  }

  const configDirExists = isDirectory(path.dirname(claudeConfigPath))
  if (configDirExists) {
    const existingConfig = fs.existsSync(claudeConfigPath)
      ? JSON.parse(fs.readFileSync(claudeConfigPath, 'utf8'))
      : { mcpServers: {} }
    if ('cloudflare' in (existingConfig?.mcpServers || {})) {
      updateStatus(
        `${chalk.green('Note:')} Replacing existing Cloudflare MCP config:\n${chalk.gray(JSON.stringify(existingConfig.mcpServers.cloudflare))}`,
      )
    }
    const newConfig = {
      ...existingConfig,
      mcpServers: {
        ...existingConfig.mcpServers,
        cloudflare: cloudflareConfig,
      },
    }
    fs.writeFileSync(claudeConfigPath, JSON.stringify(newConfig, null, 2))
  } else {
    ...
  }
}

npx @cloudflare/mcp-server-cloudflare init

// UUID生成のヘルパー関数
function generateUUID() {
  return 'xxxxxxxx-xxxx-4xxx-yxxx-xxxxxxxxxxxx'.replace(/[xy]/g, function(c) {
    const r = Math.random() * 16 | 0;
    const v = c == 'x' ? r : (r & 0x3 | 0x8);
    return v.toString(16);
  });
}

// CORSヘッダーの設定
const corsHeaders = {
  'Access-Control-Allow-Origin': '*',
  'Access-Control-Allow-Methods': 'GET, POST, PUT, DELETE, OPTIONS',
  'Access-Control-Allow-Headers': 'Content-Type',
};

// レスポンスヘルパー関数
function jsonResponse(body, status = 200) {
  return new Response(JSON.stringify(body), {
    status,
    headers: {
      'Content-Type': 'application/json',
      ...corsHeaders,
    },
  });
}

export default {
  async fetch(request, env) {
    // プリフライトリクエストの処理
    if (request.method === 'OPTIONS') {
      return new Response(null, {
        headers: corsHeaders,
      });
    }

    const url = new URL(request.url);
    const path = url.pathname;

    // ToDo一覧の取得
    if (request.method === 'GET' && path === '/todos') {
      try {
        const todoList = await env.TODO_STORE.get('todos');
        return jsonResponse(todoList ? JSON.parse(todoList) : []);
      } catch (error) {
        return jsonResponse({ error: 'Failed to fetch todos' }, 500);
      }
    }

    // ToDo の追加
    if (request.method === 'POST' && path === '/todos') {
      try {
        const body = await request.json();
        if (!body.title) {
          return jsonResponse({ error: 'Title is required' }, 400);
        }

        const todo = {
          id: generateUUID(),
          title: body.title,
          completed: false,
          createdAt: new Date().toISOString(),
        };

        const existingTodos = await env.TODO_STORE.get('todos');
        const todos = existingTodos ? JSON.parse(existingTodos) : [];
        todos.push(todo);

        await env.TODO_STORE.put('todos', JSON.stringify(todos));
        return jsonResponse(todo, 201);
      } catch (error) {
        return jsonResponse({ error: 'Failed to create todo' }, 500);
      }
    }

    // ToDo の更新
    if (request.method === 'PUT' && path.startsWith('/todos/')) {
      try {
        const id = path.split('/')[2];
        const body = await request.json();

        const existingTodos = await env.TODO_STORE.get('todos');
        if (!existingTodos) {
          return jsonResponse({ error: 'Todo not found' }, 404);
        }

        const todos = JSON.parse(existingTodos);
        const todoIndex = todos.findIndex(todo => todo.id === id);

        if (todoIndex === -1) {
          return jsonResponse({ error: 'Todo not found' }, 404);
        }

        todos[todoIndex] = {
          ...todos[todoIndex],
          ...body,
          updatedAt: new Date().toISOString(),
        };

        await env.TODO_STORE.put('todos', JSON.stringify(todos));
        return jsonResponse(todos[todoIndex]);
      } catch (error) {
        return jsonResponse({ error: 'Failed to update todo' }, 500);
      }
    }

    // ToDo の削除
    if (request.method === 'DELETE' && path.startsWith('/todos/')) {
      try {
        const id = path.split('/')[2];
        const existingTodos = await env.TODO_STORE.get('todos');

        if (!existingTodos) {
          return jsonResponse({ error: 'Todo not found' }, 404);
        }

        const todos = JSON.parse(existingTodos);
        const filteredTodos = todos.filter(todo => todo.id !== id);

        if (filteredTodos.length === todos.length) {
          return jsonResponse({ error: 'Todo not found' }, 404);
        }

        await env.TODO_STORE.put('todos', JSON.stringify(filteredTodos));
        return jsonResponse({ message: 'Todo deleted successfully' });
      } catch (error) {
        return jsonResponse({ error: 'Failed to delete todo' }, 500);
      }
    }

    // 404 Not Found
    return jsonResponse({ error: 'Not found' }, 404);
  }
};

{
  "mcpServers": {
    ...
    "github": {
      "command": "npx",
      "args": [
        "-y",
        "@modelcontextprotocol/server-github"
      ],
      "env": {
        "GITHUB_PERSONAL_ACCESS_TOKEN": "<YOUR_TOKEN>"
      }
    }
  }
}