MCP Variance Log

Instructions

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"],
      },
  ),

• src/mcp_variance_log/db_utils.py:29-52 (helper)

  _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