SQLite MCP Server

"""OpenAI-compatible function definitions for SQLite MCP tools. This module converts MCP tools into OpenAI-compatible function definitions that can be used with sahabat-ai or any OpenAI API-compatible LLM. """ from typing import Any, Dict, List, Optional from enum import Enum class ToolType(Enum): """Types of database operations.""" DATABASE_MANAGEMENT = "database_management" QUERY = "query" INSERT = "insert" UPDATE = "update" DELETE = "delete" SCHEMA = "schema" def get_openai_tools() -> List[Dict[str, Any]]: """Get all MCP tools as OpenAI function definitions. Returns: List of OpenAI-compatible tool definitions """ return [ get_open_database_tool(), get_close_database_tool(), get_execute_query_tool(), get_insert_tool(), get_update_tool(), get_delete_tool(), get_create_table_tool(), get_list_tables_tool(), get_get_table_schema_tool(), ] def get_open_database_tool() -> Dict[str, Any]: """Get open_database tool definition.""" return { "type": "function", "function": { "name": "open_database", "description": "Open or create a SQLite database file. Must be called before any other database operations.", "parameters": { "type": "object", "properties": { "path": { "type": "string", "description": "Path to the SQLite database file (e.g., './data/myapp.db')", } }, "required": ["path"], }, }, } def get_close_database_tool() -> Dict[str, Any]: """Get close_database tool definition.""" return { "type": "function", "function": { "name": "close_database", "description": "Close the current database connection. Call this when done with database operations.", "parameters": { "type": "object", "properties": {}, "required": [], }, }, } def get_execute_query_tool() -> Dict[str, Any]: """Get execute_query tool definition.""" return { "type": "function", "function": { "name": "execute_query", "description": "Execute a SELECT query to retrieve data from the database. Returns rows and column information.", "parameters": { "type": "object", "properties": { "query": { "type": "string", "description": "SQL SELECT query (e.g., 'SELECT * FROM users WHERE age > 18')", }, "parameters": { "type": "array", "items": {"type": ["string", "number", "boolean", "null"]}, "description": "Optional parameters for prepared statements to prevent SQL injection. Use ? as placeholders in query.", "default": [], }, }, "required": ["query"], }, }, } def get_insert_tool() -> Dict[str, Any]: """Get insert tool definition.""" return { "type": "function", "function": { "name": "insert", "description": "Insert a new row into a table. Returns the ID of the inserted row.", "parameters": { "type": "object", "properties": { "table": { "type": "string", "description": "Name of the table to insert into (e.g., 'users')", }, "data": { "type": "object", "description": "Column names and values as key-value pairs. Example: {\"name\": \"John\", \"email\": \"john@example.com\"}", "additionalProperties": { "type": ["string", "number", "boolean", "null"] }, }, }, "required": ["table", "data"], }, }, } def get_update_tool() -> Dict[str, Any]: """Get update tool definition.""" return { "type": "function", "function": { "name": "update", "description": "Update existing rows in a table. Specify which rows to update using the WHERE clause.", "parameters": { "type": "object", "properties": { "table": { "type": "string", "description": "Name of the table to update (e.g., 'users')", }, "data": { "type": "object", "description": "Column names and new values. Example: {\"status\": \"active\", \"age\": 30}", "additionalProperties": { "type": ["string", "number", "boolean", "null"] }, }, "where": { "type": "string", "description": "WHERE clause condition (e.g., 'id = ?', 'email = ?', 'age > ?')", }, "where_params": { "type": "array", "items": {"type": ["string", "number", "boolean", "null"]}, "description": "Parameters for the WHERE clause to prevent SQL injection", "default": [], }, }, "required": ["table", "data", "where"], }, }, } def get_delete_tool() -> Dict[str, Any]: """Get delete tool definition.""" return { "type": "function", "function": { "name": "delete", "description": "Delete rows from a table. Specify which rows to delete using the WHERE clause.", "parameters": { "type": "object", "properties": { "table": { "type": "string", "description": "Name of the table to delete from (e.g., 'users')", }, "where": { "type": "string", "description": "WHERE clause condition (e.g., 'id = ?', 'status = ?')", }, "where_params": { "type": "array", "items": {"type": ["string", "number", "boolean", "null"]}, "description": "Parameters for the WHERE clause to prevent SQL injection", "default": [], }, }, "required": ["table", "where"], }, }, } def get_create_table_tool() -> Dict[str, Any]: """Get create_table tool definition.""" return { "type": "function", "function": { "name": "create_table", "description": "Create a new table in the database with specified columns and constraints.", "parameters": { "type": "object", "properties": { "table": { "type": "string", "description": "Name of the new table (e.g., 'users')", }, "schema": { "type": "string", "description": "Column definitions (e.g., 'id INTEGER PRIMARY KEY, name TEXT NOT NULL, email TEXT UNIQUE')", }, }, "required": ["table", "schema"], }, }, } def get_list_tables_tool() -> Dict[str, Any]: """Get list_tables tool definition.""" return { "type": "function", "function": { "name": "list_tables", "description": "Get a list of all tables in the currently open database.", "parameters": { "type": "object", "properties": {}, "required": [], }, }, } def get_get_table_schema_tool() -> Dict[str, Any]: """Get get_table_schema tool definition.""" return { "type": "function", "function": { "name": "get_table_schema", "description": "Get detailed schema information for a specific table, including column names, types, and constraints.", "parameters": { "type": "object", "properties": { "table": { "type": "string", "description": "Name of the table to get schema for (e.g., 'users')", } }, "required": ["table"], }, }, } # Tool mapping for easy lookup TOOL_MAP = { "open_database": get_open_database_tool, "close_database": get_close_database_tool, "execute_query": get_execute_query_tool, "insert": get_insert_tool, "update": get_update_tool, "delete": get_delete_tool, "create_table": get_create_table_tool, "list_tables": get_list_tables_tool, "get_table_schema": get_get_table_schema_tool, } def get_tool_by_name(name: str) -> Dict[str, Any]: """Get a tool definition by name. Args: name: Tool name Returns: Tool definition Raises: ValueError: If tool not found """ if name not in TOOL_MAP: raise ValueError(f"Unknown tool: {name}") return TOOL_MAP[name]() # Usage examples FUNCTION_CALLING_EXAMPLES = { "open_database": { "description": "Opening a database", "example": { "name": "open_database", "arguments": '{"path": "./data/myapp.db"}' } }, "execute_query": { "description": "Querying users over 18 years old", "example": { "name": "execute_query", "arguments": '{"query": "SELECT * FROM users WHERE age > ?", "parameters": [18]}' } }, "insert": { "description": "Adding a new user", "example": { "name": "insert", "arguments": '{"table": "users", "data": {"name": "John Doe", "email": "john@example.com", "age": 30}}' } }, "update": { "description": "Updating a user\'s information", "example": { "name": "update", "arguments": '{"table": "users", "data": {"age": 31}, "where": "id = ?", "where_params": [1]}' } }, "delete": { "description": "Deleting a user", "example": { "name": "delete", "arguments": '{"table": "users", "where": "id = ?", "where_params": [1]}' } }, "list_tables": { "description": "Listing all tables", "example": { "name": "list_tables", "arguments": '{}' } } } if __name__ == "__main__": import json print("OpenAI-Compatible Tools for SQLite MCP Server") print("=" * 50) tools = get_openai_tools() print(f"\nTotal tools: {len(tools)}\n") for tool in tools: func = tool["function"] print(f"Tool: {func['name']}") print(f"Description: {func['description']}") print(f"Parameters: {list(func['parameters']['properties'].keys())}") print()