binary1ne/MCPServer

1import sqlite32import argparse3from mcp.server.fastmcp import FastMCP4 5mcp = FastMCP('sqlite-demo')6 7def init_db():8    conn = sqlite3.connect('demo.db')9    cursor = conn.cursor()10    cursor.execute('''11        CREATE TABLE IF NOT EXISTS people (12            id INTEGER PRIMARY KEY AUTOINCREMENT,13            name TEXT NOT NULL,14            age INTEGER NOT NULL,15            profession TEXT NOT NULL16        )17    ''')18    conn.commit()19    return conn, cursor20 21@mcp.tool()22def add_data(query: str) -> bool:23    """Add new data to the people table using a SQL INSERT query.24 25    Args:26        query (str): SQL INSERT query following this format:27            INSERT INTO people (name, age, profession)28            VALUES ('John Doe', 30, 'Engineer')29        30    Schema:31        - name: Text field (required)32        - age: Integer field (required)33        - profession: Text field (required)34        Note: 'id' field is auto-generated35    36    Returns:37        bool: True if data was added successfully, False otherwise38    39    Example:40        >>> query = '''41        ... INSERT INTO people (name, age, profession)42        ... VALUES ('Alice Smith', 25, 'Developer')43        ... '''44        >>> add_data(query)45        True46    """47    conn, cursor = init_db()48    try:49        cursor.execute(query)50        conn.commit()51        return True52    except sqlite3.Error as e:53        print(f"Error adding data: {e}")54        return False55    finally:56        conn.close()57 58@mcp.tool()59def read_data(query: str = "SELECT * FROM people") -> list:60    """Read data from the people table using a SQL SELECT query.61 62    Args:63        query (str, optional): SQL SELECT query. Defaults to "SELECT * FROM people".64            Examples:65            - "SELECT * FROM people"66            - "SELECT name, age FROM people WHERE age > 25"67            - "SELECT * FROM people ORDER BY age DESC"68    69    Returns:70        list: List of tuples containing the query results.71              For default query, tuple format is (id, name, age, profession)72    73    Example:74        >>> # Read all records75        >>> read_data()76        [(1, 'John Doe', 30, 'Engineer'), (2, 'Alice Smith', 25, 'Developer')]77        78        >>> # Read with custom query79        >>> read_data("SELECT name, profession FROM people WHERE age < 30")80        [('Alice Smith', 'Developer')]81    """82    conn, cursor = init_db()83    try:84        cursor.execute(query)85        return cursor.fetchall()86    except sqlite3.Error as e:87        print(f"Error reading data: {e}")88        return []89    finally:90        conn.close()91 92 93 94if __name__ == "__main__":95    # Start the server96    print("🚀Starting server... ")97 98    # Debug Mode99    #  uv run mcp dev server.py100 101    # Production Mode102    # uv run server.py --server_type=sse103 104    parser = argparse.ArgumentParser()105    parser.add_argument(106        "--server_type", type=str, default="sse", choices=["sse", "stdio"]107    )108 109    args = parser.parse_args()110    mcp.run(args.server_type)111 112 113 114# # Example usage115# if __name__ == "__main__":116#     # Example INSERT query117#     insert_query = """118#     INSERT INTO people (name, age, profession)119#     VALUES ('John Doe', 30, 'Engineer')120#     """121    122#     # Add data123#     if add_data(insert_query):124#         print("Data added successfully")125    126#     # Read all data127#     results = read_data()128#     print("\nAll records:")129#     for record in results:130#         print(record)131