server.py•3.14 kB
"""FastMCP server for SQLite database operations."""
from typing import Any, Dict, List, Optional
from fastmcp import FastMCP
from .db import SQLiteDatabase
# Initialize FastMCP server
mcp = FastMCP("sqlite-mcp")
# Initialize database instance
db = SQLiteDatabase()
@mcp.tool()
def open_database(path: str) -> str:
"""
Open or create a SQLite database file.
Args:
path: Path to the SQLite database file
Returns:
Success message
"""
return db.open(path)
@mcp.tool()
def close_database() -> str:
"""
Close the current database connection.
Returns:
Success message
"""
return db.close()
@mcp.tool()
def execute_query(query: str, parameters: Optional[List[Any]] = None) -> Dict[str, Any]:
"""
Execute a SELECT query and return results.
Args:
query: SQL SELECT query to execute
parameters: Query parameters for prepared statements
Returns:
Dictionary with rows and column count
"""
return db.execute_query(query, parameters)
@mcp.tool()
def insert(table: str, data: Dict[str, Any]) -> Dict[str, Any]:
"""
Insert a row into a table.
Args:
table: Table name
data: Column names and values to insert
Returns:
Dictionary with lastID and changes count
"""
return db.insert(table, data)
@mcp.tool()
def update(
table: str,
data: Dict[str, Any],
where: str,
where_params: Optional[List[Any]] = None,
) -> Dict[str, Any]:
"""
Update rows in a table.
Args:
table: Table name
data: Column names and new values
where: WHERE clause condition (e.g., 'id = ?')
where_params: Parameters for WHERE clause
Returns:
Dictionary with changes count
"""
return db.update(table, data, where, where_params)
@mcp.tool()
def delete(
table: str,
where: str,
where_params: Optional[List[Any]] = None,
) -> Dict[str, Any]:
"""
Delete rows from a table.
Args:
table: Table name
where: WHERE clause condition (e.g., 'id = ?')
where_params: Parameters for WHERE clause
Returns:
Dictionary with changes count
"""
return db.delete(table, where, where_params)
@mcp.tool()
def create_table(table: str, schema: str) -> str:
"""
Create a new table.
Args:
table: Table name
schema: Column definitions (e.g., 'id INTEGER PRIMARY KEY, name TEXT')
Returns:
Success message
"""
return db.create_table(table, schema)
@mcp.tool()
def list_tables() -> Dict[str, List[str]]:
"""
List all tables in the database.
Returns:
Dictionary with list of table names
"""
return db.list_tables()
@mcp.tool()
def get_table_schema(table: str) -> Dict[str, List[Dict[str, Any]]]:
"""
Get the schema of a table.
Args:
table: Table name
Returns:
Dictionary with column information
"""
return db.get_table_schema(table)
def main():
"""Start the FastMCP server."""
mcp.run(transport="stdio")
if __name__ == "__main__":
main()