create_table
Create new SQLite database tables to log statistical variations in conversation structure for anomaly detection.
Instructions
Create a new table in the SQLite database
Input Schema
TableJSON Schema
| Name | Required | Description | Default |
|---|---|---|---|
| query | Yes | CREATE TABLE SQL statement |
Implementation Reference
- src/mcp_variance_log/server.py:377-381 (handler)Handler for the 'create_table' tool: validates that the query starts with 'CREATE TABLE' and executes it using the database helper.
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")] - src/mcp_variance_log/server.py:198-208 (registration)Registration of the 'create_table' tool in the list_tools handler, including input schema.
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"], }, ), - _execute_query method in LogDatabase class: generic SQL executor that handles CREATE statements by committing and returning affected rows count. Used by the create_table handler.
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}") try: with closing(sqlite3.connect(self.db_path)) as conn: conn.row_factory = sqlite3.Row with closing(conn.cursor()) as cursor: if params: cursor.execute(query, params) else: cursor.execute(query) if query.strip().upper().startswith(('INSERT', 'UPDATE', 'DELETE', 'CREATE', 'DROP', 'ALTER')): conn.commit() affected = cursor.rowcount logger.debug(f"Write query affected {affected} rows") return [{"affected_rows": affected}] results = [dict(row) for row in cursor.fetchall()] logger.debug(f"Read query returned {len(results)} rows") return results except Exception as e: logger.error(f"Database error executing query: {e}") raise